spring boot集成kafka簡單實用

 

先導入spring boot整合kafka的依賴

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

然后是application.yml中關於kafka的配置

spring:
  kafka:
    bootstrap-servers: ip:端口 #kafka服務器地址 多個以,號隔開
    #生產者配置
    producer:
      batch-size: 16384  #每次批量發送信息的數量
      buffer-memory: 33554432 #達到該緩沖區大小就發送數據
      key-serializer: org.apache.kafka.common.serialization.StringSerializer #key的序列化器
      value-serializer: org.apache.kafka.common.serialization.StringSerializer #value的序列化器
    #消費者配置
    consumer:
      group-id: test #消費者分組id,同一個分組的消費者不會讀取到同一個消息
      enable-auto-commit: true #啟用自動提交偏移量
      auto-commit-interval: 100 #設置偏移量提交間隔
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer #key的反序列化器
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer #value的反序列化器

#自定義主題名,在代碼中可以讀取,方便修改主題
topic: testTopic

這是基礎配置,我生產者和消費者這里都是同一個工程進行測試了,如果是兩個不同的工程,只寫生產者或消費者的配置就行

@Component
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String,Object> kafkaTemplate;

    @Value("${topic}")
    private  String topic;

    public  void sendMessage(String message){
         kafkaTemplate.send(topic,message);
    }

}

這里新建了一個kafka生產者類用於發送信息到kafka,並且交給了spring 管理,在我們需要使用的地方直接注入進來即可.

這個KafkaTemplate<>類是spring boot提供的模板類,封裝了關於kafka的操作.

 

kafka消費者

@Component
public class KafkaConsumer {

    @KafkaListener(topics = {"${topic}"})
    public void getMessage(String message){
        System.out.println(message);
    }

}

同樣交由spring管理,然后使用@KafkaListener注解,定義要監聽的主題,至於{"${topic}"}這個是讀取配置文件中的topic配置項,也是我所定義的主題,但監聽到有信息時,

就會調用這個方法,將信息作為一個字符串參數傳遞進來,里面就可以寫你的業務邏輯了.

 

kafka大致使用就是這樣,很多細節配置,暫時也不清楚,有機會在補充吧