使用Netty/Ratpack中的json流数据

Question

我想使用Netty或Ratpack使用json数据流。我的用例是请求主体将包含大型json数据(MB中的json数组)。处理数据的一种方式是块，直到接收到完整的数据，然后开始处理。但是，我想要异步处理，这意味着一旦接收到一块json对象，就对其进行处理

我在Netty中遇到了JsonObjectDecoder，但使用它时运气不佳。这是我的ChannelInitializer类：

public class ServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
public void initChannel(SocketChannel ch) {
ChannelPipeline p = ch.pipeline();
p.addLast(new JsonObjectDecoder(true));
// HttpServerCodec is a combination of HttpRequestDecoder and HttpResponseEncoder
p.addLast(new HttpServerCodec());
//
// add gizp compressor for http response content
p.addLast(new HttpContentCompressor());
p.addLast(new HttpObjectAggregator(1048576));
p.addLast(new ChunkedWriteHandler());
p.addLast(new ServerHandler());
}
} 

我正在发送以下数据：

[
{
"timestamp": "2016-11-14 11:08:09+0100", 
"message": "message 120", 
"hostname": "myhost.com", 
"device_product": "product123", 
"device_vendor": "vendor123", 
"device_version": "1", 
"severity": "High"
},
.....
{
"timestamp": "2016-11-14 11:08:09+0100", 
"message": "message 121", 
"hostname": "myhost.com", 
"device_product": "product123", 
"device_vendor": "vendor123", 
"device_version": "1", 
"severity": "High"
}
]

但我得到了这个错误：

io.netty.handler.codec.CorruptedFrameException: invalid JSON received at byte position 0: 504f5354202f6c6f677320485454502f312e310d0a486f73743a206c6f63616c686f73743a383038300d0a436f6e6e656374696f6e3a206b6565702d616c6976650d0a436f6e74656e742d4c656e6774683a203230380d0a4163636570743a206170706c69636174696f6e2f6a736f6e0d0a506f73746d616e2d546f6b656e3a2062383064306264352d663234302d346563622d353631322d3863376139396434633934360d0a43616368652d436f6e74726f6c3a206e6f2d63616368650d0a4f726967696e3a206368726f6d652d657874656e73696f6e3a2f2f6668626a676269666c696e6a62646767656863646463626e636464646f6d6f700d0a557365722d4167656e743a204d6f7a696c6c612f352e30202857696e646f7773204e5420362e313b2057696e36343b2078363429204170706c655765624b69742f3533372e333620284b48544d4c2c206c696b65204765636b6f29204368726f6d652f35352e302e323838332e3837205361666172692f3533372e33360d0a436f6e74656e742d547970653a206170706c69636174696f6e2f6a736f6e0d0a4163636570742d456e636f64696e673a20677a69702c206465666c6174652c2062720d0a4163636570742d4c616e67756167653a20656e2d55532c656e3b713d302e382c6a613b713d302e362c66722d46523b713d302e342c66723b713d302e322c66722d43413b713d302e320d0a0d0a7b2274696d657374616d70223a2022323031362d31312d31342031313a30383a30392b30313030222c226d657373616765223a20226d65737361676520313230222c22686f73746e616d65223a20226d79686f73742e636f6d222c200a09226465766963655f70726f64756374223a202270726f64756374313233222c200a09226465766963655f76656e646f72223a202276656e646f72313233222c200a09226465766963655f76657273696f6e223a202231222c200a09227365766572697479223a202248696768220a090a097d
at io.netty.handler.codec.json.JsonObjectDecoder.decode(JsonObjectDecoder.java:163)
at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:316)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:230)
at io.netty.channel.ChannelHandlerInvokerUtil.invokeChannelReadNow(ChannelHandlerInvokerUtil.java:84)
at io.netty.channel.DefaultChannelHandlerInvoker.invokeChannelRead(DefaultChannelHandlerInvoker.java:153)
at io.netty.channel.PausableChannelEventExecutor.invokeChannelRead(PausableChannelEventExecutor.java:86)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:389)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:956)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:127)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:514)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:471)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:385)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:351)
at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:116)
at io.netty.util.internal.chmv8.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1412)
at io.netty.util.internal.chmv8.ForkJoinTask.doExec(ForkJoinTask.java:280)
at io.netty.util.internal.chmv8.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:877)
at io.netty.util.internal.chmv8.ForkJoinPool.scan(ForkJoinPool.java:1706)
at io.netty.util.internal.chmv8.ForkJoinPool.runWorker(ForkJoinPool.java:1661)
at io.netty.util.internal.chmv8.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:126)

我不知道我错过了什么。如果有人知道使用Ratpack实现这一目标的方法，请帮助我。提前谢谢。

Answer 1

问题是JSON解码器是管道中的第一个处理程序，它正试图解码HTTP帖子。如果我从您发布的错误消息中获取无效数据流，请将其解析回字节，并从中创建一个字符串(在groovy中)。。。

import javax.xml.bind.DatatypeConverter;
v = "504f5354202f6c6f677320485...<snip>";
byte[] bytes = DatatypeConverter.parseHexBinary(v);
println new String(bytes)

结果是：

POST/logs HTTP/1.1主机：localhost:8080连接：保持活动内容长度：208接受：application/json邮差令牌：<代币已移除>缓存控制：无缓存原点：镀铬扩展：//<ID已删除>用户代理：Mozilla/5.0(Windows NT 6.1；Win64；x64)AppleWebKit/537.36(KHTML，类似Gecko)Chrome浏览器/55.0.2883.87 Safari浏览器/537.36内容类型：application/json接受编码：gzip，deflate，br接受语言：en-US，en；q=0.8，ja；q=0.6，fr-fr；q=0.4，fr；q=0.2，fr-CA；q=0.2

｛"时间戳"："2016-11-14 11:08:09+0100"，"消息"："消息120"，"hostname"："myhost.com"，"device_product"："product123"，"device_vendor"："vendor123"，"device_version"："1"，"严重性"："高"｝

因此，您需要在JSON解码器之前将这些添加到管道中：

1. HttpServerCodec
2. HttpObjectAggregator(对于大型帖子，可以对数据进行分块)
3. MessageToMessageDecodeee接受[Full]HttpRequest并转发内容(作为ByteBuf)

然后JSON解码器将获得一块JSON字节，并开始向上游发送解析后的消息。

Answer 2

要在HTTPPOST中执行此操作，您需要确保请求被分块。这是你需要对你的管道做什么的近似值：

1. HttpServerCodec-将转发HttpContent的实例，除了第一个将是HttpRequest的实例
2. MessageToMessageDecoder接受HttpContent实例，提取内容ByteBuf并转发
3. JSON解码器
4. 您的JSON处理程序

在某个时刻，您将获得一个HttpContent，它也是LastHttpContent的一个实例，它将是最后一个块。

棘手的部分是，在某个时刻，其中一个HttpContents将有一个不完整的JSON序列，这将在JSON解码器中触发错误，此时您需要将ByteBuf倒带到最后一个已知的"好"位置，等待下一个块出现并完成它，因为我认为这不会自动处理。