怎么用MINA、Netty、Twisted来实现消息分割

2023-06-03 21:17

短信预约 -IT技能 免费直播动态提醒

这篇文章主要介绍了怎么用MINA、Netty、Twisted来实现消息分割，具有一定借鉴价值，感兴趣的朋友可以参考下，希望大家阅读完这篇文章之后大有收获，下面让小编带着大家一起了解一下。

本文介绍一种消息分割方式，use a fixed length header that indicates the length of the body，用一个固定字节数的Header前缀来指定Body的字节数，以此来分割消息。

怎么用MINA、Netty、Twisted来实现消息分割固定字节数的Header前缀来指定Body的字节数

上面图中 Header 固定为 4 字节，Header 中保存的是一个 4 字节（32位）的整数，例如 12 即为 0x0000000C，这个整数用来指定 Body 的长度（字节数）。当读完这么多字节的 Body 之后，又是下一条消息的 Header。

下面分别用MINA、Netty、Twisted来实现对这种消息的切合和解码。

MINA

MINA 提供了 PrefixedStringCodecFactory 来对这种类型的消息进行编码解码，PrefixedStringCodecFactory 默认 Header 的大小是4字节，当然也可以指定成1或2。

public class TcpServer {

  public static void main(String[] args) throws IOException {
    IoAcceptor acceptor = new NioSocketAcceptor();

    // 4字节的Header指定Body的字节数，对这种消息的处理
    acceptor.getFilterChain().addLast("codec",
        new ProtocolCodecFilter(new PrefixedStringCodecFactory(Charset.forName("UTF-8"))));

    acceptor.setHandler(new TcpServerHandle());
    acceptor.bind(new InetSocketAddress(8080));
  }

}

class TcpServerHandle extends IoHandlerAdapter {

  @Override
  public void exceptionCaught(IoSession session, Throwable cause)
      throws Exception {
    cause.printStackTrace();
  }

  // 接收到新的数据
  @Override
  public void messageReceived(IoSession session, Object message)
      throws Exception {

    String msg = (String) message;
    System.out.println("messageReceived:" + msg);

  }

  @Override
  public void sessionCreated(IoSession session) throws Exception {
    System.out.println("sessionCreated");
  }

  @Override
  public void sessionClosed(IoSession session) throws Exception {
    System.out.println("sessionClosed");
  }
}

Netty

Netty 使用 LengthFieldBasedFrameDecoder 来处理这种消息。下面代码中的new LengthFieldBasedFrameDecoder(80, 0, 4, 0, 4)中包含5个参数，分别是int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip。maxFrameLength为消息的最大长度，lengthFieldOffset为Header的位置，lengthFieldLength为Header的长度，lengthAdjustment为长度调整（默认Header中的值表示Body的长度，并不包含Header自己），initialBytesToStrip为去掉字节数（默认解码后返回Header+Body的全部内容，这里设为4表示去掉4字节的Header，只留下Body）。

public class TcpServer {

  public static void main(String[] args) throws InterruptedException {
    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 {
              ChannelPipeline pipeline = ch.pipeline();

              // LengthFieldBasedFrameDecoder按行分割消息，取出body
              pipeline.addLast(new LengthFieldBasedFrameDecoder(80, 0, 4, 0, 4));
              // 再按UTF-8编码转成字符串
              pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));

              pipeline.addLast(new TcpServerHandler());
            }
          });
      ChannelFuture f = b.bind(8080).sync();
      f.channel().closeFuture().sync();
    } finally {
      workerGroup.shutdownGracefully();
      bossGroup.shutdownGracefully();
    }
  }

}

class TcpServerHandler extends ChannelInboundHandlerAdapter {

  // 接收到新的数据
  @Override
  public void channelRead(ChannelHandlerContext ctx, Object msg) {

    String message = (String) msg;
    System.out.println("channelRead:" + message);
  }

  @Override
  public void channelActive(ChannelHandlerContext ctx) {
    System.out.println("channelActive");
  }

  @Override
  public void channelInactive(ChannelHandlerContext ctx) {
    System.out.println("channelInactive");
  }

  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    cause.printStackTrace();
    ctx.close();
  }
}

Twisted

在Twisted中需要继承Int32StringReceiver，不再继承Protocol。Int32StringReceiver表示固定32位（4字节）的Header，另外还有Int16StringReceiver、Int8StringReceiver等。而需要实现的接受数据事件的方法不再是dataReceived，也不是lineReceived，而是stringReceived。

# -*- coding:utf-8 –*-

from twisted.protocols.basic import Int32StringReceiver
from twisted.internet.protocol import Factory
from twisted.internet import reactor

class TcpServerHandle(Int32StringReceiver):

    # 新的连接建立
    def connectionMade(self):
        print 'connectionMade'

    # 连接断开
    def connectionLost(self, reason):
        print 'connectionLost'

    # 接收到新的数据
    def stringReceived(self, data):
        print 'stringReceived:' + data

factory = Factory()
factory.protocol = TcpServerHandle
reactor.listenTCP(8080, factory)
reactor.run()

下面是Java编写的一个客户端测试程序：

public class TcpClient {

  public static void main(String[] args) throws IOException {

    Socket socket = null;
    DataOutputStream out = null;

    try {

      socket = new Socket("localhost", 8080);
      out = new DataOutputStream(socket.getOutputStream());

      // 请求服务器
      String data1 = "牛顿";
      byte[] outputBytes1 = data1.getBytes("UTF-8");
      out.writeInt(outputBytes1.length); // write header
      out.write(outputBytes1); // write body

      String data2 = "爱因斯坦";
      byte[] outputBytes2 = data2.getBytes("UTF-8");
      out.writeInt(outputBytes2.length); // write header
      out.write(outputBytes2); // write body

      out.flush();

    } finally {
      // 关闭连接
      out.close();
      socket.close();
    }

  }

}