243장. Netty4 HTTP 구성 요소

Camel 버전 2.14로 사용 가능

netty4-http 구성 요소는 Netty4 를 사용하여 HTTP 전송을 능가하는 Netty4 구성 요소의 확장입니다.

이 camel 구성 요소는 생산자 및 소비자 엔드 포인트를 모두 지원합니다.

정보: 스트림. Netty는 스트림 기반이며, 이는 수신하는 입력이 Camel에 스트림으로 제출됨을 의미합니다. 즉, 스트림의 내용을 한 번 만 읽을 수 있습니다. 메시지 본문이 비어 있거나 데이터(예: 멀티 캐스트 작업 또는 재전송 오류 처리 수행)에 액세스해야 하는 경우 Stream 캐싱을 사용하거나 메시지 본문을 여러 번 안전하게 다시 읽을 수 있는 문자열로 변환해야 합니다. 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: InputStream. InputStream 을 메시지 본문으로 사용하고 큰 데이터 스트림(예: > 2GB)을 쓰거나 읽으려는 경우 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]