码迷,mamicode.com
首页 > 其他好文 > 详细

Kafka 消费时机问题

时间:2020-03-18 13:51:51      阅读:60      评论:0      收藏:0      [点我收藏+]

标签:one   boot   完成   row   恢复   做了   resume   while   tom   

kafka的消费时机

问题

kafka启动后开始消费的话, 如果此时消费流程中有其他依赖没有启动完成的话,比如 Redis , Mysql ,RPC等就会有空指针或其他问题。这时候就要延后kafka的消费时机。

解决

kafka启动时,不在启动时开启消费线程。

public class KafkaConsumer {

private KafkaConsumer<String, byte[]> consumer;
private ThreadPoolExecutor threadPoolExecutor;
private ThreadPoolExecutor threadConsumerExecutor;


public void init(String kafkaTopic, String groupIdConfig) {
    //init pool
    //init consumer
    。。。。。
}

//开始消费单独拿出方法
public void startConsumer(final Handler handler) {
    threadConsumerExecutor.submit(new Runnable() {
        @Override
        public void run() {
            while(true){
                ConsumerRecords<String, byte[]> data = consumer.poll(200);
                .....
                handler.onMessage(data)
                sleep(100)
            }
            
        }
    });
}

然后在SpringBoot启动后,再启用kafka的消费线程

@Component
public class ApplicationRunnerImpl implements ApplicationRunner {
    @Override
    public void run(ApplicationArguments args) throws Exception {
        //校验Redis
        //校验Mysql
        //校验RPC
        kafkaConsumer.start()
    }
}

此时启用kafkaConsumer,就不会有问题了。

升级版

可以设置标志位,来灵活做到启/停/暂停/恢复

public class KafkaConsumer {
    //标识位,是否开始消费线程
private AtomicBoolean polling = new AtomicBoolean(true);
private KafkaConsumer<String, byte[]> consumer;
private ThreadPoolExecutor threadPoolExecutor;
private ThreadPoolExecutor threadConsumerExecutor;
private Handler handler;


public void init(String kafkaTopic, String groupIdConfig) {
    //init pool
    //init consumer
    。。。。。
    startConsumer()
}

//开始消费单独拿出方法
public void startConsumer() {
    threadConsumerExecutor.submit(new Runnable() {
        @Override
        public void run() {
            while(this.polling.get()){
                ConsumerRecords<String, byte[]> data = consumer.poll(200);
                .....
                handler.onMessage(data)
                sleep(100)
            }
        }
    });
}
//暂停消费线程
public void pause(){
    this.polling.set(false)
}

//恢复消费线程
public void resume(){
    this.polling.set(true)
}

//关闭消费
public void shutDown(){
    pause();
    unsubscribe();
    threadConsumerExecutor.shutdown();
    ...

}

代码做了些简略处理。明白思路就好做,然后就可以针对业务场景细化一下了。

Kafka 消费时机问题

标签:one   boot   完成   row   恢复   做了   resume   while   tom   

原文地址:https://www.cnblogs.com/ElEGenT/p/12516578.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!