【Netty4 简单项目实践】十三、WebSocket Over ProtocolBuf

【前言】

之前用TCP模式传输ProtocolBuf模式，后面上了一个websocket传输ProtocolBuf业务，目前基本已经稳定了，现在把编解码器部分记录下来。

【server部件的组装】

核心思想是ProtocolBuf是byte[]流，而websocket对象在Netty自带的编解码中是对象，其中的数据部分是pb序列化后的字节流（即，ws的内容是二进制字节数组）所以在编码器和解码器上需要做转换

protected void initChannel(Channel ch) throws Exception {
    ch.pipeline().addLast("readTimeout", new ReadTimeoutHandler(45)); // 长时间不写会断
    ch.pipeline().addLast("HttpServerCodec", new HttpServerCodec());
    ch.pipeline().addLast("ChunkedWriter", new ChunkedWriteHandler());
    ch.pipeline().addLast("HttpAggregator", new HttpObjectAggregator(65535));
    ch.pipeline().addLast("WsProtocolDecoder",
        new WebSocketServerProtocolHandler(WEBSOCKET_PATH, null, true));
    ch.pipeline().addLast("WsBinaryDecoder", new WebSocketFrameDecoder()); // ws解码成字节
    ch.pipeline().addLast("WsEncoder", new WebSocketFramePrepender()); // 字节编码成ws
    ch.pipeline().addLast("protoBufEncoder", new ProtobufEncoder()); // bp编码成字节
    ch.pipeline().addLast("protoBufDecoder", new ProtobufDecoder(Protocol.getDefaultInstance()));
    ch.pipeline().addLast(new ProtoBufHandler());
}

核心是

ch.pipeline().addLast("WsBinaryDecoder",new WebSocketFrameDecoder()); // ws解码成字节

ch.pipeline().addLast("WsEncoder",new WebSocketFramePrepender()); // 字节编码成ws

这两段代码，他上面的是Netty自带的websocket协议的编解码器，下面是Netty自带的pb协议的编解码器，作用就是转换编码

【解码部分】

package com.seeplant.netty;

import java.util.List;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageDecoder;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;

/**
 * 从ws中解出bytebuf，传给下一次层（pbdecoder），下一层用bytebuf组织成bp
 */
public class WebSocketFrameDecoder extends MessageToMessageDecoder<WebSocketFrame> {
    @Override
    protected void decode(ChannelHandlerContext ctx, WebSocketFrame msg, List<Object> out) throws Exception {
        ByteBuf buff = msg.content();
        byte[] messageBytes = new byte[buff.readableBytes()];
        buff.readBytes(messageBytes);
        // 直接内存小心
        ByteBuf bytebuf = PooledByteBufAllocator.DEFAULT.buffer(); // 直接内存
        bytebuf.writeBytes(messageBytes);
        out.add(bytebuf.retain());
    }
}

注意给下一个解码器传递的时候，ByteBuf对象需要retain，不然ByteBuf对象会被这个解码器销毁掉。

【编码部分】

package com.seeplant.netty;

import java.util.List;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageEncoder;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;

/**
 * 从字节打包成WebSocket
 */
public class WebSocketFramePrepender extends MessageToMessageEncoder<ByteBuf> {

    @Override
    protected void encode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
        WebSocketFrame webSocketFrame = new BinaryWebSocketFrame(msg);
        out.add(webSocketFrame);
    }
}

编码器非常简单，就是把pb字节流转出websocketframe，并且是二进制形式的wsframe