扫码关注官方微信

扫码下载APP
返回顶部
首页 > 资讯 > 精选 >Netty、MINA、Twisted中如何定制自己的协议
  • 736

0
分享到

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

2023-06-03 20:06:46 736人浏览 独家记忆
摘要

这篇文章给大家分享的是有关Netty、MINA、Twisted中如何定制自己的协议的内容。小编觉得挺实用的，因此分享给大家做个参考，一起跟随小编过来看看吧。之前介绍了一些消息分割的方案，以及 MINA、Netty、Twisted 针对这些方

这篇文章给大家分享的是有关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中如何定制自己的协议”这篇文章就分享到这里了，希望以上内容可以对大家有一定的帮助，让大家可以学到更多知识，如果觉得文章不错，可以把它分享出去让更多的人看到吧！

--结束END--

本文标题: Netty、MINA、Twisted中如何定制自己的协议

本文链接: https://lsjlt.com/news/235665.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com    QQ/279061341

猜你喜欢
  • Netty、MINA、Twisted中如何定制自己的协议
    这篇文章给大家分享的是有关Netty、MINA、Twisted中如何定制自己的协议的内容。小编觉得挺实用的，因此分享给大家做个参考，一起跟随小编过来看看吧。之前介绍了一些消息分割的方案，以及 MINA、Netty、Twisted 针对这些方...
    99+
    2023-06-03
  • 在netty中如何使用native传输协议
    本篇内容主要讲解“在netty中如何使用native传输协议”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“在netty中如何使用native传输协议”吧!简介对于IO来说，除了传统的block ...
    99+
    2023-06-30
  • workerman 自定义的协议如何解决粘包拆包
    前言：         由于最近在使用 workerman 实现 Unity3D 联机游戏的服务端，虽然也可以通过 TCP 协议直接通信，但是在实际测试的过程中发现了一些小问题。         比如双方的数据包都是字符串的方式吗，还有就因...
    99+
    2023-09-07
    TCP粘包拆包 workerman unity3d PHP网游服务
  • Python中的迭代器协议是如何定义的？
    Python中的迭代器协议是如何定义的？在Python中，迭代器（Iterator）是一种实现了迭代器协议（Iterator Protocol）的对象。迭代器协议是一种定义了迭代器对象的行为的的标准规范。实现了迭代器协议的对象可以通过使用i...
    99+
    2023-10-24
    Python 迭代器协议 定义
  • linux中如何定义自己的命令
    这篇文章主要介绍linux中如何定义自己的命令，文中介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们一定要看完！学习如何创建别名：你可以将太长或难以记忆的命令打包成你自己构建的命令。命令别名Alias在 Linux shell 中指的是...
    99+
    2023-06-16
  • 如何理解Swift中的协议
    本篇内容主要讲解“如何理解Swift中的协议”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“如何理解Swift中的协议”吧!1. 前言协议定义了一个蓝图，规定了用...
    99+
    2024-04-02
  • 如何使用SUSE Studio定制自己的Linux发行版
    今天小编给大家分享一下如何使用SUSE Studio定制自己的Linux发行版的相关知识点，内容详细，逻辑清晰，相信大部分人都还太了解这方面的知识，所以分享这篇文章给大家参考一下，希望大家阅读完这篇文章后有所收获，下面我们一起来了解一下吧。...
    99+
    2023-06-27
  • 亲自教你在netty中使用TCP协议请求DNS服务器的详细过程
    目录简介DNS传输协议简介DNS的IP地址Do53/TCP在netty中的使用搭建DNS netty client发送DNS查询消息DNS查询的消息处理总结简介 DNS的全称doma...
    99+
    2024-04-02
  • vps如何绑定自己的域名
    要将自己的域名绑定到VPS，需要按照以下步骤操作：1. 在域名注册商处设置域名解析，将域名指向VPS的IP地址。登录到域名注册商的控...
    99+
    2023-09-15
    vps
  • 如何修复 Go 中协议缓冲区定义中重新定义的字段？
    问题内容 我正在使用提供程序的 .proto 文件构建 gRPC 客户端，并且有几个包含同名值的枚举。 syntax = "proto3"; enum Color { NONE ...
    99+
    2024-02-06
  • 如何在java中的使用UDP协议
    本篇文章为大家展示了如何在java中的使用UDP协议，内容简明扼要并且容易理解，绝对能使你眼前一亮，通过这篇文章的详细介绍希望你能有所收获。UDP是面向无连接的通讯协议，由于通讯不需要连接，所以可以实现广播发送。UDP通讯时不需要接收方确认...
    99+
    2023-06-06
  • Cassandra中的Gossip协议是如何工作的
    Cassandra的Gossip协议是一种用于节点之间通信和信息传播的分布式协议。它通过节点之间相互传播关于集群拓扑结构、状态和健康...
    99+
    2024-04-09
    Cassandra
  • 如何制定目标才能让自己的学习效果更好？
    每年年初的时候，相信大家都会给自己全新的一年立下各种 flag，比如提高自己的编程能力，学会写作，看 50 本书，每周健身一次等等。等到了年底，拿出年初立的清单才发现 flag 全倒了，是不是大部分人都这样？反正四哥是这样，嘿嘿。给你讲一个...
    99+
    2023-08-07
  • 如何制作自己的原生JavaScript路由
    目录前言简介JavaScript 的 History API实现自己的原生JS路由基于 History API 的 Vanilla JS 路由设置使“后退”和“前进”按钮起作用初始化...
    99+
    2024-04-02
  • 如何在Vue中调用JavaScript中的UDP协议
    Vue作为一款轻量级的JavaScript框架，可以在网页中构建灵活的用户界面，并且支持与其他JavaScript代码的交互以实现更丰富的功能。在一些特定的场景下，我们需要使用UDP协议来进行数据传输。本文将介绍如何在Vue中调用JavaS...
    99+
    2023-05-14
  • 如何理解HTTP协议中的状态码
    这篇文章主要介绍“如何理解HTTP协议中的状态码”，在日常操作中，相信很多人在如何理解HTTP协议中的状态码问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”如何理解HTTP协议中的状态码”的疑惑有所帮助！接下来...
    99+
    2023-06-08
  • spring拓展之如何定义自己的namespace
    目录spring拓展 定义自己的namespace1.查看源码认识spring是怎么加载xml配置的2.定义自己的namespacespring-namespace实现自定义标签类1...
    99+
    2024-04-02
  • 如何在Pycharm中制作自己的爬虫代码模板
    目录写作背景爬虫代码在 Pycharm 中设置代码模板写作背景 最近本菜鸡有几个网站想要爬，每个爬虫的代码不一样，但 有某种联系，可以抽出一部分通用的代码制成模板，减少代码工作量，于...
    99+
    2024-04-02
  • 如何理解WEB开发中的Python WSGI协议
    今天就跟大家聊聊有关如何理解WEB开发中的Python WSGI协议，可能很多人都不太了解，为了让大家更加了解，小编给大家总结了以下内容，希望大家根据这篇文章可以有所收获。Web应用程序开发Web应用程序的本质是什么简单描述Web应用程序的...
    99+
    2023-06-02
  • 如何使用JavaScript定义自己的ajax函数
    这篇文章将为大家详细讲解有关如何使用JavaScript定义自己的ajax函数，小编觉得挺实用的，因此分享给大家做个参考，希望大家阅读完这篇文章后可以有所收获。由于用原生js的方式发起的网络请求，都是以查询字符串的形式，提交给服务器的，用户...
    99+
    2023-06-21
软考高级职称资格查询
热门wiki
近期文章
推荐阅读
热门问答
热门标签
编程网，编程工程师的家园，是目前国内优秀的开源技术社区之一，形成了由开源软件库、代码分享、资讯、协作翻译、讨论区和博客等几大频道内容，为IT开发者提供了一个发现、使用、并交流开源技术的平台。

  • 官方手机版

  • 微信公众号

  • 商务合作