springboot還處於學習階段，又同時在學習kafka，兩者結合，繼續學習。

1、官網下載kafka

2、解壓

3、對於單節點來說，按照官網上操作即可實現訊息的傳送和接收。

但是對於客戶端，是通過 @KafkaListener 註解監聽生產者傳送的訊息，故需要修改config/server.properties 檔案

如上圖，開啟listeners ，將預設的件監聽IP+埠改為具體的伺服器地址

4、建立生產者，相關配置

public staticMapproducerConfigs() {

Mapprops =newHashMap<>();

props.put
 
(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.0.90:9092"); // broker 配置

props.put(ProducerConfig.RETRIES_CONFIG,0);//重試次數

props.put(ProducerConfig.BATCH_SIZE_CONFIG,4096);//快取每個topic-partition對應訊息佇列集合

props.put(ProducerConfig.LINGER_MS_CONFIG,1);

props.put(ProducerConfig.BUFFER_MEMORY_CONFIG,40960);// Producer非同步佇列記憶體的最大值
 


props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class); //序列化

props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);

returnprops;

}

5、重寫kafkaTemplate類方法，

   @Bean
    public static KafkaTemplate<String, String> kafkaTemplate() {
        return
 
 new KafkaTemplate<String, String>(producerFactory());
    }
    public static  ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

指向自己的config配置

6、消費者配置

public Map<String, Object> consumerConfigs() {
        Map<String, Object> propsMap = new HashMap<>();
        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.90:9092");
        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
        propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); //反序列化
        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);//反序列化
        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        return propsMap;
    }
    @Bean
    public Listener listener() {
        return new Listener();
    }

監聽使用@KafkaListener註解

@KafkaListener(topics={"test1"})
    public void listen(ConsumerRecord<?, ?> record) {
        Gson gson = new Gson();
        log.error("監聽器" + gson.toJson(record));
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            System.out.println("listen1 " + message);
            //對訊息的處理
        }
    }

訊息傳送

 @RequestMapping("/sendMsg")
    @ResponseBody
    public void sendMsg(String msg){
        KafkaTemplate<String, String> kafkaTemplate =  producerConfig.kafkaTemplate();
        kafkaTemplate.send("test1","測試訊息接受傳送");

    }

消費者接收到的資訊：

{"topic":"test1","partition":0,"offset":1,"timestamp":1496738649398,"timestampType":"CREATE_TIME","checksum":460720665,"serializedKeySize":-1,"serializedValueSize":24,"value":"測試訊息接受傳送"}