Mina、Netty、Twisted一塊兒學（五）：整合protobuf

時間 2019-11-29

標籤 mina netty twisted 一塊兒整合 protobuf 欄目 Netty 简体版

原文原文鏈接

protobuf是谷歌的Protocol Buffers的簡稱，用於結構化數據和字節碼之間互相轉換（序列化、反序列化），通常應用於網絡傳輸，可支持多種編程語言。html

protobuf如何使用這裏再也不介紹，本文主要介紹在MINA、Netty、Twisted中如何使用protobuf，不瞭解protobuf的同窗能夠去參考個人另外一篇博文。react

在前面的一篇博文中，有介紹到一種用一個固定爲4字節的前綴Header來指定Body的字節數的一種消息分割方式，在這裏一樣要使用到。只是其中Body的內容再也不是字符串，而是protobuf字節碼。git

在處理業務邏輯時，確定不但願還要對數據進行序列化和反序列化，而是但願直接操做一個對象，那麼就須要有相應的編碼器和解碼器，將序列化和反序列化的邏輯寫在編碼器和解碼器中。有關編碼器和解碼器的實現，上一篇博文中有介紹。github

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信息。session

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();  
    }  
}

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);  
    }  
}

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部分，按大字節序轉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 = TcpServerHandle  
reactor.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();  
        }  
    }  
}