Netty中通过ServerBootstrap ChannelPipeline发送消息时出现UnsupportedOperationException
UnsupportedOperationException when sending messages through ServerBootstrap ChannelPipeline in Netty
我正在使用 Netty 5.0。
我有一个互补的客户端 bootstrap，我从 netty github 中获取了 SecureChatClient.java 示例。
Wenn 我将消息从客户端 bootstrap 发送到服务器，它工作得很好。当我尝试从服务器 bootstrap 向客户端发送消息时（在首先通过客户端成功启动 connection/channel 之后），我得到一个 java.lang.UnsupportedOperationException 没有任何进一步的信息。从服务器向客户端发送消息是通过上面的代码完成的。
服务器bootstrap是否仅用于接收？
服务器bootstrap难道不是要能够像上面那样将消息写回客户端吗？我的意思是，消息可以从套接字向上通过 ChannelHandler 进入 ChannelPipeline，但只有 ChannelHandler 应该将响应写回 ChannelPipeline 并从套接字输出。因此，在 ServerBootstrap 中，用户并不意味着能够从 Pipeline 外部向 ChannelPipeline 发送消息。 （希望有意义）
还是我只是遗漏了什么？
我的代码如下：
// Ports.
int serverPort = 8080;
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast("MyMessageHandler", new MyMessageHandler());
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true);
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(serverPort).sync();
Channel ch = f.channel();
System.out.println("Server: Running!");
// Read commands from the stdin.
ChannelFuture lastWriteFuture = null;
BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
while(true)
{
String line = in.readLine();
if (line == null) break;
ByteBuf getOut = buffer(64);
getOut.writeBytes(line.getBytes());
// Sends the received line to the server.
lastWriteFuture = ch.writeAndFlush(getOut);
lastWriteFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture cf) throws Exception {
if(cf.isSuccess()) {
System.out.println("CFListener: SUCCESS! YEAH! HELL! YEAH!");
} else {
System.out.println("CFListener: failure! FAILure! FAILURE!");
System.out.println(cf.cause());
}
}
});
}
// Wait until all messages are flushed before closing the channel.
if (lastWriteFuture != null) {
lastWriteFuture.sync();
}
// Wait until the server socket is closed.
// In this example, this does not happen, but you can do that to gracefully
// shut down your server.
f.channel().closeFuture().sync();
} catch (InterruptedException | UnsupportedOperationException e) {
e.printStackTrace();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
我开始使用以下示例：https://github.com/netty/netty/tree/4.1/example/src/main/java/io/netty/example/securechat
我的问题是调用 ch.writeAndFlush 时出现以下异常：
java.lang.UnsupportedOperationException
at io.netty.channel.socket.nio.NioServerSocketChannel.filterOutboundMessage(NioServerSocketChannel.java:184)
at io.netty.channel.AbstractChannel$AbstractUnsafe.write(AbstractChannel.java:784)
at io.netty.channel.DefaultChannelPipeline$HeadContext.write(DefaultChannelPipeline.java:1278)
at io.netty.channel.ChannelHandlerInvokerUtil.invokeWriteNow(ChannelHandlerInvokerUtil.java:158)
at io.netty.channel.DefaultChannelHandlerInvoker$WriteTask.run(DefaultChannelHandlerInvoker.java:440)
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:328)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
at io.netty.util.concurrent.SingleThreadEventExecutor.run(SingleThreadEventExecutor.java:116)
at io.netty.util.internal.chmv8.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1412)
at io.netty.util.internal.chmv8.ForkJoinTask.doExec(ForkJoinTask.java:280)
at io.netty.util.internal.chmv8.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:877)
at io.netty.util.internal.chmv8.ForkJoinPool.scan(ForkJoinPool.java:1706)
at io.netty.util.internal.chmv8.ForkJoinPool.runWorker(ForkJoinPool.java:1661)
at io.netty.util.internal.chmv8.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:126)
我觉得Netty Server没有解码器，编码器。
如果你想发送字符串数据，
serverBootstrap.group(bossGroup, workerGroup).childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel channel) throws Exception {
ChannelPipeline channelPipeline = channel.pipeline();
channelPipeline.addLast("String Encoder", new StringEncoder(CharsetUtil.UTF_8));
channelPipeline.addLast("String Decoder", new StringDecoder(CharsetUtil.UTF_8));
}
});
添加服务器的初始化程序！
您不能写入ServerChannel，您只能连接到普通频道。由于这个原因，您对 writeAndFlush 的调用失败。
要向每个客户端发送消息，您应该将每个客户端的通道存储在 ChannelGroup 中并在其上调用 writeAndFlush()。
一个快速的方法是向您的 ServerBootstrap 添加另一个处理程序，将传入的连接放入 ChannelGroup 中，一个快速的实现是这样的：
// In your main:
ChannelGroup allChannels =
new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
// In your ChannelInitializer<SocketChannel>
ch.pipeline().addLast("grouper", new GlobalSendHandler());
// New class:
public class MyHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(ChannelHandlerContext ctx) {
allChannels.add(ctx.channel());
super.channelActive(ctx);
}
}
然后我们可以调用下面的方法向每个连接发送消息，这个returns一个ChannelGroupFuture而不是一个普通的ChannelFuture:
allChannels.writeAndFlush(getOut);
你的总代码看起来像上面的修复：
// Ports.
int serverPort = 8080;
ChannelGroup allChannels =
new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast("MyMessageHandler", new MyMessageHandler());
ch.pipeline().addLast("grouper", new GlobalSendHandler());
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true);
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(serverPort).sync();
Channel ch = f.channel();
System.out.println("Server: Running!");
// Read commands from the stdin.
ChannelGroupFuture lastWriteFuture = null;
BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
while(true)
{
String line = in.readLine();
if (line == null) break;
ByteBuf getOut = buffer(64);
getOut.writeBytes(line.getBytes());
// Sends the received line to the server.
lastWriteFuture = allChannels.writeAndFlush(getOut);
lastWriteFuture.addListener(new ChannelGroupFutureListener() {
@Override
public void operationComplete(ChannelGroupFuture cf) throws Exception {
if(cf.isSuccess()) {
System.out.println("CFListener: SUCCESS! YEAH! HELL! YEAH!");
} else {
System.out.println("CFListener: failure! FAILure! FAILURE!");
System.out.println(cf.cause());
}
}
});
}
// Wait until all messages are flushed before closing the channel.
if (lastWriteFuture != null) {
lastWriteFuture.sync();
}
// Wait until the server socket is closed.
// In this example, this does not happen, but you can do that to gracefully
// shut down your server.
f.channel().closeFuture().sync();
} catch (InterruptedException | UnsupportedOperationException e) {
e.printStackTrace();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
我正在使用 Netty 5.0。
我有一个互补的客户端 bootstrap，我从 netty github 中获取了 SecureChatClient.java 示例。
Wenn 我将消息从客户端 bootstrap 发送到服务器，它工作得很好。当我尝试从服务器 bootstrap 向客户端发送消息时（在首先通过客户端成功启动 connection/channel 之后），我得到一个 java.lang.UnsupportedOperationException 没有任何进一步的信息。从服务器向客户端发送消息是通过上面的代码完成的。
服务器bootstrap是否仅用于接收？
服务器bootstrap难道不是要能够像上面那样将消息写回客户端吗？我的意思是，消息可以从套接字向上通过 ChannelHandler 进入 ChannelPipeline，但只有 ChannelHandler 应该将响应写回 ChannelPipeline 并从套接字输出。因此，在 ServerBootstrap 中，用户并不意味着能够从 Pipeline 外部向 ChannelPipeline 发送消息。 （希望有意义）
还是我只是遗漏了什么？
我的代码如下：
// Ports.
int serverPort = 8080;
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast("MyMessageHandler", new MyMessageHandler());
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true);
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(serverPort).sync();
Channel ch = f.channel();
System.out.println("Server: Running!");
// Read commands from the stdin.
ChannelFuture lastWriteFuture = null;
BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
while(true)
{
String line = in.readLine();
if (line == null) break;
ByteBuf getOut = buffer(64);
getOut.writeBytes(line.getBytes());
// Sends the received line to the server.
lastWriteFuture = ch.writeAndFlush(getOut);
lastWriteFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture cf) throws Exception {
if(cf.isSuccess()) {
System.out.println("CFListener: SUCCESS! YEAH! HELL! YEAH!");
} else {
System.out.println("CFListener: failure! FAILure! FAILURE!");
System.out.println(cf.cause());
}
}
});
}
// Wait until all messages are flushed before closing the channel.
if (lastWriteFuture != null) {
lastWriteFuture.sync();
}
// Wait until the server socket is closed.
// In this example, this does not happen, but you can do that to gracefully
// shut down your server.
f.channel().closeFuture().sync();
} catch (InterruptedException | UnsupportedOperationException e) {
e.printStackTrace();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
我开始使用以下示例：https://github.com/netty/netty/tree/4.1/example/src/main/java/io/netty/example/securechat
我的问题是调用 ch.writeAndFlush 时出现以下异常：
java.lang.UnsupportedOperationException
at io.netty.channel.socket.nio.NioServerSocketChannel.filterOutboundMessage(NioServerSocketChannel.java:184)
at io.netty.channel.AbstractChannel$AbstractUnsafe.write(AbstractChannel.java:784)
at io.netty.channel.DefaultChannelPipeline$HeadContext.write(DefaultChannelPipeline.java:1278)
at io.netty.channel.ChannelHandlerInvokerUtil.invokeWriteNow(ChannelHandlerInvokerUtil.java:158)
at io.netty.channel.DefaultChannelHandlerInvoker$WriteTask.run(DefaultChannelHandlerInvoker.java:440)
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:328)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
at io.netty.util.concurrent.SingleThreadEventExecutor.run(SingleThreadEventExecutor.java:116)
at io.netty.util.internal.chmv8.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1412)
at io.netty.util.internal.chmv8.ForkJoinTask.doExec(ForkJoinTask.java:280)
at io.netty.util.internal.chmv8.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:877)
at io.netty.util.internal.chmv8.ForkJoinPool.scan(ForkJoinPool.java:1706)
at io.netty.util.internal.chmv8.ForkJoinPool.runWorker(ForkJoinPool.java:1661)
at io.netty.util.internal.chmv8.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:126)
我觉得Netty Server没有解码器，编码器。 如果你想发送字符串数据，
serverBootstrap.group(bossGroup, workerGroup).childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel channel) throws Exception {
ChannelPipeline channelPipeline = channel.pipeline();
channelPipeline.addLast("String Encoder", new StringEncoder(CharsetUtil.UTF_8));
channelPipeline.addLast("String Decoder", new StringDecoder(CharsetUtil.UTF_8));
}
});
添加服务器的初始化程序！
您不能写入ServerChannel，您只能连接到普通频道。由于这个原因，您对 writeAndFlush 的调用失败。
要向每个客户端发送消息，您应该将每个客户端的通道存储在 ChannelGroup 中并在其上调用 writeAndFlush()。
一个快速的方法是向您的 ServerBootstrap 添加另一个处理程序，将传入的连接放入 ChannelGroup 中，一个快速的实现是这样的：
// In your main:
ChannelGroup allChannels =
new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
// In your ChannelInitializer<SocketChannel>
ch.pipeline().addLast("grouper", new GlobalSendHandler());
// New class:
public class MyHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(ChannelHandlerContext ctx) {
allChannels.add(ctx.channel());
super.channelActive(ctx);
}
}
然后我们可以调用下面的方法向每个连接发送消息，这个returns一个ChannelGroupFuture而不是一个普通的ChannelFuture:
allChannels.writeAndFlush(getOut);
你的总代码看起来像上面的修复：
// Ports.
int serverPort = 8080;
ChannelGroup allChannels =
new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast("MyMessageHandler", new MyMessageHandler());
ch.pipeline().addLast("grouper", new GlobalSendHandler());
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true);
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(serverPort).sync();
Channel ch = f.channel();
System.out.println("Server: Running!");
// Read commands from the stdin.
ChannelGroupFuture lastWriteFuture = null;
BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
while(true)
{
String line = in.readLine();
if (line == null) break;
ByteBuf getOut = buffer(64);
getOut.writeBytes(line.getBytes());
// Sends the received line to the server.
lastWriteFuture = allChannels.writeAndFlush(getOut);
lastWriteFuture.addListener(new ChannelGroupFutureListener() {
@Override
public void operationComplete(ChannelGroupFuture cf) throws Exception {
if(cf.isSuccess()) {
System.out.println("CFListener: SUCCESS! YEAH! HELL! YEAH!");
} else {
System.out.println("CFListener: failure! FAILure! FAILURE!");
System.out.println(cf.cause());
}
}
});
}
// Wait until all messages are flushed before closing the channel.
if (lastWriteFuture != null) {
lastWriteFuture.sync();
}
// Wait until the server socket is closed.
// In this example, this does not happen, but you can do that to gracefully
// shut down your server.
f.channel().closeFuture().sync();
} catch (InterruptedException | UnsupportedOperationException e) {
e.printStackTrace();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}