文章目录
  1. 1. Netty
  2. 2. MINA
  3. 3. Twisted
  • MINA、Netty、Twisted一起学系列
    1. 1. 源码
  • protobuf是谷歌的Protocol Buffers的简称,用于结构化数据和字节码之间互相转换(序列化、反序列化),一般应用于网络传输,可支持多种编程语言。

    protobuf如何使用这里不再介绍,本文主要介绍在MINA、Netty、Twisted中如何使用protobuf,不了解protobuf的同学可以去参考我的另一篇博文

    前一篇博文中,有介绍到一种用一个固定为4字节的前缀Header来指定Body的字节数的一种消息分割方式,在这里同样要使用到。只是其中Body的内容不再是字符串,而是protobuf字节码。
    protobuf消息
    在处理业务逻辑时,肯定不希望还要对数据进行序列化和反序列化,而是希望直接操作一个对象,那么就需要有相应的编码器和解码器,将序列化和反序列化的逻辑写在编码器和解码器中。有关编码器和解码器的实现,上一篇博文中有介绍。

    Netty包中已经自带针对protobuf的编码器和解码器,那么就不用再自己去实现了。而MINA、Twisted还需要自己去实现protobuf的编码器和解码器。

    这里定义一个protobuf数据结构,用于描述一个学生的信息,保存为StudentMsg.proto文件:

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    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信息。

    Netty

    Netty自带protobuf的编码器和解码器,分别是ProtobufEncoder和ProtobufDecoder。需要注意的是,ProtobufEncoder和ProtobufDecoder只负责protobuf的序列化和反序列化,而处理消息Header前缀和消息分割的还需要LengthFieldBasedFrameDecoder和LengthFieldPrepender。LengthFieldBasedFrameDecoder即用于解析消息Header前缀,根据Header中指定的Body字节数截取Body,LengthFieldPrepender用于在wirte消息时在消息前面添加一个Header前缀来指定Body字节数。

    1
    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
    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对象:

    1
    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
    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一样的编码器和解码器。

    编码器:

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    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);
    }
    }

    解码器:

    1
    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
    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的编码器和解码器:

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    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一样了:

    1
    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
    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提供的方法。

    1
    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
    # -*- 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编写的一个客户端测试程序:

    1
    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
    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

    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

    源码

    https://github.com/wucao/mina-netty-twisted

    文章目录
    1. 1. Netty
    2. 2. MINA
    3. 3. Twisted
  • MINA、Netty、Twisted一起学系列
    1. 1. 源码