文章详情

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

请输入下面的图形验证码

提交验证

短信预约提醒成功

Netty、MINA、Twisted中如何定制自己的协议

2023-06-03 20:26

关注

这篇文章给大家分享的是有关Netty、MINA、Twisted中如何定制自己的协议的内容。小编觉得挺实用的,因此分享给大家做个参考,一起跟随小编过来看看吧。

之前介绍了一些消息分割的方案,以及 MINA、Netty、Twisted 针对这些方案提供的相关API。例如MINA的TextLineCodecFactory、PrefixedStringCodecFactory,Netty的LineBasedFrameDecoder、LengthFieldBasedFrameDecoder,Twisted的LineOnlyReceiver、Int32StringReceiver。

除了这些方案,还有很多其他方案,当然也可以自己定义。在这里,我们定制一个自己的方案,并分别使用MINA、Netty、Twisted实现对这种消息的解析和组装,也就是编码和解码。

之前介绍了一种用固定字节数的Header来指定Body字节数的消息分割方案,其中Header部分是常规的大字节序(Big-Endian)的4字节整数。本文中对这个方案稍作修改,将固定字节数的Header改为小字节序(Little-Endian)的4字节整数。

常规的大字节序表示一个数的话,用高字节位的存放数字的低位,比较符合人的习惯。而小字节序和大字节序正好相反,用高字节位存放数字的高位。

Netty、MINA、Twisted中如何定制自己的协议

Python中struct模块支持大小字节序的pack和unpack,在Java中可以用下面的两个方法实现小字节序字节数组转int和int转小字节序字节数组,下面的Java程序中将会用到这两个方法:

public class LittleEndian {

 
 public static byte[] toLittleEndian(int i) {
   byte[] bytes = new byte[4];
   bytes[0] = (byte) i;
   bytes[1] = (byte) (i >>> 8);
   bytes[2] = (byte) (i >>> 16);
   bytes[3] = (byte) (i >>> 24);
   return bytes;
 }

 
 public static int getLittleEndianInt(byte[] bytes) {
   int b0 = bytes[0] & 0xFF;
       int b1 = bytes[1] & 0xFF;
       int b2 = bytes[2] & 0xFF;
       int b3 = bytes[3] & 0xFF;
       return b0 + (b1 << 8) + (b2 << 16) + (b3 << 24);
 }
}

无论是MINA、Netty还是Twisted,消息的编码、解码、切合的代码,都是应该和业务逻辑代码分开,这样有利于代码的开发、重用和维护。在MINA和Netty中类似,编码、解码需要继承实现相应的Encoder、Decoder,而在Twisted中则是继承Protocol实现编码解码。虽然实现方式不同,但是它们的功能都是一样的:

  • 对消息根据一定规则进行切合,例如固定长度消息、按行、按分隔符、固定长度Header指定Body长度等;

  • 将切合后的消息由字节码转成自己想要的类型,如MINA中将IoBuffer转成字符串,这样messageReceived接收到的message参数就是String类型;

  • write的时候可以传入自定义类型的参数,由编码器完成编码。

下面分别用MINA、Netty、Twisted实现4字节的小字节序int来指定body长度的消息的编码和解码。

MINA

在MINA中对接收到的消息进行切合和解码,一般会定义一个解码器类,继承自抽象类CumulativeProtocolDecoder,实现doDecode方法:

public class MyMinaDecoder extends CumulativeProtocolDecoder {

 @Override
 protected boolean doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception {

   // 如果没有接收完Header部分(4字节),直接返回false
   if(in.remaining() < 4) {
     return false;
   } else {

     // 标记开始位置,如果一条消息没传输完成则返回到这个位置
     in.mark();

     byte[] bytes = new byte[4];
     in.get(bytes); // 读取4字节的Header

     int bodyLength = LittleEndian.getLittleEndianInt(bytes); // 按小字节序转int

     // 如果body没有接收完整,直接返回false
     if(in.remaining() < bodyLength) {
       in.reset(); // IoBuffer position回到原来标记的地方
       return false;
     } else {
       byte[] bodyBytes = new byte[bodyLength];
       in.get(bodyBytes);
       String body = new String(bodyBytes, "UTF-8");
       out.write(body); // 解析出一条消息
       return true;
     }
   }
 }
}

另外,session.write的时候要对数据编码,需要定义一个编码器,继承自抽象类ProtocolEncoderAdapter,实现encode方法:

public class MyMinaEncoder extends ProtocolEncoderAdapter {

 @Override
 public void encode(IoSession session, Object message,
     ProtocolEncoderOutput out) throws Exception {

   String msg = (String) message;
   byte[] bytes = msg.getBytes("UTF-8");
   int length = bytes.length;
   byte[] header = LittleEndian.toLittleEndian(length); // 按小字节序转成字节数组

   IoBuffer buffer = IoBuffer.allocate(length + 4);
   buffer.put(header); // header
   buffer.put(bytes); // body
   buffer.flip();
   out.write(buffer);
 }
}

服务器启动的时候加入相应的编码器和解码器:

public class TcpServer {

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

   // 指定编码解码器
   acceptor.getFilterChain().addLast("codec",
       new ProtocolCodecFilter(new MyMinaEncoder(), new MyMinaDecoder()));

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

下面是业务逻辑的代码:

public 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 {

   // MyMinaDecoder将接收到的数据由IoBuffer转为String
   String msg = (String) message;
   System.out.println("messageReceived:" + msg);

   // MyMinaEncoder将write的字符串添加了一个小字节序Header并转为字节码
   session.write("收到");
 }
}

Netty

Netty中解码器和MINA类似,解码器继承抽象类ByteToMessageDecoder,实现decode方法:

public class MyNettyDecoder extends ByteToMessageDecoder {

 @Override
 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {

   // 如果没有接收完Header部分(4字节),直接退出该方法
   if(in.readableBytes() >= 4) {

     // 标记开始位置,如果一条消息没传输完成则返回到这个位置
     in.markReaderIndex();

     byte[] bytes = new byte[4];
     in.readBytes(bytes); // 读取4字节的Header

     int bodyLength = LittleEndian.getLittleEndianInt(bytes); // header按小字节序转int

     // 如果body没有接收完整
     if(in.readableBytes() < bodyLength) {
       in.resetReaderIndex(); // ByteBuf回到标记位置
     } else {
       byte[] bodyBytes = new byte[bodyLength];
       in.readBytes(bodyBytes);
       String body = new String(bodyBytes, "UTF-8");
       out.add(body); // 解析出一条消息
     }
   }
 }
}

下面是编码器,继承自抽象类MessageToByteEncoder,实现encode方法:

public class MyNettyEncoder extends MessageToByteEncoder<String> {

 @Override
 protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out)
     throws Exception {

   byte[] bytes = msg.getBytes("UTF-8");
   int length = bytes.length;
   byte[] header = LittleEndian.toLittleEndian(length); // int按小字节序转字节数组
   out.writeBytes(header); // write header
   out.writeBytes(bytes); // write 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();

             // 加上自己的Encoder和Decoder
             pipeline.addLast(new MyNettyDecoder());
             pipeline.addLast(new MyNettyEncoder());

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

业务逻辑处理类:

public class TcpServerHandler extends ChannelInboundHandlerAdapter {

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

       // MyNettyDecoder将接收到的数据由ByteBuf转为String
       String message = (String) msg;
       System.out.println("channelRead:" + message);

       // MyNettyEncoder将write的字符串添加了一个小字节序Header并转为字节码
       ctx.writeAndFlush("收到");
   }

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

Twisted

Twisted的实现方式和MINA、Netty不太一样,其实现方式相对来说更加原始,但是越原始也越接近底层原理。

首先要定义一个MyProtocol类继承自Protocol,用于充当类似于MINA、Netty的编码、解码器。处理业务逻辑的类TcpServerHandle继承MyProtocol,重写或调用MyProtocol提供的一些方法。

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

from struct import pack, unpack
from twisted.internet.protocol import Factory
from twisted.internet.protocol import Protocol
from twisted.internet import reactor

# 编码、解码器
class MyProtocol(Protocol):

   # 用于暂时存放接收到的数据
   _buffer = b""

   def dataReceived(self, data):
       # 上次未处理的数据加上本次接收到的数据
       self._buffer = self._buffer + data
       # 一直循环直到新的消息没有接收完整
       while True:
           # 如果header接收完整
           if len(self._buffer) >= 4:
               # 按小字节序转int
               length, = unpack("<I", self._buffer[0:4])
               # 如果body接收完整
               if len(self._buffer) >= 4 + length:
                   # body部分
                   packet = self._buffer[4:4 + length]
                   # 新的一条消息接收并解码完成,调用stringReceived
                   self.stringReceived(packet)
                   # 去掉_buffer中已经处理的消息部分
                   self._buffer = self._buffer[4 + length:]
               else:
                   break;
           else:
               break;

   def stringReceived(self, data):
       raise NotImplementedError

   def sendString(self, string):
       self.transport.write(pack("<I", len(string)) + string)

# 逻辑代码
class TcpServerHandle(MyProtocol):

   # 实现MyProtocol提供的stringReceived而不是dataReceived,不然无法解码
   def stringReceived(self, data):

       # data为MyProtocol解码后的数据
       print 'stringReceived:' + data

       # 调用sendString而不是self.transport.write,不然不能进行编码
       self.sendString("收到")

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;
   OutputStream out = null;
   InputStream in = null;

   try {

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

     // 请求服务器
     String data = "我是客户端";
     byte[] outputBytes = data.getBytes("UTF-8");
     out.write(LittleEndian.toLittleEndian(outputBytes.length)); // write header
     out.write(outputBytes); // write body
     out.flush();

     // 获取响应
     byte[] inputBytes = new byte[1024];
     int length = in.read(inputBytes);
     if(length >= 4) {
       int bodyLength = LittleEndian.getLittleEndianInt(inputBytes);
       if(length >= 4 + bodyLength) {
         byte[] bodyBytes = Arrays.copyOfRange(inputBytes, 4, 4 + bodyLength);
         System.out.println("Header:" + bodyLength);
         System.out.println("Body:" + new String(bodyBytes, "UTf-8"));
       }
     }

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

用客户端分别测试上面三个TCP服务器:

MINA服务器输出结果:

messageReceived:我是客户端

Netty服务器输出结果:

channelRead:我是客户端

Twisted服务器输出结果:

stringReceived:我是客户端

客户端测试三个服务器的输出结果都是:

Header:6

Body:收到

由于一个汉字一般占3个字节,所以两个汉字对应的Header为6。

感谢各位的阅读!关于“Netty、MINA、Twisted中如何定制自己的协议”这篇文章就分享到这里了,希望以上内容可以对大家有一定的帮助,让大家可以学到更多知识,如果觉得文章不错,可以把它分享出去让更多的人看到吧!

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯