2. 程式人生 > >spirngboot 配置kafka實現group訂閱訊息

spirngboot 配置kafka實現group訂閱訊息

阿新 • • 發佈：2018-12-29

本人所使用的kafka版本是kafka_2.11-2.0.1版本，jdk1.8、zookeeper-3.3.6，kafka運行於JVM環境，依賴zookeeper註冊中心，所以需要準備這三個軟體，本文測試使用的是windows版本的kafka和zookeeper便於測試

環境搭建 步驟

1、搭建jdk環境，配置環境變數（省略，請自行百度搜索）

2、zookeeper環境配置，修改zookeeper配置檔案

將zoo_sample.cfg 修改為zoo.cfg檔案，修改zookeeper的訪問埠

啟動zookeeper，在bin目錄下zkServer.cmd雙擊即可

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

檢視kafka中的所有組

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

檢視某個consumer組的消費堆積量

kafka-consumer-groups.bat  --bootstrap-server 127.0.0.1:9092 --group lt_datacenter_processor_kafka_to_len_consumer --describe

檢視kafka某個topic下partition資訊

kafka-topics.bat --zookeeper 127.0.0.1:2181  --topic lenovo --describe 

springboot整合kafka。通過在啟動類上開啟註解方式啟動

pom檔案裡面新增kafka依賴jar包

<!-- 提供kafka核心工具包 -->
<dependency>
   <groupId>org.springframework.kafka</groupId>
   <artifactId>spring-kafka</artifactId>
   <version>1.3.5.RELEASE</version>
</dependency>

<!-- 提供AdminClient管理 AdminClientConfig-->
<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

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();
                }
            }

        }
    }
}

 

相關推薦

spirngboot 配置kafka實現group訂閱訊息

本人所使用的kafka版本是kafka_2.11-2.0.1版本，jdk1.8、zookeeper-3.3.6，kafka運行於JVM環境，依賴zookeeper註冊中心，所以需要準備這三個軟體，本文測試使用的是windows版本的kafka和zookeeper便於測試 環境搭建 步驟 1、

Kafka實現篇之訊息和日誌

http://blog.csdn.net/honglei915/article/details/37760631訊息格式日誌一個叫做“my_topic”且有兩個分割槽的的topic,它的日誌有兩個資料夾組成，my_topic_0和my_topic_1,每個資料夾裡放著具體的資

漫遊Kafka實現篇之訊息和日誌

訊息格式 日誌 一個叫做“my_topic”且有兩個分割槽的的topic,它的日誌有兩個資料夾組成，my_topic_0和my_topic_1,每個資料夾裡放著具體的資料檔案，每個資料檔案都是一系列的日誌實體，每個日誌實體有一個4個位元組的整數N標註訊息的長度，後邊

Kafka-API中介軟體MQ訊息佇列在Maven專案中的配置使用操作 (分散式釋出訂閱訊息系統)

一、 Maven依賴 <dependency> <groupId>com.foriseland.fjf.mq</groupId> <artifactId>fjf-mq-kafka</artifactId> &

用雲開發實現小程式訂閱訊息配置

10月12日微信官方釋出了小程式模板訊息能力調整的通知 : https://developers.weixin.qq.com/community/develop/doc/00008a8a7d8310b6bf4975b635a401 相比之前的有了較大的調整，主要的調整就是從開發者主動下發訊息改為使用者自主訂閱

RocketMQ中介軟體訊息佇列在Maven專案中的配置使用操作 (分散式釋出訂閱訊息系統)

一、專案引用 <dependency>     <groupId>com.foriseland.fjf.mq</groupId>     <artifactI

Spring-Kafka XML配置方法實現生產和消費

  Spring-Kafka XML配置方法實現生產和消費  1. 生產者配置xml Step1：producerProperties：設定生產者公產需要的配置 ； Step2：producerFactory：定義了生產者工廠構造方法

kafka告警簡單方案 依賴配置中心實現注有@ConfigurationProperties的bean相關屬性重新整理 簡單封裝kafka相關的api

一、前言 　　為什麼要設計kafka告警方案？現成的監控專案百度一下一大堆，KafkaOffsetMonitor、KafkaManager、 Burrow等，具體參考：kafka的訊息擠壓監控。由於本小組的專案使用的kafka叢集並沒有被公司的kafka-manager管理，所以只能自己簡單做一個告警。

ActiveMQ之釋出- 訂閱訊息模式實現

一、概念 釋出者/訂閱者模型支援向一個特定的訊息主題釋出訊息。0或多個訂閱者可能對接收來自特定訊息主題的訊息感興趣。在這種模型下，釋出者和訂閱者彼此不知道對方。這種模式好比是匿名公告板。這種模式被概括為：多個消費者可以獲得訊息 在釋出者和訂閱者之間存在時間依賴性。釋出者需要建

Redis學習（八） Redis丟失訂閱訊息分析與處理和client-output-buffer-limit屬性配置說明

Redis快取行情資料，發現程式執行一段時間後，出現subscribe執行緒不再能夠接收到訂閱的行情資料，發現是由Redis的輸出緩衝機制導致的。 Redis為了解決輸出緩衝區訊息大量堆積的隱患，設定了一些保護機制，主要採用兩種限制措施： 大小限制，當某一客戶端緩衝區超過設定值後直接關

分散式釋出訂閱訊息系統Kafka

文章目錄 Kafka概述 Kafka架構及核心概念 Kafka單節點單Broker部署之Zookeeper安裝 單節點單Broker部署 單節點多Broker部署及使用 Kafka容錯性測試

Netty實戰開發(7):Netty結合kafka實現分散式訊息佇列

在分散式遊戲伺服器系統中，訊息處理佇列主要解決問題就是解耦系統中的業務，使得每個系統看起來功能比較單一，而且解決一些全服資料共享等問題。 通常我們知道kafka是作為訊息佇列比較火的一種方式，其實還有（Active MQ，Rabbit MQ，Zero MQ）個人

Spring Cloud Config + Spring Cloud Bus + kafka實現配置中心配置動態更新

一、架構圖 當在 Git 倉庫中某個應用配置檔案中的引數更新後，只需要通過 POST方法訪問 config Server 的 /actuator/bus-refresh 介面，就可以讓所以的微服務節點更新配置。 在我們的 DEMO 中有一個 Config Server

SpringBoot開發案例之整合Kafka實現訊息佇列

　　前言 　　最近在做一款秒殺的案例，涉及到了同步鎖、資料庫鎖、分散式鎖、程序內佇列以及分散式訊息佇列，這裡對SpringBoot整合Kafka實現訊息佇列做一個簡單的記錄。 　　Kafka簡介 　　Kafka是由Apache軟體基金會開發的一個開源流處理平臺，由Scala和Java編寫

SpringBoot整合Kafka實現訊息上報

一、該篇部落格使用技術版本 SpringBoot：1.5.9.RELEASE zookeeper：zookeeper-3.4.5 kafka：kafka_2.10-0.10.2.1 二、SpringBoot整合Kafka 1、首先修改kafka的server.proper

SpringCloudBus使用Kafka實現訊息匯流排

Kafka是分散式釋出-訂閱訊息系統，最初由LinkedIn公司開發，之後成為之後成為Apache基金會的一部分，由Scala和Java編寫。Kafka是一種快速、可擴充套件的、設計內在就是分散式的，分割槽的和可複製的提交日誌服務。 在開始本文前，需要

分散式釋出訂閱訊息系統 Kafka

kafka是一種高吞吐量的分散式釋出訂閱訊息系統，她有如下特性： 通過O(1)的磁碟資料結構提供訊息的持久化，這種結構對於即使數以TB的訊息儲存也能夠保持長時間的穩定效能。 高吞吐量：即使是非常普通的硬體kafka也可以支援每秒數十萬的訊息。 支援通過kafka伺服器

釋出訂閱訊息系統--kafka的解析。

Kafka介紹： Kafka是由Apache軟體基金會開發的一個開源流處理平臺，由Scala和Java編寫。Kafka是一種高吞吐量的分散式釋出訂閱訊息系統，它可以處理消費者規模的網站中的所有動作流資料。這種動作（網頁瀏覽，搜尋和其他使用者的行動）是在現代網路上的許多社會功

springboot整合kafka實現訊息的生產與消費--訊息的生產

由於工作需要，最近在研究springboot整合kafka。做一個分散式的同步應用程式。springboot整合kafka須注意版本。對於springboot 1.5版本之前的話，需要自己去配置java configuration，而1.5版本以後則提供了auto confi

Kafka（分散式釋出訂閱訊息系統）

http://kafka.apache.org/目前越來越多的開源分散式處理系統如Apache Storm、Spark都支援與Kafka整合。 使用場景：設想這樣一個情景：想分析使用者在網站上的的瀏覽行為。這些瀏覽日誌，存資料庫浪費，直接存硬碟又怕到時候操作效率低。 此時，