您好,登錄后才能下訂單哦!
文章已獲得作者授權(quán),原文地址:xxgblog.com/2014/08/27/mina-netty-twisted-5
protobuf 是谷歌的 Protocol Buffers 的簡(jiǎn)稱,用于結(jié)構(gòu)化數(shù)據(jù)和字節(jié)碼之間互相轉(zhuǎn)換(序列化、反序列化),一般應(yīng)用于網(wǎng)絡(luò)傳輸,可支持多種編程語(yǔ)言。
protobuf 如何使用這里不再介紹,本文主要介紹在 MINA、Netty、Twisted 中如何使用 protobuf。
在前一篇文章(Netty、MINA、Twisted一起學(xué)系列04:定制自己的協(xié)議)中,有介紹到一種用一個(gè)固定為4字節(jié)的前綴Header來(lái)指定Body的字節(jié)數(shù)的一種消息分割方式,在這里同樣要使用到。只是其中 Body 的內(nèi)容不再是字符串,而是 protobuf 字節(jié)碼。
在處理業(yè)務(wù)邏輯時(shí),肯定不希望還要對(duì)數(shù)據(jù)進(jìn)行序列化和反序列化,而是希望直接操作一個(gè)對(duì)象,那么就需要有相應(yīng)的編碼器和解碼器,將序列化和反序列化的邏輯寫在編碼器和解碼器中。有關(guān)編碼器和解碼器的實(shí)現(xiàn),上一篇文章中有介紹。
Netty 包中已經(jīng)自帶針對(duì) protobuf 的編碼器和解碼器,那么就不用再自己去實(shí)現(xiàn)了。而 MINA、Twisted 還需要自己去實(shí)現(xiàn) protobuf 的編碼器和解碼器。
這里定義一個(gè)protobuf數(shù)據(jù)結(jié)構(gòu),用于描述一個(gè)學(xué)生的信息,保存為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 代碼,將代碼加入到相應(yīng)的項(xiàng)目中。生成的代碼就不再貼上來(lái)了。下面分別介紹在 Netty、MINA、Twisted 如何使用 protobuf 來(lái)傳輸 Student 信息。
1 Netty
Netty 自帶 protobuf 的編碼器和解碼器,分別是 ProtobufEncoder 和 ProtobufDecoder。需要注意的是,ProtobufEncoder 和 ProtobufDecoder 只負(fù)責(zé) protobuf 的序列化和反序列化,而處理消息 Header 前綴和消息分割的還需要 LengthFieldBasedFrameDecoder 和 LengthFieldPrepender。LengthFieldBasedFrameDecoder 即用于解析消息 Header 前綴,根據(jù) Header 中指定的 Body 字節(jié)數(shù)截取 Body,LengthFieldPrepender 用于在wirte消息時(shí)在消息前面添加一個(gè) Header 前綴來(lái)指定 Body 字節(jié)數(shù)。
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();
// 負(fù)責(zé)通過(guò)4字節(jié)Header指定的Body長(zhǎng)度將消息切割
pipeline.addLast("frameDecoder",
new LengthFieldBasedFrameDecoder(1048576, 0, 4, 0, 4));
// 負(fù)責(zé)將frameDecoder處理后的完整的一條消息的protobuf字節(jié)碼轉(zhuǎn)成Student對(duì)象
pipeline.addLast("protobufDecoder",
new ProtobufDecoder(StudentMsg.Student.getDefaultInstance()));
// 負(fù)責(zé)將寫入的字節(jié)碼加上4字節(jié)Header前綴來(lái)指定Body長(zhǎng)度
pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
// 負(fù)責(zé)將Student對(duì)象轉(zhuǎn)成protobuf字節(jié)碼
pipeline.addLast("protobufEncoder", new ProtobufEncoder());
pipeline.addLast(new TcpServerHandler());
}
});
ChannelFuture f = b.bind(8080).sync();
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
}
處理事件時(shí),接收和發(fā)送的參數(shù)直接就是Student對(duì)象。
public class TcpServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
// 讀取客戶端傳過(guò)來(lái)的Student對(duì)象
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);
}
// 新建一個(gè)Student對(duì)象傳到客戶端
StudentMsg.Student.Builder builder = StudentMsg.Student.newBuilder();
builder.setId(9);
builder.setName("服務(wù)器");
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 中沒(méi)有針對(duì) protobuf 的編碼器和解碼器,但是可以自己實(shí)現(xiàn)一個(gè)功能和 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對(duì)象轉(zhuǎn)為protobuf字節(jié)碼
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 {
// 如果沒(méi)有接收完Header部分(4字節(jié)),直接返回false
if (in.remaining() < 4) {
return false;
} else {
// 標(biāo)記開始位置,如果一條消息沒(méi)傳輸完成則返回到這個(gè)位置
in.mark();
// 讀取header部分,獲取body長(zhǎng)度
int bodyLength = in.getInt();
// 如果body沒(méi)有接收完整,直接返回false
if (in.remaining() < bodyLength) {
in.reset(); // IoBuffer position回到原來(lái)標(biāo)記的地方
return false;
} else {
byte[] bodyBytes = new byte[bodyLength];
in.get(bodyBytes); // 讀取body部分
StudentMsg.Student student = StudentMsg.Student.parseFrom(bodyBytes); // 將body中protobuf字節(jié)碼轉(zhuǎn)成Student對(duì)象
out.write(student); // 解析出一條消息
return true;
}
}
}
}
MINA 服務(wù)器加入 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));
}
}
這樣,在處理業(yè)務(wù)邏輯時(shí),就和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 {
// 讀取客戶端傳過(guò)來(lái)的Student對(duì)象
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);
}
// 新建一個(gè)Student對(duì)象傳到客戶端
StudentMsg.Student.Builder builder = StudentMsg.Student.newBuilder();
builder.setId(9);
builder.setName("服務(wù)器");
builder.setEmail("123@abc.com");
builder.addFriends("X");
builder.addFriends("Y");
StudentMsg.Student student2 = builder.build();
session.write(student2);
}
}
3. Twisted
在 Twisted 中,首先定義一個(gè) ProtobufProtocol 類,繼承 Protocol 類,充當(dāng)編碼器和解碼器。處理業(yè)務(wù)邏輯的 TcpServerHandle 類再繼承 ProtobufProtocol 類,調(diào)用或重寫 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):
# 用于暫時(shí)存放接收到的數(shù)據(jù)
_buffer = b""
def dataReceived(self, data):
# 上次未處理的數(shù)據(jù)加上本次接收到的數(shù)據(jù)
self._buffer = self._buffer + data
# 一直循環(huán)直到新的消息沒(méi)有接收完整
while True:
# 如果header接收完整
if len(self._buffer) >= 4:
# header部分,按大字節(jié)序轉(zhuǎn)int,獲取body長(zhǎng)度
length, = unpack(">I", self._buffer[0:4])
# 如果body接收完整
if len(self._buffer) >= 4 + length:
# body部分,protobuf字節(jié)碼
packet = self._buffer[4:4 + length]
# protobuf字節(jié)碼轉(zhuǎn)成Student對(duì)象
student = StudentMsg_pb2.Student()
student.ParseFromString(packet)
# 調(diào)用protobufReceived傳入Student對(duì)象
self.protobufReceived(student)
# 去掉_buffer中已經(jīng)處理的消息部分
self._buffer = self._buffer[4 + length:]
else:
break;
else:
break;
def protobufReceived(self, student):
raise NotImplementedError
def sendProtobuf(self, student):
# Student對(duì)象轉(zhuǎn)為protobuf字節(jié)碼
data = student.SerializeToString()
# 添加Header前綴指定protobuf字節(jié)碼長(zhǎng)度
self.transport.write(pack(">I", len(data)) + data)
# 邏輯代碼
class TcpServerHandle(ProtobufProtocol):
# 實(shí)現(xiàn)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
# 創(chuàng)建一個(gè)Student并發(fā)送給客戶端
student2 = StudentMsg_pb2.Student()
student2.id = 9
student2.name = '服務(wù)器'.decode('UTF-8') # 中文需要轉(zhuǎn)成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編寫的一個(gè)客戶端測(cè)試程序:
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());
// 創(chuàng)建一個(gè)Student傳給服務(wù)器
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轉(zhuǎn)成字節(jié)碼
out.writeInt(outputBytes.length); // write header
out.write(outputBytes); // write body
out.flush();
// 獲取服務(wù)器傳過(guò)來(lái)的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字節(jié)碼解析成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 {
// 關(guān)閉連接
in.close();
out.close();
socket.close();
}
}
}
用客戶端分別測(cè)試上面三個(gè)TCP服務(wù)器:
服務(wù)器輸出:
ID:1
Name:客戶端
Email:xxg@163.com
Friends:
A
B
客戶端輸出:
Header:32
Body:
ID:9
Name:服務(wù)器
Email:123@abc.com
Friends:
X
Y
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。