Shulou(Shulou.com)06/01 Report--

This article mainly introduces how to use MINA, Netty, Twisted to achieve message segmentation, with certain reference value, interested friends can refer to, I hope you have a lot of gains after reading this article, let Xiaobian take you to understand.

This article introduces a method of message segmentation, using a fixed length header that indicates the length of the body, and a fixed byte header prefix to specify the number of bytes of the Body, so as to segment the message.

固定字节数的Header前缀来指定Body的字节数

上面图中 Header 固定为 4 字节，Header 中保存的是一个 4 字节（32位）的整数，例如 12 即为 0x0000000C，这个整数用来指定 Body 的长度（字节数）。当读完这么多字节的 Body 之后，又是下一条消息的 Header。

下面分别用MINA、Netty、Twisted来实现对这种消息的切合和解码。

MINA

MINA 提供了 PrefixedStringCodecFactory 来对这种类型的消息进行编码解码，PrefixedStringCodecFactory 默认 Header 的大小是4字节，当然也可以指定成1或2。

public class TcpServer {

public static void main(String[] args) throws IOException {

IoAcceptor acceptor = new NioSocketAcceptor();

// 4字节的Header指定Body的字节数，对这种消息的处理

acceptor.getFilterChain().addLast("codec",

new ProtocolCodecFilter(new PrefixedStringCodecFactory(Charset.forName("UTF-8"))));

acceptor.setHandler(new TcpServerHandle());

acceptor.bind(new InetSocketAddress(8080));

}

}

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 {

String msg = (String) message;

System.out.println("messageReceived:" + msg);

}

@Override

public void sessionCreated(IoSession session) throws Exception {

System.out.println("sessionCreated");

}

@Override

public void sessionClosed(IoSession session) throws Exception {

System.out.println("sessionClosed");

}

}

Netty

Netty 使用 LengthFieldBasedFrameDecoder 来处理这种消息。下面代码中的new LengthFieldBasedFrameDecoder(80, 0, 4, 0, 4)中包含5个参数，分别是int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip。maxFrameLength为消息的最大长度，lengthFieldOffset为Header的位置，lengthFieldLength为Header的长度，lengthAdjustment为长度调整（默认Header中的值表示Body的长度，并不包含Header自己），initialBytesToStrip为去掉字节数（默认解码后返回Header+Body的全部内容，这里设为4表示去掉4字节的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() {

@Override

public void initChannel(SocketChannel ch)

throws Exception {

ChannelPipeline pipeline = ch.pipeline();

// LengthFieldBasedFrameDecoder按行分割消息，取出body

pipeline.addLast(new LengthFieldBasedFrameDecoder(80, 0, 4, 0, 4));

// 再按UTF-8编码转成字符串

pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));

pipeline.addLast(new TcpServerHandler());

}

});

ChannelFuture f = b.bind(8080).sync();

f.channel().closeFuture().sync();

} finally {

workerGroup.shutdownGracefully();

bossGroup.shutdownGracefully();

}

}

}

class TcpServerHandler extends ChannelInboundHandlerAdapter {

// 接收到新的数据

@Override

public void channelRead(ChannelHandlerContext ctx, Object msg) {

String message = (String) msg;

System.out.println("channelRead:" + message);

}

@Override

public void channelActive(ChannelHandlerContext ctx) {

System.out.println("channelActive");

}

@Override

public void channelInactive(ChannelHandlerContext ctx) {

System.out.println("channelInactive");

}

@Override

public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

cause.printStackTrace();

ctx.close();

}

}

Twisted

在Twisted中需要继承Int32StringReceiver，不再继承Protocol。Int32StringReceiver表示固定32位（4字节）的Header，另外还有Int16StringReceiver、Int8StringReceiver等。而需要实现的接受数据事件的方法不再是dataReceived，也不是lineReceived，而是stringReceived。

# -*- coding:utf-8 -*-

from twisted.protocols.basic import Int32StringReceiver

from twisted.internet.protocol import Factory

from twisted.internet import reactor

class TcpServerHandle(Int32StringReceiver):

# 新的连接建立

def connectionMade(self):

print 'connectionMade'

# 连接断开

def connectionLost(self, reason):

print 'connectionLost'

# 接收到新的数据

def stringReceived(self, data):

print 'stringReceived:' + data

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;

try {

socket = new Socket("localhost", 8080);

out = new DataOutputStream(socket.getOutputStream());

// 请求服务器

String data1 = "牛顿";

byte[] outputBytes1 = data1.getBytes("UTF-8");

out.writeInt(outputBytes1.length); // write header

out.write(outputBytes1); // write body

String data2 = "爱因斯坦";

byte[] outputBytes2 = data2.getBytes("UTF-8");

out.writeInt(outputBytes2.length); // write header

out.write(outputBytes2); // write body

out.flush();

} finally {

// 关闭连接

out.close();

socket.close();

}

}

}

MINA服务器输出结果：

sessionCreated

messageReceived:牛顿

messageReceived:爱因斯坦

sessionClosed

Netty服务器输出结果：

channelActive

channelRead:牛顿

channelRead:爱因斯坦

channelInactive

Twisted服务器输出结果：

connectionMade

stringReceived:牛顿

stringReceived:爱因斯坦

connectionLost

感谢你能够认真阅读完这篇文章，希望小编分享的"怎么用MINA、Netty、Twisted来实现消息分割"这篇文章对大家有帮助，同时也希望大家多多支持，关注行业资讯频道，更多相关知识等着你来学习!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.