释放双眼，带上耳机，听听看~！
protobuf是谷歌的Protocol Buffers的简称，用于结构化数据和字节码之间互相转换（序列化、反序列化），一般应用于网络传输，可支持多种编程语言。
protobuf如何使用这里不再介绍，本文主要介绍在MINA、Netty、Twisted中如何使用protobuf，不了解protobuf的同学可以去参考我的另一篇博文。
在前面的一篇博文中，有介绍到一种用一个固定为4字节的前缀Header来指定Body的字节数的一种消息分割方式，在这里同样要使用到。只是其中Body的内容不再是字符串，而是protobuf字节码。
在处理业务逻辑时，肯定不希望还要对数据进行序列化和反序列化，而是希望直接操作一个对象，那么就需要有相应的编码器和解码器，将序列化和反序列化的逻辑写在编码器和解码器中。有关编码器和解码器的实现，上一篇博文中有介绍。
Netty包中已经自带针对protobuf的编码器和解码器，那么就不用再自己去实现了。而MINA、Twisted还需要自己去实现protobuf的编码器和解码器。
这里定义一个protobuf数据结构，用于描述一个学生的信息，保存为StudentMsg.proto文件：
2
3
4
5
6
7
8
9
10
11
12
13
14
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字节数。
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
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对象：
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
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一样的编码器和解码器。
编码器：
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
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
解码器：
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
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的编码器和解码器：
2
3
4
5
6
7
8
9
10
11
12
13
14
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一样了：
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
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提供的方法。
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
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编写的一个客户端测试程序：
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
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
