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=https://www.cnblogs.com/luisblog/archive/2023/04/11/你好，我是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);
}
}
}
都看到最后了，右下角來個贊鴨！-.- 歡迎評論留言~
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/549740.html
標籤：其他