kafka 消费者线程安全问题详细探讨

news2024/9/24 1:19:40

内容概要

主要内容

常见错误案例

下面这段代码大概逻辑

初始化时 实例化KafkaConsumer, 开启线程拉取消息并且处理
资源释放回调 停止线程、调用kafkaConsumer.close进行资源释放

表面上没有问题，但实际上可能出现线程安全问题，因为poll 和 close 两个操作可能同时执行,因此存在线程安全问题, 如何修改，读者自己思考下。

    @PostConstruct
    public void consumer(){
        kafkaConsumer = new KafkaConsumer(getConfig());
        kafkaConsumer.subscribe(Arrays.asList("test_partition_num"));

        new Thread(new Runnable() {
            @Override
            public void run() {
                while(running){
                    ConsumerRecords<String,String> records = kafkaConsumer.poll(Duration.ofMillis(1000));
                    records.forEach(record->{
                        System.out.println(" partition =" + record.partition()  +" offset  = " + record.offset() + " value = " + record.value());
                    });
                }
            }
        }).start();
    }

    @PreDestroy
    public void close(){
        running = false;
        if(kafkaConsumer != null){
            kafkaConsumer.close();
        }
    }

消费者非线程安全代码解读

kafka生成者是线程安全的，但消费者是非线程安全的。KafkaConsumer

相关操作前
- 调用acquire()方法，校验线程安全问题，如果发现其他线程也在操作，则直接抛出异常。
操作完成后
- 调用release()清除痕迹

acquire()相对于加锁，release()相当于释放锁。

参看poll 方法实现，一目了然。

    private void acquire() {
        long threadId = Thread.currentThread().getId();
        if (threadId != this.currentThread.get() && !this.currentThread.compareAndSet(-1L, threadId)) {
            throw new ConcurrentModificationException("KafkaConsumer is not safe for multi-threaded access");
        } else {
            this.refcount.incrementAndGet();
        }
    }
    
      private void release() {
        if (this.refcount.decrementAndGet() == 0) {
            this.currentThread.set(-1L);
        }
    }

poll源码

如何实现消费者多线程消费消息呢

思路1

每次实例化一个 KafkaConsumer

这种方式实现简单，但每次都需要建立TCP 链接

思路2

相关操作方法加上 synchronized,获取使用Lock 加锁保证线程安全

这种方式性能较差

思路3

拉取消息使用一个线程，消息处理使用多线程

因为通常拉取消息比较快，消息处理比较耗时，由于消息处理不涉及KafkaConsumer 相关API 操作，因此不存在线程安全问题。这种方式建议消息位移设置自动提交，否则编程复杂度较高。

示例代码

ConsumerRecords<String,String> records = kafkaConsumer.poll(Duration.ofMillis(1000));

executorService.execute(()->{
    //处理消息
    records.forEach(record->{
        System.out.println(" partition =" + record.partition()  +" offset  = " + record.offset() + " value = " + record.value());
    });
});