Kafka流处理器线程安全？

Question

我知道这里之前有人问过这个问题：Kafka流并发？

但这对我来说很奇怪。根据文档（或者我可能遗漏了什么），每个分区都有一个任务，这意味着不同的处理器实例，每个任务由不同的线程执行。但是当我测试它的时候，我看到不同的线程可以得到不同的处理器实例。因此，如果你想在处理器中保持内存状态（老式的方式），你必须锁定？

public class SomeProcessor extends AbstractProcessor<String, JsonObject> {

   private final String ID = UUID.randomUUID().toString();

   @Override
   public void process(String key, JsonObject value) {
     System.out.println("Thread id: " + Thread.currentThread().getId() +" ID: " + ID);

线程ID:88 ID:c667e669-9023-494b-9345-236777e9dfda

线程ID:88 ID:c667e669-9023-494b-9345-236777e9dfda

线程ID:90 ID:0A43ECB0-26F2-440D-88E2-87E0C9CC4927

Answer 1

每个实例的线程数是一个配置参数(num.stream.threads，默认值为1)。因此，如果启动一个KafKastreams实例，就会得到num.stream.threads线程。

任务将工作分成并行单元（基于输入主题分区），并分配给线程。因此，如果您有多个任务和一个线程，所有任务都将分配给这个线程。如果您有两个线程(KafKastreams实例的总和），每个线程执行大约50%的任务。

注意：因为Kafka Streams应用程序本质上是分布式的，所以如果运行一个Kafkastreams实例和多个线程，或者运行多个Kafkastreams实例和一个线程，则没有区别。任务将分布在应用程序的所有可用线程上。

如果您希望在任务之间共享任何数据结构，并且您有一个以上的线程，那么您有责任同步对该数据结构的访问。请注意，任务到线程的分配可以在运行时更改，因此，所有访问都必须同步。但是，不建议使用此模式，因为它限制了可伸缩性。你应该用没有共享数据结构来设计你的程序！主要原因是，您的程序通常分布在多台机器上，因此不同的KafKastreams实例无论如何都无法访问共享数据结构。共享数据结构只能在单个JVM中工作，但是使用单个JVM可以防止应用程序的横向扩展。