跨线程共享KafkaConsumer时分区赋值变更的可见性如何保证

首先必须明确：KafkaConsumer 设计上就不支持多线程并发操作，如果你在线程1调用assign()方法修改分区分配的同时，线程2正在执行poll()操作，会直接抛出ConcurrentModificationException异常，这个并发安全问题比可见性问题优先级更高。

答案是不能。
就算你把KafkaConsumer的实例引用声明为volatile，也只能保证这个引用本身的内存可见性，管不到KafkaConsumer内部维护的分区分配、offset等私有状态的可见性。更重要的是volatile完全解决不了多线程并发操作consumer的线程安全问题，并发调用的异常还是会出现。

你需要保证同一时间只有一个线程操作KafkaConsumer实例，业内最常用的是如下两种方案：

方案1：锁同步所有consumer操作（不推荐） #

所有调用KafkaConsumer方法的位置（包括线程1的assign()、线程2的poll()、commitSync()等）都加同一把互斥锁，保证同一时间只有一个线程操作consumer。加锁/解锁的Happens-Before规则可以保证前一个线程对consumer的所有修改，对后一个拿到锁的线程完全可见。但这种方案的缺点很明显：poll操作持锁时间长，会导致线程1的分区更新请求阻塞很久。

方案2：所有consumer操作都交给poll线程执行（推荐） #

用线程安全的队列传递分区变更请求，线程1只负责把新的分区配置提交到队列，线程2在两次poll的间隙检查队列，自行执行分区更新操作。这种方案完全规避了多线程操作consumer的风险，也天然解决了状态可见性问题。

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ConcurrentLinkedQueue;

public class MultiThreadConsumerDemo {
    // 线程安全队列，用来传递分区变更请求
    private final ConcurrentLinkedQueue<List<TopicPartition>> partitionUpdateQueue = new ConcurrentLinkedQueue<>();
    private final KafkaConsumer<String, String> consumer;

    public MultiThreadConsumerDemo() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        consumer = new KafkaConsumer<>(props);
    }

    // 线程1调用该方法提交新的分区分配规则
    public void changePartitionAssign(List<TopicPartition> newPartitions) {
        partitionUpdateQueue.offer(newPartitions);
    }

    // 线程2执行的持续poll逻辑
    public void runPollTask() {
        try {
            while (!Thread.currentThread().isInterrupted()) {
                // 先检查有没有待执行的分区更新请求
                List<TopicPartition> newAssign = partitionUpdateQueue.poll();
                if (newAssign != null) {
                    consumer.assign(newAssign);
                }

                // 拉取消息
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                // 业务处理逻辑省略，可根据需要自行扩展
                records.forEach(record -> {
                    System.out.printf("处理消息：offset=%d, key=%s, value=%s%n", record.offset(), record.key(), record.value());
                });

                // 手动提交offset
                consumer.commitSync();
            }
        } finally {
            consumer.close();
        }
    }
}

• 所有对KafkaConsumer的操作都只在poll线程执行，完全规避了多线程并发操作的风险，同一个线程对consumer的修改天然对自身可见，不需要额外的可见性控制。
• 线程1不需要直接操作consumer实例，只需要把新的分区配置丢到线程安全的队列即可，ConcurrentLinkedQueue本身的Happens-Before规则可以保证线程1提交的配置对poll线程完全可见。
• 如果需要支持更多对consumer的控制操作（比如重置offset、暂停消费等），都可以参照这个模式，把操作请求放到队列里由poll线程统一执行。

内容的提问来源于stack exchange，提问作者Harsha Chittepu