Java 개발: 스트림 처리 및 데이터 전송을 위해 Akka Streams를 사용하는 방법

Java 개발: 스트림 처리 및 데이터 전송을 위해 Akka Streams를 사용하는 방법

소개:
빅 데이터 및 실시간 데이터 처리의 급속한 발전으로 스트림 처리 및 데이터 전송에 대한 수요가 계속 증가하고 있습니다. Java 개발에서 Akka Streams는 스트림 처리 및 데이터 전송 구현을 단순화하는 강력한 라이브러리입니다. 이 기사에서는 Akka Streams의 기본 개념과 사용법을 소개하고 자세한 코드 예제를 제공합니다.

1. Akka Streams 개요:
1.1 Akka Streams란 무엇입니까?
Akka Streams는 Akka 프레임워크의 일부이며 비동기적이고 구성 가능하며 모니터링 가능한 스트림 처리 모델을 제공합니다. 일관되지 않은 데이터 스트림 속도를 처리하기 위해 배압 메커니즘을 사용합니다. Akka Streams는 확장성과 유연성이 뛰어나 대규모 데이터 스트림을 쉽게 처리할 수 있습니다.

1.2 기본 개념:

• 소스: 파일, 데이터베이스, 네트워크 연결 등이 될 수 있는 데이터 흐름의 소스입니다. 소스는 0개 이상의 데이터 요소를 내보낼 수 있습니다.
• Flow: 필터링, 매핑, 집계 등 데이터 흐름을 작동하고 변환하는 구성 요소입니다. 흐름은 하나 이상의 데이터 요소를 수신하고 하나 이상의 데이터 요소를 출력할 수 있습니다.
• 싱크: 파일, 데이터베이스, 네트워크 연결 등이 될 수 있는 데이터 흐름의 끝점입니다. 엔드포인트는 Flow가 처리한 데이터를 수신하여 처리합니다.

2. Akka Streams 사용:
2.1 종속성 소개:
먼저 Akka Streams의 종속성을 Java 프로젝트에 도입해야 합니다. pom.xml 파일에 다음 종속성을 추가합니다.

<dependency>
    <groupId>com.typesafe.akka</groupId>
    <artifactId>akka-stream_2.12</artifactId>
    <version>2.6.17</version>
</dependency>

2.2 간단한 스트림 처리 구현:
아래에서는 간단한 예를 사용하여 스트림 처리에 Akka Streams를 사용하는 방법을 보여줍니다.

먼저 정수가 포함된 데이터 소스를 만듭니다.

Source<Integer, NotUsed> source = Source.range(1, 10);

그런 다음 소스 데이터에 2를 곱하는 흐름을 만듭니다.

Flow<Integer, Integer, NotUsed> flow = Flow.of(Integer.class).map(i -> i * 2);

다음으로 스트림 처리된 데이터를 수신할 싱크를 만듭니다.

Sink<Integer, CompletionStage<Done>> sink = Sink.foreach(System.out::println);

소스 배치, Flow와 Sink를 결합하여 완전한 스트림 처리를 구축합니다.

RunnableGraph<NotUsed> runnableGraph = source.via(flow).to(sink);

마지막으로 스트림 처리를 실행합니다.

CompletionStage<NotUsed> completionStage = runnableGraph.run(materializer);

위 코드에서는 Akka Streams에서 제공하는 다양한 구성 요소를 사용하여 데이터 소스, Flow 및 Sink를 포함한 간단한 스트림 처리를 구현합니다. 싱크대. 이러한 구성 요소를 연결함으로써 완전한 스트림 처리 프로세스를 정의하고 실행할 수 있습니다.

2.3 데이터 전송 구현:
스트림 처리 외에도 Akka Streams를 데이터 전송에 사용할 수도 있습니다. 아래에서는 데이터 전송에 Akka Streams를 사용하는 방법을 보여주기 위해 TCP 전송을 예로 들어 보겠습니다.

먼저 서버 측 스트림 처리 생성:

final Flow<ByteString, ByteString, NotUsed> serverFlow = Flow.of(ByteString.class)
    .via(Tcp().delimiter(ByteString.fromString("
"), 256, true))
    .map(ByteString::utf8String)
    .map(s -> s + " processed")
    .map(ByteString::fromString);

그런 다음 서버 시작:

final Source<Tcp.IncomingConnection, CompletionStage<Tcp.ServerBinding>> serverSource =
    Tcp().bind("localhost", 8888);

final Flow<Tcp.IncomingConnection, Tcp.IncomingConnection, NotUsed> handler = Flow.<Tcp.IncomingConnection>create()
    .mapAsync(1, connection -> {
        connection.handleWith(serverFlow, materializer);
        return CompletableFuture.completedFuture(connection);
    });

final CompletionStage<Tcp.ServerBinding> binding =
    serverSource.via(handler).to(Sink.ignore()).run(materializer);

다음으로 클라이언트 측 스트림 처리 생성:

final Sink<ByteString, CompletionStage<Done>> clientSink = Sink.ignore();

final Flow<String, ByteString, CompletionStage<OutgoingConnection>> connectionFlow =
    Tcp().outgoingConnection("localhost", 8888);

final Flow<ByteString, ByteString, CompletionStage<Done>> clientFlow = Flow.of(ByteString.class)
    .via(Tcp().delimiter(ByteString.fromString("
"), 256, true))
    .map(ByteString::utf8String)
    .map(s -> s + " processed")
    .map(ByteString::fromString);

final Flow<String, ByteString, CompletionStage<Tcp.OutgoingConnection>> flow =
    Flow.fromSinkAndSourceMat(clientSink, clientFlow, Keep.right());

CompletableFuture<Tcp.OutgoingConnection> connection =
    Source.single("data").viaMat(connectionFlow, Keep.right()).toMat(flow, Keep.left()).run(materializer);

위 코드를 사용하여 서버 측 스트림 처리를 생성합니다. 클라이언트 스트림 처리 및 TCP를 통한 데이터 전송이 가능합니다. 서버 측 스트림 처리에서는 수신된 문자열을 처리하여 클라이언트로 보냅니다. 클라이언트 측 스트림 처리에서는 수신된 문자열을 처리하여 서버로 보냅니다.

요약:
이 글에서는 Akka Streams의 기본 개념과 사용법을 소개하고 자세한 코드 예제를 제공합니다. Akka Streams를 통해 스트림 처리 및 데이터 전송을 쉽게 구현하여 데이터 처리의 효율성과 성능을 향상시킬 수 있습니다. 이 기사가 Java 개발에서 스트림 처리 및 데이터 전송에 Akka Streams를 사용하는 데 도움이 되기를 바랍니다.

위 내용은 Java 개발: 스트림 처리 및 데이터 전송을 위해 Akka Streams를 사용하는 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!