- iOS/Objective-C 元类和类别
- objective-c - -1001 错误,当 NSURLSession 通过 httpproxy 和/etc/hosts
- java - 使用网络类获取 url 地址
- ios - 推送通知中不播放声音
我看到很多关于 netty 中的分块流的问题,但其中大部分是关于出站流的解决方案,而不是入站流。
我想了解如何从 channel 获取数据并将其作为 InputStream 发送到我的业务逻辑,而无需先将所有数据加载到内存中。这是我尝试做的事情:
public class ServerRequestHandler extends MessageToMessageDecoder<HttpObject> {
private HttpServletRequest request;
private PipedOutputStream os;
private PipedInputStream is;
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
super.handlerAdded(ctx);
this.os = new PipedOutputStream();
this.is = new PipedInputStream(os);
}
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
super.handlerRemoved(ctx);
this.os.close();
this.is.close();
}
@Override
protected void decode(ChannelHandlerContext ctx, HttpObject msg, List<Object> out)
throws Exception {
if (msg instanceof HttpRequest) {
this.request = new CustomHttpRequest((HttpRequest) msg, this.is);
out.add(this.request);
}
if (msg instanceof HttpContent) {
ByteBuf body = ((HttpContent) msg).content();
if (body.readableBytes() > 0)
body.readBytes(os, body.readableBytes());
if (msg instanceof LastHttpContent) {
os.close();
}
}
}
}
然后我有另一个处理程序,它将获取我的 CustomHttpRequest 并发送到我称之为 ServiceHandler 的地方,我的业务逻辑将从 InputStream 中读取。
public class ServiceRouterHandler extends SimpleChannelInboundHandler<CustomHttpRequest> {
...
@Override
public void channelRead0(ChannelHandlerContext ctx, CustomHttpRequest request) throws IOException {
...
future = serviceHandler.handle(request, response);
...
这是行不通的,因为当我的处理程序将 CustomHttpRequest 转发到 ServiceHandler 并尝试从 InputStream 读取时,线程处于阻塞状态,并且 HttpContent 从未在我的解码器中处理。
我知道我可以尝试为我的业务逻辑创建一个单独的线程,但我觉得我在这里让事情过于复杂了。
我查看了 ByteBufInputStream,但它是这样说的
Please note that it only reads up to the number of readable bytes determined at the moment of construction.
所以我认为它不适用于 Chunked Http 请求。另外,我看到了 ChunkedWriteHandler,这对于 Oubound block 来说似乎很好,但我找不到像 ChunkedReadHandler 这样的东西......
所以我的问题是:最好的方法是什么?我的要求是:
- 在发送 ServiceHandlers 之前不要将数据保存在内存中;
- ServiceHandlers API 应该与 netty 无关(这就是为什么我使用我的 CustomHttpRequest,而不是 Netty 的 HttpRequest);
更新我已经在 CustomHttpRequest 上使用更具 react 性的方法来实现它。现在,请求不向 ServiceHandlers 提供 InputStream 以便它们可以读取(这是阻塞的),而是 CustomHttpRequest 现在有一个返回 Future 的 readInto(OutputStream)
方法,所有的服务处理程序将在这个输出流被填满时执行。这是它的样子
public class CustomHttpRequest {
...constructors and other methods hidden...
private final SettableFuture<Void> writeCompleteFuture = SettableFuture.create();
private final SettableFuture<OutputStream> outputStreamFuture = SettableFuture.create();
private ListenableFuture<Void> lastWriteFuture = Futures.transform(outputStreamFuture, x-> null);
public ListenableFuture<Void> readInto(OutputStream os) throws IOException {
outputStreamFuture.set(os);
return this.writeCompleteFuture;
}
ListenableFuture<Void> writeChunk(byte[] buf) {
this.lastWriteFuture = Futures.transform(lastWriteFuture, (AsyncFunction<Void, Void>) (os) -> {
outputStreamFuture.get().write(buf);
return Futures.immediateFuture(null);
});
return lastWriteFuture;
}
void complete() {
ListenableFuture<Void> future =
Futures.transform(lastWriteFuture, (AsyncFunction<Void, Void>) x -> {
outputStreamFuture.get().close();
return Futures.immediateFuture(null);
});
addFinallyCallback(future, () -> {
this.writeCompleteFuture.set(null);
});
}
}
我更新后的 ServletRequestHandler 看起来像这样:
public class ServerRequestHandler extends MessageToMessageDecoder<HttpObject> {
private NettyHttpServletRequestAdaptor request;
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
super.handlerAdded(ctx);
}
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
super.handlerRemoved(ctx);
}
@Override
protected void decode(ChannelHandlerContext ctx, HttpObject msg, List<Object> out)
throws Exception {
if (msg instanceof HttpRequest) {
HttpRequest request = (HttpRequest) msg;
this.request = new CustomHttpRequest(request, ctx.channel());
out.add(this.request);
}
if (msg instanceof HttpContent) {
ByteBuf buf = ((HttpContent) msg).content();
byte[] bytes = new byte[buf.readableBytes()];
buf.readBytes(bytes);
this.request.writeChunk(bytes);
if (msg instanceof LastHttpContent) {
this.request.complete();
}
}
}
}
这工作得很好,但仍然请注意,这里的所有内容都是在单个线程中完成的,也许对于大数据,我可能想要生成一个新线程来为其他 channel 释放该线程。
最佳答案
您走在正确的轨道上 - 如果您的 serviceHandler.handle(request, response);
调用正在进行阻塞读取,您需要为其创建一个新线程。请记住,应该只有少量的 Netty 工作线程,因此您不应该在工作线程中进行任何阻塞调用。
另一个要问的问题是,您的服务处理程序需要阻塞吗?它有什么作用?如果它无论如何都在通过网络铲除数据,你能以非阻塞的方式将它合并到 Netty 管道中吗?这样,一切都是异步的,不需要阻塞调用和额外线程。
关于java - Netty 分块输入流,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/48668959/
我正在使用 WCF 并希望将大文件从客户端上传到服务器。我已经调查并决定遵循 http://msdn.microsoft.com/en-us/library/aa717050.aspx 中概述的分块方
我试图了解有关 Transfer-Encoding:chunked 的更多信息。引用了一些文章: http://zoompf.com/blog/2012/05/too-chunky和 "Transfe
我们正在评估 HDF5 在分块数据集方面的性能。 特别是我们试图弄清楚是否可以跨不同的连续块进行读取以及这样做会如何影响性能? 例如。我们有一个块大小为 10 的数据集,一个有 100 个值的数据集,
使用 Eloquent,如何根据 chunk 中的条件终止分块函数的关闭?我试过返回,但这似乎只终止当前块而不是所有块。此时,我想停止从数据库中检索记录。 $query->chunk(self::CH
有没有办法在不删除所选文件的情况下重新启动 plupload 上传? plupload.stop() 停止上传,但如果我使用 start() 再次启动上传,它会从上次停止的地方继续。相反,我希望它再次
我有带有“id,名称”的文件1和带有“id,地址”的文件2。我无法加载第一个文件(小于 2Gb):它在 76k 行(带有 block 连接)和只有 2 列后崩溃...我也无法在第二个文件上 read_
我正在尝试从头开始设计一个系统,我想在其中通过 servlet 加载文本行。生产线的生产需要一些时间。因此,我希望能够在它们到达时在我的浏览器中逐步显示它们,一次显示几个。我想从 javascript
能否请您提供一个示例,说明如何在 Android 中读取来自 Web 服务的分块响应 谢谢 编辑:我尝试调用一个 soap 网络服务,它用代表图像的 base64 编码字符串回复我 代码如下: Str
我想制作一个无限平铺 map ,从(-max_int,-max_int)到(max_int,max_int),所以我要制作一个基本结构: chunk,每个 chunk 包含 char tiles[w]
这是一个典型的场景:评估一个页面,并且有一个缓冲区 - 一旦缓冲区已满,评估的页面部分就会发送到浏览器。这使用 HTTP 1.1 分块编码。 但是,其中一个 block 中可能会发生错误(在第一个 b
如何从给定模式的句子中获取所有 block 。例子 NP:{} 标记的句子: [("money", "NN"), ("market", "NN") ("fund", "NN")] 如果我解析我得到 (
我正在使用以下代码将 CSV 文件拆分为多个 block (来自 here) def worker(chunk): print len(chunk) def keyfunc(row):
我想我已经很接近这个了,我有以下 dropzone 配置: Dropzone.options.myDZ = { chunking: true, chunkSize: 500000, ret
因为我在更常规的基础上使用 WebSocket 连接,所以我对事情在幕后的工作方式很感兴趣。因此,我研究了无休止的规范文档一段时间,但到目前为止,我真的找不到任何关于对传输流本身进行分 block 。
我有一个 slice ,其中包含约 210 万个日志字符串,我想创建一个 slice ,字符串尽可能均匀分布。 这是我目前所拥有的: // logs is a slice with ~2.1 mill
问题: 我有一个大约为 [350000, 1] 的向量,我希望计算成对距离。这导致 [350000, 350000] 整数数据类型的矩阵不适合 RAM。我最终想得到一个 bool 值(适合 RAM),
我想将 JSONP 用于具有 x 域脚本编写的项目,但不太关心 IE 中的 2048 个字符限制。 如果字符大小超过 2048,JSONP 是否自动支持“分块”?如果是的话,有人可以分享一些例子吗?
我目前正在开发 2d 角色扮演游戏,例如《最终幻想 1-4》。基本上,我的平铺 map 可以加载, Sprite 可以在 map 上自由行走。 如何处理与平铺 map 的碰撞? 我创建了三个独立的图
Treetagger 可以进行词性标记和文本分块,这意味着提取口头和名词性从句,如这个德语示例所示: $ echo 'Das ist ein Test.' | cmd/tagger-chunker-g
我应该从服务器流式传输端点,该端点返回带有传输编码的 json:分块。 我有以下代码,但无法读取响应。我尝试了 responseBody.streamBytes() 并将输入流转换为字符串,但我不能在
我是一名优秀的程序员,十分优秀!