我已经实现了一个服务器应用程序，它使用netty框架使用ChannelInblundHandlerAdapter读取传入字节。
标题中显示的问题是，我不定期地从客户端获取内容，我认为这些内容在~1.500字节后被切断。例如：在这种情况下，我应该收到一个大的 JSON 数组。因为它被切割了，所以我无法解析它。
在使用消息之前，我尝试使用管道中的附加 ByteToMessageDecoder 通道对消息进行解码。但这并不能解决问题。我在 JSON 中没有分隔符，我可以检查并再次将两个(或更多)部分粘在一起。
这是我的管道配置：
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) {
ch.pipeline().addLast(new IdleStateHandler(45,0,0));
ch.pipeline().addLast(new MyByteToMessageDecoder());
ch.pipeline().addLast(new GatewayCommunicationHandler());
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.option(ChannelOption.SO_RCVBUF, 8192)
.childOption(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(8192))
.childOption(ChannelOption.SO_KEEPALIVE, true);
initRestServer();
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(Config.gatewayPort).sync();
f.channel().closeFuture().sync();
这是我的ByteToMessageDecoder：(我知道这是一团糟，但我不知道如何处理它)
public class MyByteToMessageDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
byte[] receivedBytes = new byte[in.readableBytes()];
in.getBytes(in.readerIndex(), receivedBytes);
if (receivedBytes[in.readableBytes()-1] != (byte) 0) {
out.add(receivedBytes);
return;
}
int lenForOutBytes = 0;
for (Object o : out) {
byte[] bytes = (byte[]) o;
lenForOutBytes += bytes.length;
}
byte[] outBytes = new byte[lenForOutBytes];
for (Object o : out) {
byte[] bytes = (byte[]) o;
if (out.size() == 1) {
outBytes = (byte[]) out.get(0);
}
else {
int i = 0;
for (int j = 0; j < bytes.length; j++) {
outBytes[i + j] = bytes[j];
}
i += bytes.length;
}
}
ctx.fireChannelRead(outBytes);
in.resetReaderIndex();
}
...
还有其他人有这样的问题吗？
感谢您的回复
布乔
我已经看到这个问题经常发生，所以我故意比平时更广泛一些
发生此问题的原因是 TCP 是基于流的，而不是基于数据包的。
这基本上发生：
- [客户端] 想要发送 10k 字节的数据
- [客户端] 将数据发送到 TCP 层
- [客户端] TCP 层拆分数据包，它知道最大数据包大小为 1500(这是几乎所有网络使用的默认 MTU)
- [client] 客户端将数据包发送到包含 40 字节作为标头和 1460 字节作为数据的服务器
- 【server】Netty接收第一个数据包数据，并直接调用你的函数，第一个数据包包含1460字节的数据
- [服务器] 在函数需要处理剩余数据的时间(初始数据 - 1260)
所以解决这个问题，有多种方法
在消息前面加上长度：
虽然这通常是解决数据包的最简单方法，但在同时处理小型和大型消息时，它也是效率最低的方法。这也需要更改协议。
基本思想是在发送数据包之前预先添加长度，这允许您正确拆分消息
优势
- 无需循环数据来过滤掉字符或阻止禁止的字符
- 如果您的网络中有中继系统，则它们不必对消息边界进行任何硬解析
弊
消息- 的长度必须知道优势，在长消息中，这是内存扩展
如何？
如果您使用标准的整数字段，这非常简单，因为 Netty 为此构建了类：
LengthFieldBasedFrameDecoderLengthFieldPrepender
在管道中以以下方式使用它
// int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip
pipeline.addLast(new LengthFieldBasedFrameDecoder(1024 * 4, 0, 2, 0, 2));
// int lengthFieldLength, int lengthAdjustment
pipeline.addLast(new LengthFieldPrepender(2, 0));
这基本上是像下面这样构建数据包：
您发送：
DATA: 12B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 65 6c 6c 6f 20 57 6f 72 6c 64 21 |Hello World! |
+--------+-------------------------------------------------+----------------+
LengthFieldPrepender将其转换为：
DATA: 14B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 0c 48 65 6c 6c 6f 20 57 6f 72 6c 64 21 |..Hello World! |
+--------+-------------------------------------------------+----------------+
然后，当您收到消息时，LengthFieldBasedFrameDecoder将其解码为：
DATA: 12B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 65 6c 6c 6f 20 57 6f 72 6c 64 21 |Hello World! |
+--------+-------------------------------------------------+----------------+
在简单分隔符上拆分消息
一些协议采用不同的方法，而不是在固定长度上拆分，而是在分隔符上拆分。一种快速的查看方法是，Java 中的字符串以"结尾，文本文件中的行以换行符结尾，自然文本中的段落以双换行符结尾。
优势
- 如果您知道某个数据不包含字符，则相对容易生成，例如 JSON 通常不包含空格，因此用空格分隔消息很容易。
- 易于从脚本语言实现，因为不需要状态
弊
- 与框架字符冲突可能会使邮件大小膨胀
- 长度事先是未知的，所以要么在代码中设置硬编码限制，要么继续读取，直到内存不足或数据结束
- 即使您对数据包不感兴趣，也需要读取每个字符
如何？
从 Netty 发送消息时，您需要手动将分隔符添加到消息本身，接收时可以使用DelimiterBasedFrameDecoder将传入流解码为消息。
示例管道：
在管道中以以下方式使用它
// int maxFrameLength, ByteBuf... delimiters
pipeline.addLast(1024 * 4, DelimiterBasedFrameDecoder(Delimiters.lineDelimiter()));
发送消息时，您需要手动添加分隔符：
DATA: 14B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 65 6c 6c 6f 20 57 6f 72 6c 64 21 0d 0a |Hello World!.. |
+--------+-------------------------------------------------+----------------+
接收消息时，DelimiterBasedFrameDecoder会将消息转换为框架：
DATA: 12B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 65 6c 6c 6f 20 57 6f 72 6c 64 21 |Hello World! |
+--------+-------------------------------------------------+----------------+
拆分复杂的业务分隔符
并非所有的框架都很容易，如果避免某些解决方案实际上是最好的，但有时，您确实需要做一些肮脏的工作。
优势
- 几乎可以处理所有现有的数据结构
- 无需修改实验方案
弊
- 通常你必须检查每个字节
- 代码可能难以遵循
- 快速解决方案可能会在它认为格式错误的输入中给出奇怪的错误
这分为两类：
- 基于现有解码器
- 模式检测
基于现有解码器
使用这些解决方案，您基本上使用其他框架中的现有解码器来解析数据包，并检测其处理中的故障。
GSON 和ReplayingDecoder的示例：
public class GSONDecoder
extends ReplayingDecoder<Void> {
Gson gson = new GsonBuilder().create();
protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out)
throws Exception {
out.add(gson.fromJson(new ByteBufInputStream(buf, false), Object.class));
}
}
模式检测
如果要使用模式检测方法，则需要了解协议。让我们为 JSON 制作一个模式检测解码器。
基于JSON的结构，我们做以下假设：
- JSON基于匹配的
{和}对，以及[和] {和}的匹配对应忽略""在前面加上时应忽略- 如果
前面加上，则从左到右解析时应忽略
基于这些属性，让我们根据以下假设进行ByteToMessageDecoder：
public static class JSONDecoder extends ByteToMessageDecoder {
// Notice, this class is designed for JSON without a charset definition at the start, adding this is hard as we basicly have to call differend
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
in.markReaderIndex();
int fromIndex = in.readerIndex();
int unclosedCurlyBracketsSeen = 0;
boolean inQuotedSection = false;
boolean nonWhitespaceSeen = false;
boolean slashSeen = false;
while (in.isReadable()) {
boolean newSlashSeenState = false;
byte character = in.readByte();
if (character == '{' && !inQuotedSection) {
unclosedCurlyBracketsSeen++;
}
if (character == '}' && !inQuotedSection) {
unclosedCurlyBracketsSeen--;
}
if (character == '[' && !inQuotedSection) {
unclosedCurlyBracketsSeen++;
}
if (character == ']' && !inQuotedSection) {
unclosedCurlyBracketsSeen--;
}
if (character == '"' && !slashSeen) {
inQuotedSection = !inQuotedSection;
}
if (character == '\' && !slashSeen) {
newSlashSeenState = true;
}
if (!Character.isWhitespace(character)) {
nonWhitespaceSeen = true;
}
slashSeen = newSlashSeenState;
if(unclosedCurlyBracketsSeen == 0 && nonWhitespaceSeen) {
int targetIndex = in.readerIndex();
out.add(in.slice(fromIndex, targetIndex - fromIndex).retain());
return;
}
}
// End of stream reached, but our JSON is not complete, reset our progress!
in.resetReaderIndex();
}
}
接收消息时，它是这样工作的：
DATA: 35B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 7b 22 68 69 21 22 2c 22 53 74 72 69 6e 67 3a 20 |{"hi!","String: |
|00000010| 5c 22 48 69 5c 22 22 7d 20 20 7b 22 73 6c 61 73 |"Hi""} {"slas|
|00000020| 68 22 3a |h": |
+--------+-------------------------------------------------+----------------+
DATA: 34B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 22 5c 5c 22 7d 7b 22 4e 65 73 74 65 64 3a 22 3a |"\"}{"Nested:":|
|00000010| 7b 22 64 65 65 70 65 72 22 3a 7b 22 6f 6b 22 7d |{"deeper":{"ok"}|
|00000020| 7d 7d |}} |
+--------+-------------------------------------------------+----------------+
如您所见，我们收到了 2 条消息，其中 1 条甚至被分成 2 个"虚拟 TCP"数据包，这由我们的"JSON 解码器"转换为以下 ByteBuf 数据包：
DATA: 24B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 7b 22 68 69 21 22 2c 22 53 74 72 69 6e 67 3a 20 |{"hi!","String: |
|00000010| 5c 22 48 69 5c 22 22 7d |"Hi""} |
+--------+-------------------------------------------------+----------------+
DATA: 16B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 20 20 7b 22 73 6c 61 73 68 22 3a 22 5c 5c 22 7d | {"slash":"\"}|
+--------+-------------------------------------------------+----------------+
DATA: 29B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 7b 22 4e 65 73 74 65 64 3a 22 3a 7b 22 64 65 65 |{"Nested:":{"dee|
|00000010| 70 65 72 22 3a 7b 22 6f 6b 22 7d 7d 7d |per":{"ok"}}} |
+--------+-------------------------------------------------+----------------+