Netty 触发了一个 exceptionCaught() 事件，它到达了 TextWebsocketEncoder 管道的尾部

Question

我尝试进行简单的网络套接字解码然后编码，但是当它通过 TextWebsocketDecoder 处理程序时出现此异常：

io.netty.channel.DefaultChannelPipeline$TailContext exceptionCaught
WARNING: An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception.
io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1
    at io.netty.buffer.AbstractReferenceCountedByteBuf.release(AbstractReferenceCountedByteBuf.java:101)
    at io.netty.buffer.DefaultByteBufHolder.release(DefaultByteBufHolder.java:73)
    at io.netty.util.ReferenceCountUtil.release(ReferenceCountUtil.java:59)
    at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:112)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:318)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:304)
    at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:318)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:304)
    at io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler.channelRead(WebSocketServerProtocolHandler.java:147)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:318)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:304)
    at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:318)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:304)
    at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:276)
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:263)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:318)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:304)
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:846)
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
    at io.netty.util.concurrent.SingleThreadEventExecutor.run(SingleThreadEventExecutor.java:112)
    at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
    at java.lang.Thread.run(Thread.java:745)

我拥有的是简单的初始化器，它可以找到直到 TextWebsocketEncoder:

public class ServerInitializer extends ChannelInitializer<Channel> {
    private final ChannelGroup group;

    public GameServerInitializer(ChannelGroup group) {
        this.group = group;
    }

    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new HttpServerCodec());
        pipeline.addLast(new HttpObjectAggregator(64 * 1024));
        pipeline.addLast(new ChunkedWriteHandler());
        pipeline.addLast(new HttpRequestHandler("/ws"));
        pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
        pipeline.addLast(new TextWebSocketFrameHandler(group));
        pipeline.addLast("textWebsocketDecoder",new TextWebsocketDecoder());
        pipeline.addLast("textWebsocketEncoder",new TextWebsocketEncoder());
    }
}

TextWebSocketFrameHandler

public class TextWebSocketFrameHandler  extends SimpleChannelInboundHandler<TextWebSocketFrame>{
     private final ChannelGroup group;

        public TextWebSocketFrameHandler(ChannelGroup group) {
            this.group = group;
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {

                ctx.pipeline().remove(HttpRequestHandler.class);

                group.writeAndFlush(new TextWebSocketFrame("Client " + ctx.channel() + " joined"));

                group.add(ctx.channel());

            } else {
                super.userEventTriggered(ctx, evt);
            }
        }

        @Override
        public void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
             ctx.fireChannelRead(msg);
            //group.writeAndFlush(msg.retain());
        }
}

这是 TextWebsocketDecoder 和 TextWebsocketEncoder：

TextWebsocket解码器：

public class TextWebsocketDecoder extends MessageToMessageDecoder<TextWebSocketFrame>
{

    @Override
    protected void decode(ChannelHandlerContext ctx, TextWebSocketFrame frame, List<Object> out) throws Exception
    {
        String json = frame.text(); 
        JSONObject jsonObject = new JSONObject(json);
        int type = jsonObject.getInt("type");
        JSONArray msgJsonArray = jsonObject.getJSONArray("msg");
        String user = msgJsonArray.getString(0);
        String pass = msgJsonArray.getString(1);
        String connectionkey = msgJsonArray.getString(2);
        int timestamp = jsonObject.getInt("timestamp");

        JSONObject responseJson = new JSONObject();
        responseJson.put("type",Config.LOGIN_SUCCESS);
        responseJson.put("connectionkey",connectionkey);

        out.add(responseJson); // After This im getting the exception !!!
    }
}

TextWebsocketEncoder

import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageEncoder;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;

public class TextWebsocketEncoder extends MessageToMessageEncoder<JSONObject>
{

    @Override
    protected void encode(ChannelHandlerContext arg0, JSONObject arg1, List<Object> out) throws Exception {
        String json = arg1.toString();
        out.add(new TextWebSocketFrame(json));          
    } 

}

Answer 1

您使用 SimpleChannelInboundHandler auto-releases 根据文档捕获数据。

因此，当您调用 ctx.fireChannelRead(msg); 将 msg 传递给管道上的其他处理程序时，会出现问题，因为 msg 将被释放。

要解决此问题，您可以使用 ChannelInboundHandlerAdapter 或者您可以通过调用适当的构造函数来停止 SimpleChannelInboundHandler 的 auto-releasing 进程，或者您可以在触发之前调用 ReferenceCountUtil.retain(msg);管道上方。

在此处查看 SimpleChannelInboundHandler 的文档： http://netty.io/4.0/api/io/netty/channel/SimpleChannelInboundHandler.html

并在此处阅读有关引用计数对象的信息（netty 4 的新概念）： http://netty.io/wiki/reference-counted-objects.html

Answer 2

异常

在您的 TextWebSocketFrameHandler 中，您正在调用 ctx.fireChannelRead(msg);，这会将消息向上传递 1 个链，但是 MessageToMessageDecoder 不准备处理此问题。为了解释这个问题，我需要解释 MessageToMessageDecoder 是如何工作的。

MessageToMessageDecoder 通过捕获来自上游的每条消息并将它们传递给您的自定义代码来工作，您的自定义代码处理工作，而 mtmd 处理您传入的资源的关闭。

由于您将引用传递给另一方，实际上您多次关闭了 WebSocketFrame，从而导致错误。 MessageToMessageDecoder 甚至会在 javadoc.

中警告您

为了解决这个问题，我们按照手册中的说明，让我们的频道阅读以下内容：

@Override
public void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
     msg.retain(); // ferrybig: fixed bug 
     ctx.fireChannelRead(msg);
     //group.writeAndFlush(msg.retain());
}

发不回的问题

在您的评论中，您声明代码不会返回任何内容。这是预期的，因为您的管道仅使用数据并将其向上传递到链中。要解决此问题，需要对您的管道进行一些返工。

我们需要交换json-webframe解码器和编码器的顺序：
```
pipeline.addLast("textWebsocketDecoder",new TextWebsocketEncoder());
pipeline.addLast("textWebsocketEncoder",new TextWebsocketDecoder());
```
这是因为您的解码器正在生成将发送回 ↑ 处理程序链的输出，如果解码器高于该输出，编码器将看不到该输出。（你的解码器不应该按照netty的命名方式叫解码器）
我们需要更改您的解码器以将生成的数据实际发送回 ↑ 链而不是 ↓ 进入 non-existing void。

为了进行这些更改，我们将让 TextWebSocketDecoder 扩展 ChannelInboundHandlerAdapter 而不是 MessageToMessageDecoder<TextWebSocketFrame>，因为我们正在处理消息而不是将它们传递给其他处理程序。

我们正在将解码方法的签名更改为 channelRead(ChannelHandlerContext ctx, Object msg)，并添加一些样板代码：
```
public void channelRead(ChannelHandlerContext ctx, Object msg) /* throws Exception */
    TextWebSocketFrame frame = (TextWebSocketFrame) msg;
    try {
        /* Remaining code, follow the steps further of see end result */
    } finally {
        frame.release();
    }
}
```

我们调整代码以将结果向上传递而不是向下传递：

public void channelRead(ChannelHandlerContext ctx, Object msg) /* throws Exception */
    TextWebSocketFrame frame = (TextWebSocketFrame) msg;
    try {

        String json = frame.text(); 
        JSONObject jsonObject = new JSONObject(json);
        int type = jsonObject.getInt("type");
        JSONArray msgJsonArray = jsonObject.getJSONArray("msg");
        String user = msgJsonArray.getString(0);
        String pass = msgJsonArray.getString(1);
        String connectionkey = msgJsonArray.getString(2);
        int timestamp = jsonObject.getInt("timestamp");

        JSONObject responseJson = new JSONObject();
        responseJson.put("type",Config.LOGIN_SUCCESS);
        responseJson.put("connectionkey",connectionkey);

        ctx.writeAndFlush(responseJson)
    } finally {
        frame.release();
    }
}

请注意，您可能会想从异常中删除我们之前的代码，但是这样做会在运行 netty 的异步特性下触发未定义的行为。

Netty 触发了一个 exceptionCaught() 事件，它到达了 TextWebsocketEncoder 管道的尾部

Netty getting An exceptionCaught() event was fired, and it reached at the tail of the pipeline on TextWebsocketEncoder

java

exception

netty

异常

发不回的问题