【发布时间】：2023-02-11 05:07:00
【问题描述】：
请问关于 Reactor Kafka 的小问题。
我有一个非常简单的 Reactor Kafka 项目。
package com.example.micrometer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.function.Consumer;
@SpringBootApplication
public class StreamReactiveConsumerApplication implements CommandLineRunner {
private static final Logger log = LoggerFactory.getLogger(StreamReactiveConsumerApplication.class);
public static void main(String... args) {
new SpringApplicationBuilder(StreamReactiveConsumerApplication.class).run(args);
}
@Override
public void run(String... args) {
}
@Bean
Consumer<Flux<Message<String>>> consume() {
return flux -> flux.flatMap(one -> myHandle(one) ).subscribe();
}
private Mono<String> myHandle(Message<String> one) {
log.info("<==== look at this thread" + "\u001B[32m" + one.getPayload() + "\u001B[0m");
String payload = one.getPayload();
String decryptedPayload = complexInMemoryDecryption(payload); //this is NON blocking, takes 1 second
String complexMatrix = convertDecryptedPayloadToGiantMatrix(decryptedPayload); //this is NON blocking, takes 1 second
String newMatrix = matrixComputation(complexMatrix); //this is NON blocking, takes 1 second
return myNonBlockingReactiveRepository.save(complexMatrix);
}
}
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.example</groupId>
<artifactId>streamreactiveconsumer</artifactId>
<version>1.0-SNAPSHOT</version>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>3.0.2</version>
<relativePath/>
</parent>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>2022.0.1</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
（注意，不是Spring Kafka项目，不是Spring Cloud Stream项目）
我正在使用一个包含 3 个分区的主题。发送消息的速率是每秒一条消息。
消息的消费和处理每条消息需要 3 秒左右的时间。
重要提示：请注意该处理不包含任何阻塞操作。它是内存解密+巨型矩阵计算的巨人。它是 BlockHound 测试的非阻塞。
实际的:
当我使用 Reactor Kafka 项目使用消息时，整个使用只发生在一个线程上。一切都发生在container-0-C-1
一切都会发生在container-0-C-1，用2个CPU、4个CPU、8个CPU的硬件测试
2023-02-06 10:42:59 8384 INFO --- [KafkaConsumerDestination{consumerDestinationName='prod_audit_hdfs', partitions=3, dlqName='null'}.container-0-C-1] [stream-reactive-consumer,,] c.e.m.StreamReactiveConsumerApplication :
2023-02-06 10:42:59 8384 INFO --- [KafkaConsumerDestination{consumerDestinationName='prod_audit_hdfs', partitions=3, dlqName='null'}.container-0-C-1] [stream-reactive-consumer,,] c.e.m.StreamReactiveConsumerApplication :
2023-02-06 10:42:59 8384 INFO --- [KafkaConsumerDestination{consumerDestinationName='prod_audit_hdfs', partitions=3, dlqName='null'}.container-0-C-1] [stream-reactive-consumer,,] c.e.m.StreamReactiveConsumerApplication :
预期的: 我们从基于 http webflux 迁移到基于 Kafka 消费。业务逻辑没有一点改变。
在 Reactor Netty Spring webflux 应用程序上，我们可以看到处理是从多个线程发生的，对应于反应堆核心。在具有多个内核的机器上，这很容易跟上。
[or-http-epoll-1] [or-http-epoll-2] [or-http-epoll-3] [or-http-epoll-4]
只需在这些 reactor-http-epoll-N 中的任何一个之间切换即可进行处理。 我可以看到 reactor-http-epoll-1 何时处理第一条消息的复杂内存计算，reactor-http-epoll-3 将处理第二条消息的计算，等等......并行性很明显
我知道有办法“扩展”这个应用程序，但这是 Reactor Kafka 本身的问题。
我希望这些消息可以并行处理。第一条消息使用某种 container-0-C-1，第二条消息使用 container-0-C-2，等等...
请问我怎样才能做到这一点？ 我错过了什么？
谢谢
【问题讨论】：
标签： java spring-boot apache-kafka parallel-processing reactor-kafka