Shulou(Shulou.com)06/02 Report--

The article has been authorized by the author, original address: xxgblog.com/2014/08/27/mina-netty-twisted-5

Protobuf is the abbreviation of Google's Protocol Buffers, which is used for the conversion between structured data and bytecode (serialization, deserialization). It is generally used in network transmission and can support a variety of programming languages.

How to use protobuf is not discussed here, but this article focuses on how to use protobuf in MINA, Netty, and Twisted.

In the previous article (Netty, MINA, Twisted learn Series 04: customize your own protocol), we introduced a message segmentation method that uses a fixed prefix of 4 bytes Header to specify the number of bytes of Body, which is also used here. It's just that the content of Body is no longer a string, but a protobuf bytecode.

When dealing with business logic, you certainly don't want to serialize and deserialize the data, but you want to manipulate an object directly, so you need to have the corresponding encoder and decoder to write the serialization and deserialization logic in the encoder and decoder. The implementation of encoders and decoders was introduced in the previous article.

The Netty package already comes with an encoder and decoder for protobuf, so you don't have to implement it yourself. MINA and Twisted also need to implement the encoder and decoder of protobuf.

A protobuf data structure is defined here, which is used to describe the information of a student and saved as a StudentMsg.proto file:

Message Student {

/ / ID

Required int32 id = 1

/ / name

Required string name = 2

/ / email

Optional string email = 3

/ / Friends

Repeated string friends = 4

}

Use StudentMsg.proto to generate Java and Python code respectively, and add the code to the corresponding project. The generated code is no longer posted. The following describes how to use protobuf to transmit Student information in Netty, MINA and Twisted, respectively.

1 Netty

Netty has its own protobuf encoder and decoder, which are ProtobufEncoder and ProtobufDecoder respectively. It is important to note that ProtobufEncoder and ProtobufDecoder are only responsible for serialization and deserialization of protobuf, while LengthFieldBasedFrameDecoder and LengthFieldPrepender are also required to handle message Header prefixes and message segmentation. LengthFieldBasedFrameDecoder is used to parse the message Header prefix and intercept the Body,LengthFieldPrepender according to the number of Body bytes specified in the Header to specify the number of Body bytes by adding a Header prefix to the wirte message.

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 () {

@ Override

Public void initChannel (SocketChannel ch)

Throws Exception {

ChannelPipeline pipeline = ch.pipeline ()

/ / responsible for slicing messages with the Body length specified by 4-byte Header

Pipeline.addLast ("frameDecoder"

New LengthFieldBasedFrameDecoder (1048576, 0, 4, 0, 4))

/ / responsible for converting the protobuf bytecode of a complete message processed by frameDecoder into a Student object

Pipeline.addLast ("protobufDecoder"

New ProtobufDecoder (StudentMsg.Student.getDefaultInstance ())

/ / is responsible for specifying the Body length by adding a 4-byte Header prefix to the written bytecode

Pipeline.addLast ("frameEncoder", new LengthFieldPrepender (4))

/ / responsible for converting Student objects to protobuf bytecodes

Pipeline.addLast ("protobufEncoder", new ProtobufEncoder ())

Pipeline.addLast (new TcpServerHandler ())

}

});

ChannelFuture f = b.bind (8080) .sync ()

F.channel (). CloseFuture (). Sync ()

} finally {

WorkerGroup.shutdownGracefully ()

BossGroup.shutdownGracefully ()

}

}

}

When handling events, the parameters received and sent are directly Student objects.

Public class TcpServerHandler extends ChannelInboundHandlerAdapter {

@ Override

Public void channelRead (ChannelHandlerContext ctx, Object msg) {

/ / read the Student object passed by the client

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 friends = student.getFriendsList ()

For (String friend: friends) {

System.out.println (friend)

}

/ / create a new Student object and transfer it to the client

StudentMsg.Student.Builder builder = StudentMsg.Student.newBuilder ()

Builder.setId (9)

Builder.setName ("server")

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

There are no encoders and decoders for protobuf in MINA, but you can implement an encoder and decoder with the same function as Netty.

Encoder:

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 object is converted to protobuf bytecode

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)

}

}

Decoder:

Public class MinaProtobufDecoder extends CumulativeProtocolDecoder {

@ Override

Protected boolean doDecode (IoSession session, IoBuffer in

ProtocolDecoderOutput out) throws Exception {

/ / if you have not received the Header part (4 bytes), return false directly

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 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, unpack from twisted.internet.protocol import Factory from twisted.internet.protocol import Protocol from twisted.internet import reactor import 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 part, transfer to int in large byte order to get body length

Length, = unpack ("> I", self._buffer [0:4])

# if the body receives the full

If len (self._buffer) > = 4 + length:

# body part, protobuf bytecode

Packet = self._buffer [4:4 + length]

# protobuf bytecode converted to Student object

Student = StudentMsg_pb2.Student ()

Student.ParseFromString (packet)

# call protobufReceived to pass in Student object

Self.protobufReceived (student)

# remove the message parts that have been processed in _ buffer

Self._buffer = self._buffer [4 + length:]

Else:

Break

Else:

Break

Def protobufReceived (self, student):

Raise NotImplementedError

Def sendProtobuf (self, student):

# Student object is converted to protobuf bytecode

Data = student.SerializeToString ()

# add Header prefix to specify protobuf bytecode length

Self.transport.write (pack ("> I", len (data)) + data)

# logical code

Class TcpServerHandle (ProtobufProtocol):

# implement the protobufReceived provided by ProtobufProtocol

Def protobufReceived (self, student):

# output the received Student

Print 'ID:' + str (student.id)

Print 'Name:' + student.name

Print 'Email:' + student.email

Print 'Friends:'

For friend in student.friends:

Print friend

# create a Student and send it to the client

Student2 = StudentMsg_pb2.Student ()

Student2.id = 9

Student2.name = 'server' .decode ('UTF-8') # Chinese needs to be converted to UTF-8 string

Student2.email = '123 roomabc.com'

Student2.friends.append ('X')

Student2.friends.append ('Y')

Self.sendProtobuf (student2)

Factory = Factory ()

Factory.protocol = TcpServerHandle

Reactor.listenTCP (8080, factory)

Reactor.run ()

Here is a client-side test program written by 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 ())

/ / create a Student and pass it to the server

StudentMsg.Student.Builder builder = StudentMsg.Student.newBuilder ()

Builder.setId (1)

Builder.setName ("client")

Builder.setEmail ("xxg@163.com")

Builder.addFriends ("A")

Builder.addFriends ("B")

StudentMsg.Student student = builder.build ()

Byte [] outputBytes = student.toByteArray (); / / Student converted to bytecode

Out.writeInt (outputBytes.length); / / write header

Out.write (outputBytes); / / write body

Out.flush ()

/ / get the Student sent from the server

Int bodyLength = in.readInt (); / / read header

Byte [] bodyBytes = new byte [bodyLength]

In.readFully (bodyBytes); / / read body

StudentMsg.Student student2 = StudentMsg.Student.parseFrom (bodyBytes); / / body bytecode is parsed into 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 friends = student2.getFriendsList ()

For (String friend: friends) {

System.out.println (friend)

}

} finally {

/ / close the connection

In.close ()

Out.close ()

Socket.close ()

}

}

}

Test the above three TCP servers with the client:

Server output:

ID:1

Name: client

Email:xxg@163.com

Friends:

A

B

Client output:

Header:32

Body:

ID:9

Name: server

Email:123@abc.com

Friends:

X

Y

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.