1、协议举例

1.1、redis协议举例

  1. NioEventLoopGroup worker = new NioEventLoopGroup();
  2. byte[] LINE = {13, 10};
  3. try {
  4.     Bootstrap bootstrap = new Bootstrap();
  5.     bootstrap.channel(NioSocketChannel.class);
  6.     bootstrap.group(worker);
  7.     bootstrap.handler(new ChannelInitializer<SocketChannel>() {
  8.         @Override
  9.         protected void initChannel(SocketChannel ch) {
  10.             ch.pipeline().addLast(new LoggingHandler());
  11.             ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  12.                 // 会在连接 channel 建立成功后，会触发 active 事件
  13.                 @Override
  14.                 public void channelActive(ChannelHandlerContext ctx) {
  15.                     set(ctx);
  16.                     get(ctx);
  17.                 }
  18.                 private void get(ChannelHandlerContext ctx) {
  19.                     ByteBuf buf = ctx.alloc().buffer();
  20.                     buf.writeBytes("*2".getBytes());
  21.                     buf.writeBytes(LINE);
  22.                     buf.writeBytes("$3".getBytes());
  23.                     buf.writeBytes(LINE);
  24.                     buf.writeBytes("get".getBytes());
  25.                     buf.writeBytes(LINE);
  26.                     buf.writeBytes("$3".getBytes());
  27.                     buf.writeBytes(LINE);
  28.                     buf.writeBytes("aaa".getBytes());
  29.                     buf.writeBytes(LINE);
  30.                     ctx.writeAndFlush(buf);
  31.                 }
  32.                 private void set(ChannelHandlerContext ctx) {
  33.                     ByteBuf buf = ctx.alloc().buffer();
  34.                     buf.writeBytes("*3".getBytes());
  35.                     buf.writeBytes(LINE);
  36.                     buf.writeBytes("$3".getBytes());
  37.                     buf.writeBytes(LINE);
  38.                     buf.writeBytes("set".getBytes());
  39.                     buf.writeBytes(LINE);
  40.                     buf.writeBytes("$3".getBytes());
  41.                     buf.writeBytes(LINE);
  42.                     buf.writeBytes("aaa".getBytes());
  43.                     buf.writeBytes(LINE);
  44.                     buf.writeBytes("$3".getBytes());
  45.                     buf.writeBytes(LINE);
  46.                     buf.writeBytes("bbb".getBytes());
  47.                     buf.writeBytes(LINE);
  48.                     ctx.writeAndFlush(buf);
  49.                 }
  51.                 @Override
  52.                 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  53.                     ByteBuf buf = (ByteBuf) msg;
  54.                     System.out.println(buf.toString(Charset.defaultCharset()));
  55.                 }
  56.             });
  57.         }
  58.     });
  59.     ChannelFuture channelFuture = bootstrap.connect("localhost", 6379).sync();
  60.     channelFuture.channel().closeFuture().sync();
  61. } catch (InterruptedException e) {
  62.     log.error("client error", e);
  63. } finally {
  64.     worker.shutdownGracefully();
  65. }

1.2、http协议举例

  1. @Slf4j
  2. public class TestHttp {
  4.     public static void main(String[] args) {
  5.         NioEventLoopGroup boss = new NioEventLoopGroup();
  6.         NioEventLoopGroup worker = new NioEventLoopGroup();
  7.         try {
  8.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  9.             serverBootstrap.channel(NioServerSocketChannel.class);
  10.             serverBootstrap.group(boss, worker);
  11.             serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
  12.                 @Override
  13.                 protected void initChannel(SocketChannel ch) throws Exception {
  14.                     ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
  15.                     ch.pipeline().addLast(new HttpServerCodec());
  16.                     ch.pipeline().addLast(new SimpleChannelInboundHandler<HttpRequest>() {
  17.                         @Override
  18.                         protected void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) throws Exception {
  19.                             // 获取请求
  20.                             log.debug(msg.uri());
  22.                             // 返回响应
  23.                             DefaultFullHttpResponse response =
  24.                                     new DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK);
  26.                             byte[] bytes = "<h1>Hello, world!</h1>".getBytes();
  28.                             response.headers().setInt(CONTENT_LENGTH, bytes.length);
  29.                             response.content().writeBytes(bytes);
  31.                             // 写回响应
  32.                             ctx.writeAndFlush(response);
  33.                         }
  34.                     });
  36.                      /*ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  37.                 @Override
  38.                 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  39.                     log.debug("{}", msg.getClass());
  41.                     if (msg instanceof HttpRequest) { // 请求行，请求头
  43.                     } else if (msg instanceof HttpContent) { //请求体
  45.                     }
  46.                 }
  47.             });*/
  48.                 }
  49.             });
  50.             ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();
  51.             channelFuture.channel().closeFuture().sync();
  52.         } catch (InterruptedException e) {
  53.             log.error("server error", e);
  54.         } finally {
  55.             boss.shutdownGracefully();
  56.             worker.shutdownGracefully();
  57.         }
  58.     }
  59. }

2、自定义协议

2.1、自定义协议要素

  • 魔数，用来在第一时间判定是否是无效数据包
  • 版本号，可以支持协议的升级
  • 序列化算法，消息正文到底采用哪种序列化反序列化方式，可以由此扩展，例如：json、protobuf、hessian、jdk
  • 指令类型，是登录、注册、单聊、群聊… 跟业务相关
  • 请求序号，为了双工通信，提供异步能力
  • 正文长度
  • 消息正文

2.2、示例

  1. @Slf4j
  2. public class MessageCodec extends ByteToMessageCodec<Message> {
  6.     @Override
  7.     protected void encode(ChannelHandlerContext channelHandlerContext, Message message, ByteBuf byteBuf) throws Exception {
  8.         //1、魔数 4字节
  9.         byteBuf.writeInt(Integer.MAX_VALUE);
  10.         //2、版本号 1字节
  11.         byteBuf.writeByte(1);
  12.         //3、序列化算法 1 json 1字节
  13.         byteBuf.writeByte(1);
  14.         //无效填充 1字节
  15.         byteBuf.writeByte(128);
  16.         //4、消息类型 1字节
  17.         byteBuf.writeByte(message.getMessageType());
  18.         //5、请求序号 4字节
  19.         byteBuf.writeInt(message.getSeqNo());
  20.         //6、数据长度 4字节
  21.         final byte[] bytes = JSON.toJSONString(message).getBytes(StandardCharsets.UTF_8);
  22.         byteBuf.writeInt(bytes.length);
  23.         //7、数据
  24.         byteBuf.writeBytes(bytes);
  26.     }
  28.     @Override
  29.     protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
  30.         //1、魔数 4字节
  31.         final int magicNumber = byteBuf.readInt();
  32.         //2、版本号 1字节
  33.         final byte version = byteBuf.readByte();
  34.         //3、序列化算法 1 json 1字节
  35.         final byte serializerType = byteBuf.readByte();
  36.         //无效填充 1字节
  37.         byteBuf.readByte();
  38.         //4、消息类型 1字节
  39.         final byte messageType = byteBuf.readByte();
  40.         //5、请求序号 4字节
  41.         final int seqNo = byteBuf.readInt();
  42.         //6、数据长度 4字节
  43.         final int length = byteBuf.readInt();
  44.         //7、数据
  45.         byte[] bytes = new byte[length];
  46.         byteBuf.readBytes(bytes,0,length);
  47.         final Message message = JSON.parseObject(bytes.toString(), Message.class);
  48.         log.info("magicNumber: {} , version: {} , serializerType: {} , messageType: {} , seqNo: {} , length: {} , data: {}"
  49.                 ,magicNumber,version,serializerType,messageType,seqNo,length,message);
  50.         list.add(message);
  51.     }
  53.     public static void main(String[] args) {
  54.         final EmbeddedChannel embeddedChannel = new EmbeddedChannel(new LoggingHandler(LogLevel.DEBUG),
  55.                 new LengthFieldBasedFrameDecoder(1024, 12, 4, 0, 0),
  56.                 new MessageCodec());
  58.         LoginRequestMessage loginRequestMessage = new LoginRequestMessage("zs","123456");
  59.         embeddedChannel.writeOutbound(loginRequestMessage);
  60.     }
  61. }
  1. @Data
  2. public abstract class Message implements Serializable {
  3.     private String seqNo;
  5.     public abstract Integer getMessageType();
  7.     public abstract Integer getSeqNo();
  8. }
  1. @Data
  2. public class LoginRequestMessage extends Message{
  4.     private String userName;
  6.     private String password;
  8.     public LoginRequestMessage(String userName, String password) {
  9.         this.userName = userName;
  10.         this.password = password;
  11.     }
  13.     @Override
  14.     public Integer getMessageType() {
  15.         return 1;
  16.     }
  18.     @Override
  19.     public Integer getSeqNo() {
  20.         return 1;
  21.     }
  22. }