Spring Boot 整合 Kafka

Kafka 環境搭建

kafka 安裝、設定、啟動、測試說明：

1. 安裝：直接官網下載安裝包，解壓到指定位置即可（kafka 依賴的 Zookeeper 在檔案中已包含）
下載地址：https://kafka.apache.org/downloads
範例版本：kafka_2.13-2.8.0.tgz
下載後可本地解壓安裝，解壓位置自選，如 D:\Java 下
解壓命令：tar -zxvf kafka_2.13-2.8.0.tgz
PS：可在 idea 命令列視窗或 git 提供的命令視窗中進行命令操作
使用 git 提供的命令視窗：空白資料夾中右鍵——》Git Bash Here 即可開啟

2. 新增地址設定
在 D:\Java\kafka_2.13-2.8.0\config\server.properties 中搜尋新增以下兩行設定：
listeners=PLAINTEXT://localhost:9092
advertised.listeners=PLAINTEXT://localhost:9092
說明：以上設定預設是註釋掉的，可搜尋找到，根據需求進行自定義地址設定

重要說明：以下命令操作預設都是在 D:\Java\kafka_2.13-2.8.0\ 即 kafaka 根目錄下進行！

3. 使用組態檔方式後臺啟動/關閉 Zookeeper 服務
啟動：bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
關閉【自選】：bin/zookeeper-server-stop.sh -daemon config/zookeeper.properties

4. 使用組態檔方式後臺啟動/關閉 kafka 服務
啟動：bin/kafka-server-start.sh -daemon config/server.properties
關閉【自選】：bin/kafka-server-stop.sh -daemon config/server.properties 

5. 服務測試

5.1 建立主題
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic Hello-Kafka

5.2 檢視主題（可能需要查一會兒）
bin/kafka-topics.sh --list --zookeeper localhost:2181

說明：傳送訊息和監聽訊息需要開啟兩個視窗進行測試！

5.3 傳送訊息（kafka 根目錄下新建視窗）
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Hello-Kafka
輸入以上命令回車後，可繼續輸入內容測試訊息傳送

5.4 監聽訊息（kafka 根目錄下新建視窗）
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic Hello-Kafka --from-beginning
輸入以上命令後，可觀察訊息接收情況，並且可在訊息傳送視窗繼續傳送訊息測試此監聽視窗的接收情況，正常接收，則服務環境搭建成功。

Spring Boot 整合 Kafka

環境：自行建立 Spring Boot 專案，新增測試依賴，並啟動 Zookeeper 和 kafka 服務。

注意：Zookeeper 預設好像佔用 8080 埠，自己注意埠佔用問題。

1. 新增依賴

<!-- spring-kafka -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

2. 新增設定

# kafka 設定
spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      # 發生錯誤後，訊息重發的次數。
      retries: 1
      #當有多個訊息需要被傳送到同一個分割區時，生產者會把它們放在同一個批次裡。該引數指定了一個批次可以使用的記憶體大小，按照位元組數計算。
      batch-size: 16384
      # 設定生產者記憶體緩衝區的大小。
      buffer-memory: 33554432
      # 鍵的序列化方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 值的序列化方式
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # acks=0 ： 生產者在成功寫入訊息之前不會等待任何來自伺服器的響應。
      # acks=1 ： 只要叢集的首領節點收到訊息，生產者就會收到一個來自伺服器成功響應。
      # acks=all ：只有當所有參與複製的節點全部收到訊息時，生產者才會收到一個來自伺服器的成功響應。
      acks: 1
    consumer:
      # 自動提交的時間間隔 在spring boot 2.X 版本中這裡採用的是值的型別為Duration 需要符合特定的格式，如1S,1M,2H,5D
      auto-commit-interval: 1S
      # 該屬性指定了消費者在讀取一個沒有偏移量的分割區或者偏移量無效的情況下該作何處理：
      # latest（預設值）在偏移量無效的情況下，消費者將從最新的記錄開始讀取資料（在消費者啟動之後生成的記錄）
      # earliest ：在偏移量無效的情況下，消費者將從起始位置讀取分割區的記錄
      auto-offset-reset: earliest
      # 是否自動提交偏移量，預設值是true,為了避免出現重複資料和資料丟失，可以把它設定為false,然後手動提交偏移量
      enable-auto-commit: false
      # 鍵的反序列化方式
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # 值的反序列化方式
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    listener:
      # 在偵聽器容器中執行的執行緒數。
      concurrency: 5
      # listner負責ack，每呼叫一次，就立即commit
      ack-mode: manual_immediate
      missing-topics-fatal: false

3. 建立訊息生產者

@Component
public class KafkaProducer {

    private Logger logger = LoggerFactory.getLogger(KafkaProducer.class);

    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    public static final String TOPIC_TEST = "Hello-Kafka";

    public static final String TOPIC_GROUP = "test-consumer-group";

    public void send(Object obj) {
        String obj2String = JSON.toJSONString(obj);
        logger.info("準備傳送訊息為：{}", obj2String);

        // 傳送訊息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_TEST, obj);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                //傳送失敗的處理
                logger.info(TOPIC_TEST + " - 生產者 傳送訊息失敗：" + throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                //成功的處理
                logger.info(TOPIC_TEST + " - 生產者 傳送訊息成功：" + stringObjectSendResult.toString());
            }
        });
    }

}

4. 建立訊息消費者

@Component
public class KafkaConsumer {

    private Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);

    @KafkaListener(topics = KafkaProducer.TOPIC_TEST, groupId = KafkaProducer.TOPIC_GROUP)
    public void topicTest(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        Optional<?> message = Optional.ofNullable(record.value());
        if (message.isPresent()) { // 包含非空值，則執行
            Object msg = message.get();
            logger.info("topic_test 消費了： Topic:" + topic + ",Message:" + msg);
            ack.acknowledge(); // 確認成功消費一個訊息
        }
    }

}

5. 訊息傳送測試

@RunWith(SpringRunner.class)
@SpringBootTest
public class KafkaProducerTest {

    private Logger logger = LoggerFactory.getLogger(KafkaProducerTest.class);

    @Resource
    private KafkaProducer kafkaProducer; // 注意使用自己建立的，看清楚！

    /*
      測試之前需要開啟 Kafka 服務
      啟動 Zookeeper：bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
      啟動 Kafka：bin/kafka-server-start.sh -daemon config/server.properties

      測試結果資料：

      準備傳送訊息為："你好，我是Lottery 001"
      Hello-Kafka - 生產者 傳送訊息成功：SendResult [producerRecord=ProducerRecord(topic=Hello-Kafka, partition=null,
      headers=RecordHeaders(headers = [], isReadOnly = true), key=null, value=你好，我是Lottery 001, timestamp=null),
      recordMetadata=Hello-Kafka-0@47]

      topic_test 消費了： Topic:Hello-Kafka,Message:你好，我是Lottery 001
     */
    @Test
    public void test_send() throws InterruptedException {
        // 迴圈傳送訊息
        while (true) {
            kafkaProducer.send("你好，我是Lottery 001");
            Thread.sleep(3500);
        }
    }

}