多Kafka消费者同步与高可用部署：确保同实例运行需求

问题根源分析 #

你之前用@KafkaListener(topics = {"topic1", "topic2"}, concurrency=2)搭配RoundRobinAssignor的方案失效，核心原因是：

• concurrency=2会在单个实例内启动两个独立消费者线程，多实例部署时，Kafka消费者组的分区分配逻辑可能把两个Topic的分区分别分配给不同实例的线程，导致两个消费者不在同一实例，无法用内存锁同步。
• 这种方式无法将两个消费者绑定为一个整体，做不到“要么同时运行，要么同时切换”的要求。

推荐方案一：基于Kafka消费者组的单实例独占消费 #

利用Kafka消费者组的特性，让单个实例的消费者同时接管两个Topic的分区，多实例部署时只有一个实例处于活跃消费状态，故障时自动触发重平衡切换。

实现步骤 #

1. 单消费者监听两个Topic：用一个@KafkaListener同时监听两个Topic，不设置concurrency（默认值为1，即单个消费线程）。
2. 内存锁同步数据库操作：在消费逻辑中用同一内存锁同步两个Topic的数据库变更。
3. 多实例部署：多个应用实例使用同一消费者组ID，Kafka会自动保证只有一个实例的消费者被分配到两个Topic的分区，其他实例处于空闲状态。

代码示例 #

@Component
public class CombinedTopicConsumer {
    // 内存锁，用于同步两个Topic的数据库操作
    private final Lock dbSyncLock = new ReentrantLock();

    @KafkaListener(topics = {"topic1", "topic2"}, groupId = "combined-consumer-group")
    public void consumeMessages(ConsumerRecord<String, String> record, Acknowledgment ack) {
        dbSyncLock.lock();
        try {
            // 根据Topic区分处理逻辑
            if ("topic1".equals(record.topic())) {
                handleTopic1Message(record);
            } else if ("topic2".equals(record.topic())) {
                handleTopic2Message(record);
            }
            // 手动提交偏移量（根据业务需求选择自动/手动提交）
            ack.acknowledge();
        } finally {
            dbSyncLock.unlock();
        }
    }

    private void handleTopic1Message(ConsumerRecord<String, String> record) {
        // Topic1的业务处理+数据库变更逻辑
    }

    private void handleTopic2Message(ConsumerRecord<String, String> record) {
        // Topic2的业务处理+数据库变更逻辑
    }
}

优势 #

• 无需额外依赖组件，完全基于Kafka原生特性实现。
• 故障切换由Kafka自动触发，无需手动编写切换逻辑。
• 天然保证两个Topic的消费都在同一实例，内存锁有效。

备选方案二：基于分布式锁的实例激活控制 #

如果对Kafka重平衡的切换时间不满意，可以用分布式锁（如Redis、ZooKeeper）手动控制只有一个实例能启动两个消费者，其他实例处于 standby 状态。

核心逻辑 #

1. 应用启动时尝试获取分布式锁。
2. 获取锁成功的实例，启动两个独立的消费者（分别监听topic1和topic2），用内存锁同步数据库操作。
3. 获取锁失败的实例，定期尝试抢锁，当活跃实例故障锁释放后，自动激活自身的两个消费者。
4. 活跃实例需定期续期分布式锁，避免因网络波动导致误切换。

伪代码示例 #

@Component
public class ConsumerActivator implements ApplicationListener<ContextRefreshedEvent> {
    private final Lock dbSyncLock = new ReentrantLock();
    private final RedisDistributedLock distributedLock; // 自定义Redis锁实现
    private Consumer<String, String> topic1Consumer;
    private Consumer<String, String> topic2Consumer;
    private volatile boolean isActiveInstance = false;

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        // 启动定时任务，尝试激活消费者
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(this::tryActivateConsumers, 0, 5, TimeUnit.SECONDS);
    }

    private void tryActivateConsumers() {
        if (isActiveInstance) {
            // 续期分布式锁
            distributedLock.renewLock("kafka-active-consumer-lock");
            return;
        }
        // 尝试获取锁，有效期30秒
        if (distributedLock.acquireLock("kafka-active-consumer-lock", 30)) {
            isActiveInstance = true;
            startTopic1Consumer();
            startTopic2Consumer();
            // 启动锁续期任务
            startLockRenewTask();
        }
    }

    private void startTopic1Consumer() {
        topic1Consumer = new KafkaConsumer<>(buildConsumerConfig());
        topic1Consumer.subscribe(Collections.singletonList("topic1"));
        new Thread(() -> {
            while (isActiveInstance) {
                ConsumerRecords<String, String> records = topic1Consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    dbSyncLock.lock();
                    try {
                        handleTopic1Message(record);
                        topic1Consumer.commitSync();
                    } finally {
                        dbSyncLock.unlock();
                    }
                }
            }
        }).start();
    }

    // startTopic2Consumer、handleTopic2Message、buildConsumerConfig等方法逻辑类似
}

优势 #

• 可自定义故障切换时间（由锁过期时间决定）。
• 对消费逻辑的控制更精细。

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