第 243 章 Netty4 HTTP Component

作为 Camel 版本 2.14 可用

netty4-http 组件是对 Netty4 组件的一个扩展，通过 Netty4 分离 HTTP 传输。

此 camel 组件支持制作者和消费者端点。

INFO： 流.Netty 基于流，这意味着其接收的输入被提交到 Camel 作为流。这意味着，一次只能读取流的内容 。如果发现消息正文显示为空的情况，或者您需要多次访问数据（例如：执行多播或重新传送错误处理），您应该使用 Stream 缓存或将消息正文转换为可安全地重新读取多次的 String。注意 Netty4 HTTP 使用 io.netty.handler.codec.http.HttpObjectAggregator 将整个流读取到内存中，以构建完整的 http 消息。但是生成的消息仍是一个基于流的消息，它可以被读取一次。

Maven 用户需要将以下依赖项添加到其 pom.xml 中：

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-netty4-http</artifactId>
    <version>x.x.x</version>
    <!-- use the same version as your Camel core version -->
</dependency>

INFO： 输入Stream.如果将 InputStream 用作消息正文，并且您想要写入或读取大型数据流（如 > 2 GB），则必须通过将 disableStreamCache 参数设置为 true 来使用流传输支持。

示例 1：将大型数据流上传到服务器

// Upload a large data stream to the server
from("direct:upstream-call")
        .bean(Helper.class, "prepareStream")
        .to("netty-http:http://localhost:{{port}}/upstream?disableStreamCache=true")
        .log("get ${body}");

// Read a large data stream from the client
from("netty-http:http://0.0.0.0:{{port}}/upstream?disableStreamCache=true")
        .bean(Helper.class, "processStream")
        .to("mock:stream-size");

示例 2：从服务器下载大型数据流

// Download a large data stream from the server
from("direct:download-call")
        .to("netty-http:http://localhost:{{port}}/downstream?disableStreamCache=true")
        .bean(Helper.class, "asyncProcessStream")
        .log("get ${body}");

// Write a large data stream to the client
from("netty-http:http://0.0.0.0:{{port}}/downstream?disableStreamCache=true")
        .bean(Helper.class, "prepareStream");

在下载示例中，您必须从其他线程的 InputStream 中读取以避免阻断底层流处理程序（请参阅 asyncProcessStream）。

    public static void processStream(Exchange exchange) throws Exception {
        InputStream is = exchange.getIn().getBody(InputStream.class);

        byte[] buffer = new byte[1024];
        long read = 0;
        long total = 0;
        while ((read = is.read(buffer, 0, buffer.length)) != -1) {
            total += read;
        }

        exchange.getIn().setBody(new Long(total));
    }

    public static CompletableFuture<Void> asyncProcessStream(Exchange exchange) {
        return CompletableFuture.runAsync(() -> {
            try {
                processStream(exchange);
            } catch (Exception e) {
                exchange.setException(e);
            }
        });
    }

netty4-http:http://0.0.0.0:8080[?options]