中文字幕av专区_日韩电影在线播放_精品国产精品久久一区免费式_av在线免费观看网站

溫馨提示×

溫馨提示×

您好,登錄后才能下訂單哦!

密碼登錄×
登錄注冊×
其他方式登錄
點擊 登錄注冊 即表示同意《億速云用戶服務條款》

Netty、MINA、Twisted一起學系列05:整合protobuf

發布時間:2020-08-14 18:14:56 來源:ITPUB博客 閱讀:176 作者:程序員私房菜 欄目:互聯網科技

文章已獲得作者授權,原文地址: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, 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();
       }
   }
}

用客戶端分別測試上面三個TCP服務器:

服務器輸出:

ID:1

Name:客戶端

Email:xxg@163.com

Friends:

A

B

客戶端輸出:

Header:32

Body:

ID:9

Name:服務器

Email:123@abc.com

Friends:

X

Y

向AI問一下細節

免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。

AI

井陉县| 富阳市| 上高县| 庆云县| 焉耆| 墨玉县| 承德市| 凌云县| 佛山市| 宜阳县| 博野县| 改则县| 丰县| 江安县| 清水河县| 蓝山县| 仁寿县| 红桥区| 北宁市| 湖州市| 安阳县| 临清市| 嘉鱼县| 桂东县| 永城市| 闵行区| 潜江市| 明水县| 扬州市| 布尔津县| 称多县| 陇川县| 鹤壁市| 同江市| 商南县| 云阳县| 讷河市| 青川县| 正定县| 含山县| 井研县|