一、引入依賴

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

二、配置yml

spring:
   kafka:     # 指定kafka 代理地址，可以多個
      bootstrap-servers: 47.52.199.52:9092
      template:    # 指定預設topic id
        default-topic: producer
      listener:   # 指定listener 容器中的執行緒數，用於提高併發量
        concurrency: 5
      consumer:
        group-id: myGroup # 指定預設消費者group id
        client-id: 200
        max-poll-records: 200
        auto-offset-reset: earliest # 最早未被消費的offset
      producer:
        batch-size: 1000 # 每次批量傳送訊息的數量
        retries: 3
        client-id: 200
 

三、生成者使用示例

package com.example.demo.kafka;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
 
import org.springframework.util.concurrent.ListenableFuture;

import java.util.concurrent.ExecutionException;

@Component
public class Producer {
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    /**
     * 傳送訊息到kafka
     */
    public RecordMetadata sendChannelMess(String topic, String message) {
        ListenableFuture
 
<SendResult<String, String>> future = kafkaTemplate.send(topic,message);
        RecordMetadata recordMetadata = null;
        try {
            recordMetadata = future.get().getRecordMetadata();
        } catch (InterruptedException|ExecutionException e) {
            e.printStackTrace();
            System.out.println("傳送失敗");
        }
        System.out.println("傳送成功");
        System.out.println("partition:"+recordMetadata.partition());
        System.out.println("offset:"+recordMetadata.offset());
        System.out.println("topic:"+recordMetadata.topic());

        return recordMetadata;
    }
}

四、消費者使用示例

package com.example.demo.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class Consumer {

    /**
     * 有訊息就讀取,只讀取訊息value
     */
    @KafkaListener(topics = {"test13"})
    public void receiveMessage(String message){
        //收到通道的訊息之後執行秒殺操作
        System.out.println(message);
    }

    /**
     * 有訊息就讀取,批量讀取訊息value
     */
    @KafkaListener(topics = "test12")
    public void onMessage(List<String> crs) {
        for(String str : crs){
            System.out.println("test12:" + str);
        }
    }

    /**
     * 有訊息就讀取,讀取訊息topic，offset，key，value等資訊
     */
    @KafkaListener(topics = "test14")
    public void listenT1(ConsumerRecord<?, ?> cr){
        System.out.println("listenT1收到訊息,topic:>>>" + cr.topic() + "  offset:>>" + cr.offset()+ "  key:>>" + cr.key() + "  value:>>" + cr.value());
    }
}