spirngboot 配置kafka实现group订阅消息

布满荆棘的人生 2022-04-17 07:06 127阅读 0赞

本人所使用的kafka版本是kafka\_2.11-2.0.1版本，jdk1.8、zookeeper-3.3.6，kafka运行于JVM环境，依赖zookeeper注册中心，所以需要准备这三个软件，本文测试使用的是windows版本的kafka和zookeeper便于测试

环境搭建 步骤

1、搭建jdk环境，配置环境变量（省略，请自行百度搜索）

![20181114145712630.png][]

2、zookeeper环境配置，修改zookeeper配置文件

将zoo\_sample.cfg 修改为zoo.cfg文件，修改zookeeper的访问端口

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpcWlfcQ_size_16_color_FFFFFF_t_70][]

启动zookeeper，在bin目录下zkServer.cmd双击即可

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpcWlfcQ_size_16_color_FFFFFF_t_70 1][]

3.1 下载安装文件： http://kafka.apache.org/downloads.html  
3.2 解压文件（本文解压到 E:\\worksoft\\kafka\_2.11-2.0.1）  
3.3 打开E:\\worksoft\\kafka\_2.11-2.0.1\\config  
3.4 从文本编辑器里打开 server.properties  
3.5 把 log.dirs的值改成 “E:\\worksoft\\kafka\_2.11-2.0.1\\kafka-logs”  
3.6 打开cmd  
3.7 进入kafka文件目录: cd /d E:\\worksoft\\kafka\_2.11-2.0.1\\  
3.8 输入并执行以打开kafka:  
.\\bin\\windows\\kafka-server-start.bat .\\config\\server.properties  
   
4. 创建topics  
4.1 打开cmd 并进入E:\\worksoft\\kafka\_2.11-2.0.1\\bin\\windows  
4.2 创建一个topic：  
kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

5. 打开一个Producer：  
cd /d E:\\worksoft\\kafka\_2.11-2.0.1\\bin\\windows  
kafka-console-producer.bat --broker-list localhost:9092 --topic test

6. 打开一个Consumer:  
cd /d E:\\worksoft\\kafka\_2.11-2.0.1\\bin\\windows  
kafka-console-consumer.bat --zookeeper localhost:2181 --topic test

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpcWlfcQ_size_16_color_FFFFFF_t_70 2][]

查看kafka中的所有组

./kafka-consumer-groups.bat  --bootstrap-server 127.0.0.1:9092 --list

![2018111415054642.png][]

查看某个consumer组的消费堆积量

kafka-consumer-groups.bat  --bootstrap-server 127.0.0.1:9092 --group lt\_datacenter\_processor\_kafka\_to\_len\_consumer --describe

![20181114150939636.png][]

查看kafka某个topic下partition信息

kafka-topics.bat --zookeeper 127.0.0.1:2181  --topic lenovo --describe   
![20181114151152976.png][]

springboot集成kafka。通过在启动类上开启注解方式启动

pom文件里面添加kafka依赖jar包

<dependency>
       <groupId>org.springframework.kafka</groupId>
       <artifactId>spring-kafka</artifactId>
       <version>1.3.5.RELEASE</version>
    </dependency>
    
    
    <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>kafka-clients</artifactId>
       <version>1.0.1</version>
    </dependency>

/**
     * 自动启用 Kafka
     */
    @Target(ElementType.TYPE)
    @Retention(RetentionPolicy.RUNTIME)
    @Import(KafkaConsumerConfiguration.class)
    @Documented
    public @interface EnableKafkaConsumer {
    }

@EnableKafka
    public class KafkaConsumerConfiguration {
        public static final Logger log = LoggerFactory.getLogger(KafkaConsumerConfiguration.class);
    
        /**
         * 引用kafkaProperties属性类
         */
        @Resource
        private KafkaProperties kafkaProperties;
    
    
        /**
         * 消费者线程使用自定义线程池
         *
         * @return
         */
        @Bean(name = "consumerTaskExecutor")
        public ThreadPoolTaskExecutor consumerTaskExecutor() {
    
    
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            //线程池活跃的线程数
            executor.setCorePoolSize(20);
            //executor.setCorePoolSize(Runtime.getRuntime().availableProcessors());
            //线程池最大活跃的线程数
            executor.setMaxPoolSize(200);
            //线程名称前缀
            executor.setThreadNamePrefix("kafkaThread-C-");
            //线程池维护线程所允许的空闲时间
            executor.setKeepAliveSeconds(30 * 60);
            //核心线程池也会请0
            executor.setAllowCoreThreadTimeOut(true);
            //线程池拒绝机制，抛弃最久的线程
    //        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            //executor.setWaitForTasksToCompleteOnShutdown(true);
            //线程池队列
            executor.setQueueCapacity(10000);
            executor.initialize();
    
            return executor;
        }
    
    
        /**
         * 重写设置自定义线程池
         *
         * @param <K>
         * @param <V>
         */
        public class CustomConcurrentKafkaListenerContainerFactory<K, V> extends ConcurrentKafkaListenerContainerFactory<K, V> {
    
            /**
             * The executor for threads that poll the consumer.
             */
            private AsyncListenableTaskExecutor consumerTaskExecutor;
    
    
            public CustomConcurrentKafkaListenerContainerFactory(AsyncListenableTaskExecutor consumerTaskExecutor) {
                this.consumerTaskExecutor = consumerTaskExecutor;
            }
    
            @Override
            protected void initializeContainer(ConcurrentMessageListenerContainer<K, V> instance) {
                super.initializeContainer(instance);
                instance.getContainerProperties().setConsumerTaskExecutor(consumerTaskExecutor);
            }
        }
    
        /**
         * 注入消费者监听器
         *
         * @return
         */
        @Bean
        public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory() {
    
            ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new CustomConcurrentKafkaListenerContainerFactory<Object, Object>(consumerTaskExecutor());
    
            if (null != kafkaProperties.getListener().getConcurrency()) {
                factory.setConcurrency(kafkaProperties.getListener().getConcurrency());
            }
            factory.setConsumerFactory(consumerFactory());
            factory.setBatchListener(true);
            return factory;
        }
     
        /**
         * 注入消费者工厂
         *
         * @return
         */
        @Bean
        public ConsumerFactory<Object, Object> consumerFactory() {
            return new DefaultKafkaConsumerFactory<>(kafkaProperties.buildConsumerProperties());
        }
    
        /**
         * 注入消费者
         *
         * @return
         */
        @Bean
        public KafkaConsumer<Object, Object> consumer() {
    
            log.info("<<<<<<<<<<<<<<< 加载 KafkaConsumer 服务 >>>>>>>>>>>>>>>>>>");
    
            return new KafkaConsumer<>(kafkaProperties.buildConsumerProperties());
        }
    
    
        /**
         * Admin管理配置
         *
         * @return
         */
        @Bean
        public KafkaAdmin kafkaAdmin() {
    
            log.info("<<<<<<<<<<<<<<< 加载 KafkaAdmin 服务 >>>>>>>>>>>>>>>>>>");
    
            Map<String, Object> configs = new HashMap<>();
            configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
            // 使用内置的Kafka
            //configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, StringUtils.arrayToCommaDelimitedString(kafkaEmbedded().getBrokerAddresses()));
            return new KafkaAdmin(configs);
        }
    
    }

@Component
    public class KafkaConsumerUtils {
    
        private static KafkaConsumer consumer;
    
        public KafkaConsumerUtils() {
        }
    
        @Autowired
        public KafkaConsumerUtils(KafkaConsumer kafkaConsumer) {
            KafkaConsumerUtils.consumer = kafkaConsumer;
            System.out.println("kafka消费者初始化完成...");
        }
    
        private static volatile KafkaConsumerUtils instance;
    
        /**
         * 单例模式，线程安全
         *
         * @return
         * @throws Exception
         */
        public static KafkaConsumerUtils getIstance() {
            // 对象实例化时与否判断（不使用同步代码块，instance不等于null时，直接返回对象，提高运行效率）
            if (instance == null) {
                //同步代码块（对象未初始化时，使用同步代码块，保证多线程访问时对象在第一次创建后，不再重复被创建）
                synchronized (KafkaConsumerUtils.class) {
                    //未初始化，则初始instance变量
                    if (instance == null) {
                        instance = new KafkaConsumerUtils();
                        //instance.afterPropertySet();//初始化配置和Pool池
                    }
                }
            }
            return instance;
        }
    
    }

spring:
      kafka:
          bootstrap-servers: 127.0.0.1:9092 #zookeeper配置地址
    #      client-id: wi_datacenter_processor #自定义客户端，并发设置此项会造成异常Error registering AppInfo mbean
          listener:
            #ack-mode: batch # 批量提交Offset
            #ack-time:
    #        poll-timeout: 5000 #pool Time
          consumer:
            group-id: lt_datacenter_processor_lenovo
            auto-offset-reset: earliest
            auto-commit-interval: 1000
            enable-auto-commit: true #手动提交Offset
            key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
            value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
            max-poll-records: 5 # 每次返回100条数据
          producer:
            retries: 0
            batch-size: 16384
            buffer-memory: 33554432
            key-serializer: org.apache.kafka.common.serialization.StringSerializer
            value-serializer: org.apache.kafka.common.serialization.StringSerializer

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpcWlfcQ_size_16_color_FFFFFF_t_70 3][]

import cn.litsoft.configuration.SpringContextUtil;
    import cn.litsoft.configuration.kafka.KafkaConsumerConfiguration;
    import cn.litsoft.iot.ailog.constant.ConstantGlobal;
    import cn.litsoft.iot.common.utils.thread.ThreadUtil;
    import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.CommandLineRunner;
    import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
    import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    import java.util.Arrays;
    import java.util.Map;
    import java.util.Optional;
    import java.util.concurrent.CountDownLatch;
    
    /**
     * trs数据消费kafka
     */
    @Component
    @ConditionalOnProperty(name = "media.kafka.consumer.enabled", havingValue = "true") // 开启注解才会启动
    public class KafaConsumerManual implements CommandLineRunner {
        public static final Logger log = LoggerFactory.getLogger(KafkaConsumerConfiguration.class);
    
        public KafaConsumerManual() {
            log.info("初始化：LenovoKafKaContentConsumer");
        }
    
    
        @Autowired
        private KafkaProperties kafkaProperties;
    
        public KafkaConsumer kafkaConsumer() {
            Map<String, Object> props = kafkaProperties.buildConsumerProperties();
            props.put(ConsumerConfig.GROUP_ID_CONFIG, ConstantGlobal.KAFKA_TO_ES_CONSUMER_GROUP_ID);
            return new KafkaConsumer(props);
        }
    
        /**
         * 简短任务使用默认线程池处理
         */
        @Resource(name = "manualTaskExecutor")
        private ThreadPoolTaskExecutor taskExecutor;
    
        @Override
        public void run(String... strings) {
    
            KafkaConsumer consumer = kafkaConsumer();
            taskExecutor.execute(new ThreadRunnable(consumer));// 丢 到线程池里面去运行
        }
    
    
        public class ThreadRunnable implements Runnable {
    
            private KafkaConsumer consumer;
    
            public ThreadRunnable(KafkaConsumer consumer) {
                this.consumer = consumer;
            }
    
            @Override
            public void run() {
    
                // 订阅主题
                consumer.subscribe(Arrays.asList("lenovo"));
                ThreadPoolTaskExecutor taskExecutor = SpringContextUtil.getBean("lkTaskExecutor");//线程池子
    
                //死循环不停的从broker中拿数据
                ConsumerRecords<String, String> list = null;
                while (true) {
                    list = consumer.poll(100);
    
                    if (null == list || list.count() == 0) {
                        log.info("业务【KafaConsumerManual】取出数据为空，休息一段时间进行等待");
                        ThreadUtil.sleep(60 * 1000);
                        continue;
                    }
    
                    log.info("业务【KafaConsumerManual】取出数据[{}]条", list.count());
    
                    final CountDownLatch countDownLatch = new CountDownLatch(list.count());
    
                    list.forEach(record -> taskExecutor.execute(new Runnable() {
                        @Override
                        public void run() {
                            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                            if (kafkaMessage.isPresent()) {
                                try {
                                    System.out.println("获取到kafka消息   ："+kafkaMessage.get().toString());
    //                                HashMap hashMap = (HashMap) JsonHelper.jsonToMap(kafkaMessage.get().toString());
    
                                } catch (Exception ex) {
                                    ThreadUtil.sleep(1000);//暂停一会，防止异常信息刷屏
                                    log.error("[KafaConsumerManual]发生异常：[{}]", ex);
                                } finally {
                                    countDownLatch.countDown();
                                }
    
                            } else {
                                countDownLatch.countDown();
                            }
                        }
                    }));
    
                    try {
                        countDownLatch.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        consumer.commitAsync();
                    }
                }
    
            }
        }
    }

[20181114145712630.png]: /images/20220417/b4bcf3227fc649b48120072e722ec978.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpcWlfcQ_size_16_color_FFFFFF_t_70]: /images/20220417/ced81a1fb4aa423e89d5ce7c5c230775.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpcWlfcQ_size_16_color_FFFFFF_t_70 1]: /images/20220417/9f4ef5074e9244c8a840bf2b547997b8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpcWlfcQ_size_16_color_FFFFFF_t_70 2]: /images/20220417/2af90c0e23ce4df780303476d31bfbae.png
[2018111415054642.png]: /images/20220417/66efc9c03b814398b0ca6627b7ce26b8.png
[20181114150939636.png]: /images/20220417/6a33ee2d33be423a8f86204fedcc8ac6.png
[20181114151152976.png]: /images/20220417/5d58179b303040c49fe705656815f598.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpcWlfcQ_size_16_color_FFFFFF_t_70 3]: /images/20220417/04e8b73d38f84f67ab2096c46d2b6d01.png