Mina、Netty、Twisted一起学(五):整合protobuf

释放双眼,带上耳机,听听看~!

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

protobuf如何使用这里不再介绍,本文主要介绍在MINA、Netty、Twisted中如何使用protobuf,不了解protobuf的同学可以去参考我的另一篇博文。

在前面的一篇博文中,有介绍到一种用一个固定为4字节的前缀Header来指定Body的字节数的一种消息分割方式,在这里同样要使用到。只是其中Body的内容不再是字符串,而是protobuf字节码。

Mina、Netty、Twisted一起学(五):整合protobuf

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

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

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


1
2
3
4
5
6
7
8
9
10
11
12
13
14
1message Student {  
2    // ID  
3    required int32 id = 1;    
4  
5    // 姓名  
6    required string name = 2;  
7  
8    // email  
9    optional string email = 3;  
10  
11    // 朋友  
12    repeated string friends = 4;  
13}  
14

用StudentMsg.proto分别生成Java和Python代码,将代码加入到相应的项目中。生成的代码就不再贴上来了。

下面分别介绍在Netty、MINA、Twisted如何使用protobuf来传输Student信息。

**Netty:
**

Netty自带protobuf的编码器和解码器,分别是ProtobufEncoder和ProtobufDecoder。需要注意的是,ProtobufEncoder和ProtobufDecoder只负责protobuf的序列化和反序列化,而处理消息Header前缀和消息分割的还需要LengthFieldBasedFrameDecoder和LengthFieldPrepender。LengthFieldBasedFrameDecoder即用于解析消息Header前缀,根据Header中指定的Body字节数截取Body,LengthFieldPrepender用于在wirte消息时在消息前面添加一个Header前缀来指定Body字节数。


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
1public class TcpServer {  
2  
3    public static void main(String[] args) throws InterruptedException {  
4        EventLoopGroup bossGroup = new NioEventLoopGroup();  
5        EventLoopGroup workerGroup = new NioEventLoopGroup();  
6        try {  
7            ServerBootstrap b = new ServerBootstrap();  
8            b.group(bossGroup, workerGroup)  
9                    .channel(NioServerSocketChannel.class)  
10                    .childHandler(new ChannelInitializer<SocketChannel>() {  
11                        @Override  
12                        public void initChannel(SocketChannel ch)  
13                                throws Exception {  
14                            ChannelPipeline pipeline = ch.pipeline();  
15    
16                            // 负责通过4字节Header指定的Body长度将消息切割  
17                            pipeline.addLast("frameDecoder",  
18                                    new LengthFieldBasedFrameDecoder(1048576, 0, 4, 0, 4));  
19                              
20                            // 负责将frameDecoder处理后的完整的一条消息的protobuf字节码转成Student对象  
21                            pipeline.addLast("protobufDecoder",  
22                                    new ProtobufDecoder(StudentMsg.Student.getDefaultInstance()));  
23  
24                            // 负责将写入的字节码加上4字节Header前缀来指定Body长度  
25                            pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));  
26                              
27                            // 负责将Student对象转成protobuf字节码  
28                            pipeline.addLast("protobufEncoder", new ProtobufEncoder());  
29  
30                            pipeline.addLast(new TcpServerHandler());  
31                        }  
32                    });  
33            ChannelFuture f = b.bind(8080).sync();  
34            f.channel().closeFuture().sync();  
35        } finally {  
36            workerGroup.shutdownGracefully();  
37            bossGroup.shutdownGracefully();  
38        }  
39    }  
40}
41

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


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
1public class TcpServerHandler extends ChannelInboundHandlerAdapter {  
2  
3    @Override  
4    public void channelRead(ChannelHandlerContext ctx, Object msg) {  
5          
6        // 读取客户端传过来的Student对象  
7        StudentMsg.Student student = (StudentMsg.Student) msg;  
8        System.out.println("ID:" + student.getId());  
9        System.out.println("Name:" + student.getName());  
10        System.out.println("Email:" + student.getEmail());  
11        System.out.println("Friends:");  
12        List<String> friends = student.getFriendsList();  
13        for(String friend : friends) {  
14            System.out.println(friend);  
15        }  
16  
17        // 新建一个Student对象传到客户端  
18        StudentMsg.Student.Builder builder = StudentMsg.Student.newBuilder();  
19        builder.setId(9);  
20        builder.setName("服务器");  
21        builder.setEmail("123@abc.com");  
22        builder.addFriends("X");  
23        builder.addFriends("Y");  
24        StudentMsg.Student student2 = builder.build();  
25        ctx.writeAndFlush(student2);  
26    }  
27  
28    @Override  
29    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {  
30        cause.printStackTrace();  
31        ctx.close();  
32    }  
33}  
34

MINA:

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

编码器:


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
1public class MinaProtobufEncoder extends ProtocolEncoderAdapter {  
2  
3    @Override  
4    public void encode(IoSession session, Object message,  
5            ProtocolEncoderOutput out) throws Exception {  
6  
7        StudentMsg.Student student = (StudentMsg.Student) message;  
8        byte[] bytes = student.toByteArray(); // Student对象转为protobuf字节码  
9        int length = bytes.length;  
10          
11        IoBuffer buffer = IoBuffer.allocate(length + 4);  
12        buffer.putInt(length); // write header  
13        buffer.put(bytes); // write body  
14        buffer.flip();  
15        out.write(buffer);  
16    }  
17}  
18

解码器:


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
1public class MinaProtobufDecoder extends CumulativeProtocolDecoder {  
2  
3    @Override  
4    protected boolean doDecode(IoSession session, IoBuffer in,  
5            ProtocolDecoderOutput out) throws Exception {  
6  
7        // 如果没有接收完Header部分(4字节),直接返回false  
8        if (in.remaining() < 4) {  
9            return false;  
10        } else {  
11  
12            // 标记开始位置,如果一条消息没传输完成则返回到这个位置  
13            in.mark();  
14  
15            // 读取header部分,获取body长度  
16            int bodyLength = in.getInt();  
17  
18            // 如果body没有接收完整,直接返回false  
19            if (in.remaining() < bodyLength) {  
20                in.reset(); // IoBuffer position回到原来标记的地方  
21                return false;  
22            } else {  
23                byte[] bodyBytes = new byte[bodyLength];  
24                in.get(bodyBytes); // 读取body部分  
25                StudentMsg.Student student = StudentMsg.Student.parseFrom(bodyBytes); // 将body中protobuf字节码转成Student对象  
26                out.write(student); // 解析出一条消息  
27                return true;  
28            }  
29        }  
30    }  
31}  
32

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


1
2
3
4
5
6
7
8
9
10
11
12
13
14
1public class TcpServer {  
2  
3    public static void main(String[] args) throws IOException {  
4        IoAcceptor acceptor = new NioSocketAcceptor();  
5  
6        // 指定protobuf的编码器和解码器  
7        acceptor.getFilterChain().addLast("codec",  
8                new ProtocolCodecFilter(new MinaProtobufEncoder(), new MinaProtobufDecoder()));  
9  
10        acceptor.setHandler(new TcpServerHandle());  
11        acceptor.bind(new InetSocketAddress(8080));  
12    }  
13}  
14

这样,在处理业务逻辑时,就和Netty一样了:


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
1public class TcpServerHandle extends IoHandlerAdapter {  
2  
3    @Override  
4    public void exceptionCaught(IoSession session, Throwable cause)  
5            throws Exception {  
6        cause.printStackTrace();  
7    }  
8      
9    @Override  
10    public void messageReceived(IoSession session, Object message)  
11            throws Exception {  
12  
13        // 读取客户端传过来的Student对象  
14        StudentMsg.Student student = (StudentMsg.Student) message;  
15        System.out.println("ID:" + student.getId());  
16        System.out.println("Name:" + student.getName());  
17        System.out.println("Email:" + student.getEmail());  
18        System.out.println("Friends:");  
19        List<String> friends = student.getFriendsList();  
20        for(String friend : friends) {  
21            System.out.println(friend);  
22        }  
23  
24        // 新建一个Student对象传到客户端  
25        StudentMsg.Student.Builder builder = StudentMsg.Student.newBuilder();  
26        builder.setId(9);  
27        builder.setName("服务器");  
28        builder.setEmail("123@abc.com");  
29        builder.addFriends("X");  
30        builder.addFriends("Y");  
31        StudentMsg.Student student2 = builder.build();  
32        session.write(student2);  
33    }  
34}  
35

Twisted:

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


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
1# -*- coding:utf-8 –*-  
2  
3from struct import pack, unpack  
4from twisted.internet.protocol import Factory  
5from twisted.internet.protocol import Protocol  
6from twisted.internet import reactor  
7import StudentMsg_pb2  
8  
9# protobuf编码、解码器  
10class ProtobufProtocol(Protocol):  
11  
12    # 用于暂时存放接收到的数据  
13    _buffer = b""  
14  
15    def dataReceived(self, data):  
16        # 上次未处理的数据加上本次接收到的数据  
17        self._buffer = self._buffer + data  
18        # 一直循环直到新的消息没有接收完整  
19        while True:  
20            # 如果header接收完整  
21            if len(self._buffer) >= 4:  
22                # header部分,按大字节序转int,获取body长度  
23                length, = unpack(">I", self._buffer[0:4])  
24                # 如果body接收完整  
25                if len(self._buffer) >= 4 + length:  
26                    # body部分,protobuf字节码  
27                    packet = self._buffer[4:4 + length]  
28                      
29                    # protobuf字节码转成Student对象  
30                    student = StudentMsg_pb2.Student()  
31                    student.ParseFromString(packet)  
32                      
33                    # 调用protobufReceived传入Student对象  
34                    self.protobufReceived(student)  
35                      
36                    # 去掉_buffer中已经处理的消息部分  
37                    self._buffer = self._buffer[4 + length:]  
38                else:  
39                    break;  
40            else:  
41                break;  
42  
43    def protobufReceived(self, student):  
44        raise NotImplementedError  
45  
46    def sendProtobuf(self, student):  
47        # Student对象转为protobuf字节码  
48        data = student.SerializeToString()  
49        # 添加Header前缀指定protobuf字节码长度  
50        self.transport.write(pack(">I", len(data)) + data)  
51  
52# 逻辑代码  
53class TcpServerHandle(ProtobufProtocol):  
54  
55    # 实现ProtobufProtocol提供的protobufReceived  
56    def protobufReceived(self, student):  
57  
58        # 将接收到的Student输出  
59        print 'ID:' + str(student.id)  
60        print 'Name:' + student.name  
61        print 'Email:' + student.email  
62        print 'Friends:'  
63        for friend in student.friends:  
64            print friend  
65  
66        # 创建一个Student并发送给客户端  
67        student2 = StudentMsg_pb2.Student()  
68        student2.id = 9  
69        student2.name = '服务器'.decode('UTF-8') # 中文需要转成UTF-8字符串  
70        student2.email = '123@abc.com'  
71        student2.friends.append('X')  
72        student2.friends.append('Y')  
73        self.sendProtobuf(student2)  
74  
75factory = Factory()  
76factory.protocol = TcpServerHandle  
77reactor.listenTCP(8080, factory)  
78reactor.run()  
79

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


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
1public class TcpClient {  
2  
3    public static void main(String[] args) throws IOException {  
4  
5        Socket socket = null;  
6        DataOutputStream out = null;  
7        DataInputStream in = null;  
8          
9        try {  
10  
11            socket = new Socket("localhost", 8080);  
12            out = new DataOutputStream(socket.getOutputStream());  
13            in = new DataInputStream(socket.getInputStream());  
14              
15            // 创建一个Student传给服务器  
16            StudentMsg.Student.Builder builder = StudentMsg.Student.newBuilder();  
17            builder.setId(1);  
18            builder.setName("客户端");  
19            builder.setEmail("xxg@163.com");  
20            builder.addFriends("A");  
21            builder.addFriends("B");  
22            StudentMsg.Student student = builder.build();  
23            byte[] outputBytes = student.toByteArray(); // Student转成字节码  
24            out.writeInt(outputBytes.length); // write header  
25            out.write(outputBytes); // write body  
26            out.flush();  
27              
28            // 获取服务器传过来的Student  
29            int bodyLength = in.readInt();  // read header  
30            byte[] bodyBytes = new byte[bodyLength];  
31            in.readFully(bodyBytes);  // read body  
32            StudentMsg.Student student2 = StudentMsg.Student.parseFrom(bodyBytes); // body字节码解析成Student  
33            System.out.println("Header:" + bodyLength);  
34            System.out.println("Body:");  
35            System.out.println("ID:" + student2.getId());  
36            System.out.println("Name:" + student2.getName());  
37            System.out.println("Email:" + student2.getEmail());  
38            System.out.println("Friends:");  
39            List<String> friends = student2.getFriendsList();  
40            for(String friend : friends) {  
41                System.out.println(friend);  
42            }  
43  
44        } finally {  
45            // 关闭连接  
46            in.close();  
47            out.close();  
48            socket.close();  
49        }  
50    }  
51}  
52

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

服务器输出:

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

客户端输出:

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

MINA、Netty、Twisted一起学系列

MINA、Netty、Twisted一起学(一):实现简单的TCP服务器

MINA、Netty、Twisted一起学(二):TCP消息边界问题及按行分割消息

MINA、Netty、Twisted一起学(三):TCP消息固定大小的前缀(Header)

MINA、Netty、Twisted一起学(四):定制自己的协议

MINA、Netty、Twisted一起学(五):整合protobuf

MINA、Netty、Twisted一起学(六):session

MINA、Netty、Twisted一起学(七):发布/订阅(Publish/Subscribe)

MINA、Netty、Twisted一起学(八):HTTP服务器

MINA、Netty、Twisted一起学(九):异步IO和回调函数

MINA、Netty、Twisted一起学(十):线程模型

MINA、Netty、Twisted一起学(十一):SSL/TLS

MINA、Netty、Twisted一起学(十二):HTTPS

源码

https://github.com/wucao/mina-netty-twisted

给TA打赏
共{{data.count}}人
人已打赏
安全技术

bootstrap栅格系统自定义列

2021-12-21 16:36:11

安全技术

从零搭建自己的SpringBoot后台框架(二十三)

2022-1-12 12:36:11

个人中心
购物车
优惠劵
今日签到
有新私信 私信列表
搜索