歡迎來到DIVCSS5查找CSS資料與學習DIV CSS布局技術!
SpringBoot(八) SpringBoot整合Kafka

一、準備工作
提前說明:如果你運行出問題,請檢查Kafka的版本與SpringBoot的版本是否與我文中的一致,本文中的環境已經經過測試。

Kafka服務版本為 kafka_2.11-1.1.0 (Scala), 也就是1.1.0
SpringBoot版本:1.5.10.RELEASE

提前啟動zk,kafka,并且創建一個Topic

[[email protected] kafka_2.11-1.1.0]# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test_topic 

確保你的kafka能夠訪問,如果訪問不了,需要打開外網訪問。
config/server.properties

advertised.listeners=PLAINTEXT://192.168.239.128:9092

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

<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.11.0.0</version>
</dependency>

二、項目結構
為了更加體現實際開發需求,一般生產者都是在調用某些接口的服務處理完邏輯之后然后往kafka里面扔數據,然后有一個消費者不停的監控這個Topic,然后處理數據,所以這里把生產者作為一個接口,消費者放到kafka這個目錄下,注意@Component注解,不然掃描不到@KafkaListener

 

三、具體實現代碼
SpringBoot配置文件
application.yml

spring:
kafka:
bootstrap-servers: 192.168.239.128:9092
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
consumer:
group-id: test
enable-auto-commit: true
auto-commit-interval: 1000
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

生產者
package cn.saytime.web;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
* 測試kafka生產者
*/
@RestController
@RequestMapping("kafka")
public class TestKafkaProducerController {

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

@RequestMapping("send")
public String send(String msg){
kafkaTemplate.send("test_topic", msg);
return "success";
}

}
消費者
這里的消費者會監聽這個主題,有消息就會執行,不需要進行while(true)

package cn.saytime.kafka;

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

/**
* kafka消費者測試
*/
@Component
public class TestConsumer {

@KafkaListener(topics = "test_topic")
public void listen (ConsumerRecord<?, ?> record) throws Exception {
System.out.printf("topic = %s, offset = %d, value = %s \n", record.topic(), record.offset(), record.value());
}
}

項目啟動類

package cn.saytime;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class TestApplication{

public static void main(String[] args) {
SpringApplication.run(TestApplication.class, args);
}
}

四、測試
運行項目,執行:http://localhost:8080/kafka/send?msg=hello

控制臺輸出:

topic = test_topic, offset = 19, value = hello 
1
為了體現消費者不止執行一次就結束,再調用一次接口: 
http://localhost:8080/kafka/send?msg=kafka

topic = test_topic, offset = 20, value = kafka 
1
所以可以看到這里消費者實際上是不停的poll Topic數據的。

如需轉載,請注明文章出處和來源網址:http://www.133025.buzz/html/h57859.shtml

德甲最新射手榜