Mina、Netty、Twisted一起学（四）：定制自己的协议

在前面的博文中，介绍一些消息分割的方案，以及MINA、Netty、Twisted针对这些方案提供的相关API。例如MINA的TextLineCodecFactory、PrefixedStringCodecFactory，Netty的LineBasedFrameDecoder、LengthFieldBasedFrameDecoder，Twisted的LineOnlyReceiver、Int32StringReceiver。

除了这些方案，还有很多其他方案，当然也可以自己定义。在这里，我们定制一个自己的方案，并分别使用MINA、Netty、Twisted实现对这种消息的解析和组装，也就是编码和解码。

上一篇博文中介绍了一种用固定字节数的Header来指定Body字节数的消息分割方案，其中Header部分是常规的大字节序（Big-Endian）的4字节整数。本文中对这个方案稍作修改，将固定字节数的Header改为小字节序（Little-Endian）的4字节整数。

常规的大字节序表示一个数的话，用高字节位的存放数字的低位，比较符合人的习惯。而小字节序和大字节序正好相反，用高字节位存放数字的高位。

Python中struct模块支持大小字节序的pack和unpack，在Java中可以用下面的两个方法实现小字节序字节数组转int和int转小字节序字节数组，下面的Java程序中将会用到这两个方法：

public class LittleEndian {  

    /**
     * 将int转成4字节的小字节序字节数组
     */
    public static byte[] toLittleEndian(int i) {
        byte[] bytes = new byte[4];
        bytes[0] = (byte) i;
        bytes[1] = (byte) (i >>> 8);
        bytes[2] = (byte) (i >>> 16);
        bytes[3] = (byte) (i >>> 24);
        return bytes;
    }  

    /**
     * 将小字节序的4字节的字节数组转成int
     */
    public static int getLittleEndianInt(byte[] bytes) {
        int b0 = bytes[0] & 0xFF;
        int b1 = bytes[1] & 0xFF;
        int b2 = bytes[2] & 0xFF;
        int b3 = bytes[3] & 0xFF;
        return b0 + (b1 << 8) + (b2 << 16) + (b3 << 24);
    }
}

无论是MINA、Netty还是Twisted，消息的编码、解码、切合的代码，都是应该和业务逻辑代码分开，这样有利于代码的开发、重用和维护。在MINA和Netty中类似，编码、解码需要继承实现相应的Encoder、Decoder，而在Twisted中则是继承Protocol实现编码解码。虽然实现方式不同，但是它们的功能都是一样的：

1、对消息根据一定规则进行切合，例如固定长度消息、按行、按分隔符、固定长度Header指定Body长度等；

2、将切合后的消息由字节码转成自己想要的类型，如MINA中将IoBuffer转成字符串，这样messageReceived接收到的message参数就是String类型；

3、write的时候可以传入自定义类型的参数，由编码器完成编码。

下面分别用MINA、Netty、Twisted实现4字节的小字节序int来指定body长度的消息的编码解码。

MINA：

在MINA中对接收到的消息进行切合和解码，一般会定义一个解码器类，继承自抽象类CumulativeProtocolDecoder，实现doDecode方法：

public class MyMinaDecoder 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();  

            byte[] bytes = new byte[4];
            in.get(bytes); // 读取4字节的Header  

            int bodyLength = LittleEndian.getLittleEndianInt(bytes); // 按小字节序转int  

            // 如果body没有接收完整，直接返回false
            if(in.remaining() < bodyLength) {
                in.reset(); // IoBuffer position回到原来标记的地方
                return false;
            } else {
                byte[] bodyBytes = new byte[bodyLength];
                in.get(bodyBytes);
                String body = new String(bodyBytes, "UTF-8");
                out.write(body); // 解析出一条消息
                return true;
            }
        }
    }
}

另外，session.write的时候要对数据编码，需要定义一个编码器，继承自抽象类ProtocolEncoderAdapter，实现encode方法：

public class MyMinaEncoder extends ProtocolEncoderAdapter {  

    @Override
    public void encode(IoSession session, Object message,
            ProtocolEncoderOutput out) throws Exception {  

        String msg = (String) message;
        byte[] bytes = msg.getBytes("UTF-8");
        int length = bytes.length;
        byte[] header = LittleEndian.toLittleEndian(length); // 按小字节序转成字节数组  

        IoBuffer buffer = IoBuffer.allocate(length + 4);
        buffer.put(header); // header
        buffer.put(bytes); // body
        buffer.flip();
        out.write(buffer);
    }
}

在服务器启动的时候加入相应的编码器和解码器：

public class TcpServer {  

    public static void main(String[] args) throws IOException {
        IoAcceptor acceptor = new NioSocketAcceptor();  

        // 指定编码解码器
        acceptor.getFilterChain().addLast("codec",
                new ProtocolCodecFilter(new MyMinaEncoder(), new MyMinaDecoder()));  

        acceptor.setHandler(new TcpServerHandle());
        acceptor.bind(new InetSocketAddress(8080));
    }
}

下面是业务逻辑的代码：

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 {  

        // MyMinaDecoder将接收到的数据由IoBuffer转为String
        String msg = (String) message;
        System.out.println("messageReceived:" + msg);  

        // MyMinaEncoder将write的字符串添加了一个小字节序Header并转为字节码
        session.write("收到");
    }
}

Netty：

Netty中解码器和MINA类似，解码器继承抽象类ByteToMessageDecoder，实现decode方法：

public class MyNettyDecoder extends ByteToMessageDecoder {  

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {  

        // 如果没有接收完Header部分（4字节），直接退出该方法
        if(in.readableBytes() >= 4) {  

            // 标记开始位置，如果一条消息没传输完成则返回到这个位置
            in.markReaderIndex();  

            byte[] bytes = new byte[4];
            in.readBytes(bytes); // 读取4字节的Header  

            int bodyLength = LittleEndian.getLittleEndianInt(bytes); // header按小字节序转int  

            // 如果body没有接收完整
            if(in.readableBytes() < bodyLength) {
                in.resetReaderIndex(); // ByteBuf回到标记位置
            } else {
                byte[] bodyBytes = new byte[bodyLength];
                in.readBytes(bodyBytes);
                String body = new String(bodyBytes, "UTF-8");
                out.add(body); // 解析出一条消息
            }
        }
    }
}

下面是编码器，继承自抽象类MessageToByteEncoder，实现encode方法：

public class MyNettyEncoder extends MessageToByteEncoder<String> {  

    @Override
    protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out)
            throws Exception {  

        byte[] bytes = msg.getBytes("UTF-8");
        int length = bytes.length;
        byte[] header = LittleEndian.toLittleEndian(length); // int按小字节序转字节数组
        out.writeBytes(header); // write header
        out.writeBytes(bytes); // write 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();  

                            // 加上自己的Encoder和Decoder
                            pipeline.addLast(new MyNettyDecoder());
                            pipeline.addLast(new MyNettyEncoder());  

                            pipeline.addLast(new TcpServerHandler());
                        }
                    });
            ChannelFuture f = b.bind(8080).sync();
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

业务逻辑处理类：

public class TcpServerHandler extends ChannelInboundHandlerAdapter {  

    // 接收到新的数据
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {  

        // MyNettyDecoder将接收到的数据由ByteBuf转为String
        String message = (String) msg;
        System.out.println("channelRead:" + message);  

        // MyNettyEncoder将write的字符串添加了一个小字节序Header并转为字节码
        ctx.writeAndFlush("收到");
    }  

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

Twisted：

Twisted的实现方式和MINA、Netty不太一样，其实现方式相对来说更加原始，但是越原始也越接近底层原理。

首先要定义一个MyProtocol类继承自Protocol，用于充当类似于MINA、Netty的编码、解码器。处理业务逻辑的类TcpServerHandle继承MyProtocol，重写或调用MyProtocol提供的一些方法。

# -*- 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  

# 编码、解码器
class MyProtocol(Protocol):  

    # 用于暂时存放接收到的数据
    _buffer = b""  

    def dataReceived(self, data):
        # 上次未处理的数据加上本次接收到的数据
        self._buffer = self._buffer + data
        # 一直循环直到新的消息没有接收完整
        while True:
            # 如果header接收完整
            if len(self._buffer) >= 4:
                # 按小字节序转int
                length, = unpack("<I", self._buffer[0:4])
                # 如果body接收完整
                if len(self._buffer) >= 4 + length:
                    # body部分
                    packet = self._buffer[4:4 + length]
                    # 新的一条消息接收并解码完成，调用stringReceived
                    self.stringReceived(packet)
                    # 去掉_buffer中已经处理的消息部分
                    self._buffer = self._buffer[4 + length:]
                else:
                    break;
            else:
                break;  

    def stringReceived(self, data):
        raise NotImplementedError  

    def sendString(self, string):
        self.transport.write(pack("<I", len(string)) + string)  

# 逻辑代码
class TcpServerHandle(MyProtocol):  

    # 实现MyProtocol提供的stringReceived而不是dataReceived，不然无法解码
    def stringReceived(self, data):  

        # data为MyProtocol解码后的数据
        print ‘stringReceived:‘ + data  

        # 调用sendString而不是self.transport.write，不然不能进行编码
        self.sendString("收到")  

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;
        OutputStream out = null;
        InputStream in = null;  

        try {  

            socket = new Socket("localhost", 8080);
            out = socket.getOutputStream();
            in = socket.getInputStream();  

            // 请求服务器
            String data = "我是客户端";
            byte[] outputBytes = data.getBytes("UTF-8");
            out.write(LittleEndian.toLittleEndian(outputBytes.length)); // write header
            out.write(outputBytes); // write body
            out.flush();  

            // 获取响应
            byte[] inputBytes = new byte[1024];
            int length = in.read(inputBytes);
            if(length >= 4) {
                int bodyLength = LittleEndian.getLittleEndianInt(inputBytes);
                if(length >= 4 + bodyLength) {
                    byte[] bodyBytes = Arrays.copyOfRange(inputBytes, 4, 4 + bodyLength);
                    System.out.println("Header:" + bodyLength);
                    System.out.println("Body:" + new String(bodyBytes, "UTf-8"));
                }
            }  

        } finally {
            // 关闭连接
            in.close();
            out.close();
            socket.close();
        }
    }
}

用客户端分别测试上面三个TCP服务器：

MINA服务器输出结果：

messageReceived:我是客户端

Netty服务器输出结果：

channelRead:我是客户端

Twisted服务器输出结果：

channelRead:我是客户端

客户端测试三个服务器的输出结果都是：

Header:6
Body:收到

由于一个汉字一般占3个字节，所以两个汉字对应的Header为6。

时间： 2024-10-28 10:30:45

Mina、Netty、Twisted一起学（四）：定制自己的协议的相关文章

Mina、Netty、Twisted一起学（五）：整合protobuf

protobuf是谷歌的Protocol Buffers的简称,用于结构化数据和字节码之间互相转换(序列化.反序列化),一般应用于网络传输,可支持多种编程语言. protobuf怎样使用这里不再介绍,本文主要介绍在MINA.Netty.Twisted中怎样使用protobuf,不了解protobuf的同学能够去參考我的还有一篇博文. 在前面的一篇博文中.有介绍到一种用一个固定为4字节的前缀Header来指定Body的字节数的一种消息切割方式.在这里相同要使用到. 仅仅是当中Body的内容不再是字

Mina、Netty、Twisted一起学（十）：线程模型

要想开发一个高性能的TCPserver,熟悉所使用框架的线程模型非常重要.MINA.Netty.Twisted本身都是高性能的网络框架,假设再搭配上高效率的代码.才干实现一个高大上的server. 可是假设不了解它们的线程模型.就非常难写出高性能的代码.框架本身效率再高.程序写的太差,那么server总体的性能也不会太高.就像一个电脑,CPU再好.内存小硬盘慢散热差,总体的性能也不会太高. 玩过Android开发的同学会知道,在Android应用中有一个非常重要线程:UI线程(即主线程). UI

Mina、Netty、Twisted一起学（七）：公布/订阅（Publish/Subscribe）

消息传递有非常多种方式.请求/响应(Request/Reply)是最经常使用的.在前面的博文的样例中.非常多都是採用请求/响应的方式.当server接收到消息后,会马上write回写一条消息到client. HTTP协议也是基于请求/响应的方式. 可是请求/响应并不能满足全部的消息传递的需求,有些需求可能须要服务端主动推送消息到client,而不是被动的等待请求后再给出响应. 公布/订阅(Publish/Subscribe)是一种server主动发送消息到client的消息传递方式.订阅者Sub

Mina、Netty、Twisted一起学：实现简单的TCP服务器

MINA.Netty.Twisted为什么放在一起学习?首先,不妨先看一下他们官方网站对其的介绍: MINA: Apache MINA is a network application framework which helps users develop high performance and high scalability network applications easily. It provides an abstract event-driven asynchronous API

Mina、Netty、Twisted一起学（八）：HTTP服务器

HTTP协议应该是目前使用最多的应用层协议了,用浏览器打开一个网站就是使用HTTP协议进行数据传输. HTTP协议也是基于TCP协议,所以也有服务器和客户端.HTTP客户端一般是浏览器,当然还有可能是其他东西.HTTP服务器,也就是Web服务器,目前已经有很多成熟的产品,例如Apache HTTP Server.Tomcat.Nginx.IIS等. 本文的内容不是讲解如何使用以上的HTTP服务器,而是要分别用MINA.Netty.Twisted实现一个简单的HTTP服务器. 首先,要简单了解一下

入木三分学网络第一篇--VRRP协议详解第一篇（转）

因为keepalived使用了VRRP协议,所有有必要熟悉一下. 虚拟路由冗余协议(Virtual Router Redundancy Protocol,简称VRRP)是解决局域网中配置静态网关时,静态网关出现单点失效现象的路由协议. VRRP广泛应用在边缘网络中,它的设计目标是支持特定情况下IP数据流量失败转移不会引起混乱,允许主机使用单路由器(位于一个虚拟路由器组中, 在该组中,只有一台路由器--master路由器工作,转发数据包,其它路由器是backup路由器,不参与转发数据包),以及在实

Mina、Netty、Twisted一起学（一）：实现简单的TCP服务器

MINA.Netty.Twisted为什么放在一起学习?首先,不妨先看一下他们官方网站对其的介绍: MINA: Apache MINA is a network application framework which helps users develop high performance and high scalability network applications easily. It provides an abstract event-driven asynchronous API

Mina、Netty、Twisted一起学（九）：异步IO和回调函数

用过JavaScript或者jQuery的同学都知道,JavaScript特别是jQuery中存在大量的回调函数,比如Ajax.jQuery的动画等. $.get(url, function() { doSomething1(); // (3) }); // (1) doSomething2(); // (2) 上面的代码是jQuery的Ajax,因为Ajax是异步的,所以在请求URL的过程中并不会堵塞程序,也就是程序运行到(1)并不用等待Ajax请求的结果,就继续往下运行(2).而$.get的

Mina、Netty、Twisted一起学（七）：发布/订阅（Publish/Subscribe）

消息传递有很多种方式,请求/响应(Request/Reply)是最常用的.在前面的博文的例子中,很多都是采用请求/响应的方式,当服务器接收到消息后,会立即write回写一条消息到客户端.HTTP协议也是基于请求/响应的方式. 但是请求/响应并不能满足所有的消息传递的需求,有些需求可能需要服务端主动推送消息到客户端,而不是被动的等待请求后再给出响应. 发布/订阅(Publish/Subscribe)是一种服务器主动发送消息到客户端的消息传递方式.订阅者Subscriber连接到服务器客户端后,相当