Kafka 및 Quarkus를 사용하여 장기간 프로세스를 병렬화하는 POC

poc-kafka-quarkus

이 프로젝트는 GitHub에서 사용할 수 있습니다.

POC는 Kafka 및 Quarkus를 사용하여 장시간 프로세스를 병렬화합니다.

처리할 항목이 많고 각 항목을 완료하는 데 오랜 시간이 걸린다고 상상해 보십시오.

이 POC는 전체 프로세스 시간을 줄이는 하나의 솔루션을 제안합니다.

아이디어는 간단합니다. 하나의 애플리케이션인 producer 는 Kafka 주제에 메시지를 게시하고 다른 애플리케이션인 processor 는 이러한 메시지를 읽고 처리합니다.

메시지를 순서대로 처리할 필요가 없는 경우 SmallRye Reactive Message를 사용하여 스레드 풀을 생성하고 여러 메시지를 동시에 처리하도록 애플리케이션을 구성할 수 있습니다.

Kafka는 또한 애플리케이션의 여러 인스턴스에서 해당 주제의 메시지를 사용할 수 있도록 합니다. 이를 달성하려면 토픽의 파티션 번호를 구성해야 합니다. 예를 들어 주제의 파티션 번호가 3이고 애플리케이션의 인스턴스 4개를 실행하는 경우 3개의 인스턴스가 특정 파티션 하나를 처리할 수 있고 네 번째 인스턴스는 유휴 상태로 유지됩니다.

이 POC에서는 두 가지 아이디어를 모두 사용합니다. 즉, 여러 인스턴스에서 처리 애플리케이션을 실행하고 각 인스턴스에는 여러 메시지를 동시에 처리하는 자체 스레드 풀이 있습니다.

활동 중인 POC

run 응용 프로그램만 Swagger-UI에 액세스하여 처리할 항목의 수량을 알리고 실행한 후 producer 및 processor 인스턴스의 콘솔 로그에서 결과를 확인합니다.

멋진

Swagger-UI가 켜져 있습니다http://localhost:8080/q/swagger-ui.

메시지를 보내는 프로듀서

메시지 수신 및 처리 프로세서 인스턴스

파티션 "2"가 있는 첫 번째 인스턴스



파티션 "0"및 "1"이 있는 두 번째 인스턴스

실행 지침

Docker Compose로 모든 것을 실행

두 프로젝트 컴파일

프로세서 폴더를 입력하고 다음을 실행하십시오.

$ ./mvnw package

생산자 폴더를 입력하고 다음을 실행합니다.

$ ./mvnw package

Docker Compose 실행

루트 폴더에서 다음을 실행합니다.

$ docker-compose up

processor 의 여러 인스턴스를 실행하려면 scale 인수를 전달하면 됩니다.

$ docker-compose up --scale processor=2

docker-compose 파일에 있는 Kafka의 NUM_PARTITIONS 매개변수는 토픽이 가질 파티션 수를 나타냅니다.

개발 모드에서 실행

생산자 및 프로세서 시작

프로세서 폴더를 입력하고 다음을 실행하십시오.

$ mvn quarkus:dev

생산자 폴더를 입력하고 다음을 실행합니다.

$ mvn quarkus:dev -Ddebug=5006

이 모드에서 Quarkus는 하나의 Kafka 이미지를 자동으로 다운로드하여 실행합니다.

카프카 개념

파티션

num.partitions 매개변수는 주제에 포함될 파티션 수를 정의합니다.

소비자 그룹

소비자 그룹은 하나의 ID로 정의됩니다.

하나의 소비자 그룹은 주제로 전송된 모든 메시지를 수신합니다.

하나의 소비자 그룹에는 'n'개의 응용 프로그램 인스턴스가 실행될 수 있습니다.

소비자 그룹의 각 인스턴스는 주제의 일부 부분에서 메시지를 처리합니다.

소비자 그룹 및 파티션

소비자 그룹의 각 인스턴스는 주제의 하나 이상의 파티션을 읽을 수 있습니다.

소비자가 파티션보다 많으면 일부 소비자는 유휴 상태로 유지됩니다.

Quarkus 및 SmallRye 반응성 메시지 팁

여러 작업자 스레드에서 메시지 처리

메시지를 순서대로 처리할 필요가 없는 경우 다음 주석을 사용할 수 있습니다.

@Incoming("extraction-requests")
@Blocking(ordered = false, value = "my-custom-pool")
public void read(Client client) {

application.properties에서 스레드 수my-custom-pool를 알려줍니다.

smallrye.messaging.worker.my-custom-pool.max-concurrency=3

Quarkus 개발 모드

주제 이름(이 경우 추출 요청)을 알리는 이 매개변수를 사용하여 주제가 Kafka 테스트 컨테이너에 포함할 파티션 수를 알릴 수 있습니다.

quarkus.kafka.devservices.topic-partitions.extraction-requests=3

학점

https://quarkus.io/guides/kafka

https://strimzi.io/docs/operators/latest/using.html

작은 라이

https://smallrye.io/smallrye-reactive-messaging/smallrye-reactive-messaging/3.13/index.html

https://smallrye.io/smallrye-reactive-messaging/smallrye-reactive-messaging/3.1/advanced/blocking.html

https://smallrye.io/smallrye-reactive-messaging/smallrye-reactive-messaging/3.1/emitter/emitter.html#_emitter_and_channel