Netty、MINA、Twisted一起学系列05：整合protobuf

发布时间：2020-08-14 18:14:56 阅读：195 作者：程序员私房菜栏目：互联网科技

开发者测试专用服务器限时活动，0元免费领，库存有限，领完即止！点击查看>>

文章已获得作者授权，原文地址：xxgblog.com/2014/08/27/mina-netty-twisted-5

protobuf 是谷歌的 Protocol Buffers 的简称，用于结构化数据和字节码之间互相转换（序列化、反序列化），一般应用于网络传输，可支持多种编程语言。

protobuf 如何使用这里不再介绍，本文主要介绍在 MINA、Netty、Twisted 中如何使用 protobuf。

在前一篇文章（Netty、MINA、Twisted一起学系列04：定制自己的协议）中，有介绍到一种用一个固定为4字节的前缀Header来指定Body的字节数的一种消息分割方式，在这里同样要使用到。只是其中 Body 的内容不再是字符串，而是 protobuf 字节码。

Netty、MINA、Twisted一起学系列05：整合protobuf

在处理业务逻辑时，肯定不希望还要对数据进行序列化和反序列化，而是希望直接操作一个对象，那么就需要有相应的编码器和解码器，将序列化和反序列化的逻辑写在编码器和解码器中。有关编码器和解码器的实现，上一篇文章中有介绍。

Netty 包中已经自带针对 protobuf 的编码器和解码器，那么就不用再自己去实现了。而 MINA、Twisted 还需要自己去实现 protobuf 的编码器和解码器。

这里定义一个protobuf数据结构，用于描述一个学生的信息，保存为StudentMsg.proto文件：

message Student {    // ID    required int32 id = 1;      // 姓名    required string name = 2;    // email    optional string email = 3;    // 朋友    repeated string friends = 4;}

用 StudentMsg.proto 分别生成 Java 和 Python 代码，将代码加入到相应的项目中。生成的代码就不再贴上来了。下面分别介绍在 Netty、MINA、Twisted 如何使用 protobuf 来传输 Student 信息。

1 Netty

Netty 自带 protobuf 的编码器和解码器，分别是 ProtobufEncoder 和 ProtobufDecoder。需要注意的是，ProtobufEncoder 和 ProtobufDecoder 只负责 protobuf 的序列化和反序列化，而处理消息 Header 前缀和消息分割的还需要 LengthFieldBasedFrameDecoder 和 LengthFieldPrepender。LengthFieldBasedFrameDecoder 即用于解析消息 Header 前缀，根据 Header 中指定的 Body 字节数截取 Body，LengthFieldPrepender 用于在wirte消息时在消息前面添加一个 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();              // 负责通过4字节Header指定的Body长度将消息切割              pipeline.addLast("frameDecoder",                  new LengthFieldBasedFrameDecoder(1048576, 0, 4, 0, 4));              // 负责将frameDecoder处理后的完整的一条消息的protobuf字节码转成Student对象              pipeline.addLast("protobufDecoder",                  new ProtobufDecoder(StudentMsg.Student.getDefaultInstance()));              // 负责将写入的字节码加上4字节Header前缀来指定Body长度              pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));              // 负责将Student对象转成protobuf字节码              pipeline.addLast("protobufEncoder", new ProtobufEncoder());                            pipeline.addLast(new TcpServerHandler());                        }                    });            ChannelFuture f = b.bind(8080).sync();            f.channel().closeFuture().sync();        } finally {            workerGroup.shutdownGracefully();            bossGroup.shutdownGracefully();        }    }}

处理事件时，接收和发送的参数直接就是Student对象。

public class TcpServerHandler extends ChannelInboundHandlerAdapter {    @Override    public void channelRead(ChannelHandlerContext ctx, Object msg) {        // 读取客户端传过来的Student对象      StudentMsg.Student student = (StudentMsg.Student) msg;        System.out.println("ID:" + student.getId());        System.out.println("Name:" + student.getName());        System.out.println("Email:" + student.getEmail());        System.out.println("Friends:");        List<String> friends = student.getFriendsList();        for(String friend : friends) {          System.out.println(friend);        }        // 新建一个Student对象传到客户端        StudentMsg.Student.Builder builder = StudentMsg.Student.newBuilder();        builder.setId(9);        builder.setName("服务器");        builder.setEmail("123@abc.com");        builder.addFriends("X");        builder.addFriends("Y");        StudentMsg.Student student2 = builder.build();        ctx.writeAndFlush(student2);    }    @Override    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {        cause.printStackTrace();        ctx.close();    }}

2. MINA

在 MINA 中没有针对 protobuf 的编码器和解码器，但是可以自己实现一个功能和 Netty 一样的编码器和解码器。

编码器：

public class MinaProtobufEncoder extends ProtocolEncoderAdapter {    @Override    public void encode(IoSession session, Object message,            ProtocolEncoderOutput out) throws Exception {      StudentMsg.Student student = (StudentMsg.Student) message;        byte[] bytes = student.toByteArray(); // Student对象转为protobuf字节码        int length = bytes.length;        IoBuffer buffer = IoBuffer.allocate(length + 4);        buffer.putInt(length); // write header        buffer.put(bytes); // write body        buffer.flip();        out.write(buffer);    }}

解码器：

public class MinaProtobufDecoder 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();      // 读取header部分，获取body长度      int bodyLength = in.getInt();      // 如果body没有接收完整，直接返回false      if (in.remaining() < bodyLength) {        in.reset(); // IoBuffer position回到原来标记的地方        return false;      } else {        byte[] bodyBytes = new byte[bodyLength];        in.get(bodyBytes); // 读取body部分        StudentMsg.Student student = StudentMsg.Student.parseFrom(bodyBytes); // 将body中protobuf字节码转成Student对象        out.write(student); // 解析出一条消息        return true;      }    }  }}

MINA 服务器加入 protobuf 的编码器和解码器：

public class TcpServer {    public static void main(String[] args) throws IOException {        IoAcceptor acceptor = new NioSocketAcceptor();        // 指定protobuf的编码器和解码器        acceptor.getFilterChain().addLast("codec",                new ProtocolCodecFilter(new MinaProtobufEncoder(), new MinaProtobufDecoder()));        acceptor.setHandler(new TcpServerHandle());        acceptor.bind(new InetSocketAddress(8080));    }}

这样，在处理业务逻辑时，就和Netty一样了：

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 {      // 读取客户端传过来的Student对象      StudentMsg.Student student = (StudentMsg.Student) message;        System.out.println("ID:" + student.getId());        System.out.println("Name:" + student.getName());        System.out.println("Email:" + student.getEmail());        System.out.println("Friends:");        List<String> friends = student.getFriendsList();        for(String friend : friends) {          System.out.println(friend);        }        // 新建一个Student对象传到客户端        StudentMsg.Student.Builder builder = StudentMsg.Student.newBuilder();        builder.setId(9);        builder.setName("服务器");        builder.setEmail("123@abc.com");        builder.addFriends("X");        builder.addFriends("Y");        StudentMsg.Student student2 = builder.build();        session.write(student2);    }}

3. Twisted

在 Twisted 中，首先定义一个 ProtobufProtocol 类，继承 Protocol 类，充当编码器和解码器。处理业务逻辑的 TcpServerHandle 类再继承 ProtobufProtocol 类，调用或重写 ProtobufProtocol 提供的方法。

# -*- coding:utf-8 –*-from struct import pack, unpackfrom twisted.internet.protocol import Factoryfrom twisted.internet.protocol import Protocolfrom twisted.internet import reactorimport StudentMsg_pb2# protobuf编码、解码器class ProtobufProtocol(Protocol):    # 用于暂时存放接收到的数据    _buffer = b""    def dataReceived(self, data):        # 上次未处理的数据加上本次接收到的数据        self._buffer = self._buffer + data        # 一直循环直到新的消息没有接收完整        while True:            # 如果header接收完整            if len(self._buffer) >= 4:                # header部分，按大字节序转int，获取body长度                length, = unpack(">I", self._buffer[0:4])                # 如果body接收完整                if len(self._buffer) >= 4 + length:                    # body部分，protobuf字节码                    packet = self._buffer[4:4 + length]                    # protobuf字节码转成Student对象                    student = StudentMsg_pb2.Student()                    student.ParseFromString(packet)                    # 调用protobufReceived传入Student对象                    self.protobufReceived(student)                    # 去掉_buffer中已经处理的消息部分                    self._buffer = self._buffer[4 + length:]                else:                    break;            else:                break;    def protobufReceived(self, student):        raise NotImplementedError    def sendProtobuf(self, student):        # Student对象转为protobuf字节码        data = student.SerializeToString()        # 添加Header前缀指定protobuf字节码长度        self.transport.write(pack(">I", len(data)) + data)# 逻辑代码class TcpServerHandle(ProtobufProtocol):    # 实现ProtobufProtocol提供的protobufReceived    def protobufReceived(self, student):        # 将接收到的Student输出        print 'ID:' + str(student.id)        print 'Name:' + student.name        print 'Email:' + student.email        print 'Friends:'        for friend in student.friends:            print friend        # 创建一个Student并发送给客户端        student2 = StudentMsg_pb2.Student()        student2.id = 9        student2.name = '服务器'.decode('UTF-8') # 中文需要转成UTF-8字符串        student2.email = '123@abc.com'        student2.friends.append('X')        student2.friends.append('Y')        self.sendProtobuf(student2)factory = Factory()factory.protocol = TcpServerHandlereactor.listenTCP(8080, factory)reactor.run()

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

public class TcpClient {    public static void main(String[] args) throws IOException {        Socket socket = null;        DataOutputStream out = null;        DataInputStream in = null;        try {            socket = new Socket("localhost", 8080);            out = new DataOutputStream(socket.getOutputStream());            in = new DataInputStream(socket.getInputStream());            // 创建一个Student传给服务器            StudentMsg.Student.Builder builder = StudentMsg.Student.newBuilder();            builder.setId(1);            builder.setName("客户端");            builder.setEmail("xxg@163.com");            builder.addFriends("A");            builder.addFriends("B");            StudentMsg.Student student = builder.build();            byte[] outputBytes = student.toByteArray(); // Student转成字节码            out.writeInt(outputBytes.length); // write header            out.write(outputBytes); // write body            out.flush();            // 获取服务器传过来的Student            int bodyLength = in.readInt();  // read header            byte[] bodyBytes = new byte[bodyLength];            in.readFully(bodyBytes);  // read body            StudentMsg.Student student2 = StudentMsg.Student.parseFrom(bodyBytes); // body字节码解析成Student            System.out.println("Header:" + bodyLength);            System.out.println("Body:");            System.out.println("ID:" + student2.getId());            System.out.println("Name:" + student2.getName());            System.out.println("Email:" + student2.getEmail());            System.out.println("Friends:");            List<String> friends = student2.getFriendsList();            for(String friend : friends) {              System.out.println(friend);            }        } finally {            // 关闭连接            in.close();            out.close();            socket.close();        }    }}

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

服务器输出：

ID:1
Name:客户端
Email:xxg@163.com
Friends:
A
B

客户端输出：

Header:32
Body:
ID:9
Name:服务器
Email:123@abc.com
Friends:
X
Y

亿速云「云服务器」，即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘，价格低至29元/月。点击查看>>

向AI问一下细节

Netty、MINA、Twisted一起学系列05：整合protobuf

猜你喜欢

最新资讯

相关推荐

开发者交流群：

相关标签