溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
小編這次要給大家分享的是Java如何實現Kafka生產者消費者，文章內容豐富，感興趣的小伙伴可以來了解一下，希望大家閱讀完這篇文章之后能夠有所收獲。
Kafka的結構與RabbitMQ類似，消息生產者向Kafka服務器發送消息，Kafka接收消息后，再投遞給消費者。
生產者的消費會被發送到Topic中，Topic中保存著各類數據，每一條數據都使用鍵、值進行保存。
每一個Topic中都包含一個或多個物理分區（Partition），分區維護著消息的內容和索引，它們有可能被保存在不同服務器。
新建一個Maven項目，pom.xml 加入依賴：
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.3.0</version>
</dependency>1、編寫生產者
將消息投遞到Kafka服務器的名稱為“topic1”的Topic中
package com.example.kafkatest;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
public class Producer {
public static void main(String[] args) {
//配置信息
Properties props = new Properties();
//kafka服務器地址
props.put("bootstrap.servers", "localhost:9092");
//設置數據key和value的序列化處理類
props.put("key.serializer", StringSerializer.class);
props.put("value.serializer", StringSerializer.class);
//創建生產者實例
KafkaProducer<String,String> producer = new KafkaProducer<>(props);
ProducerRecord record = new ProducerRecord<String, String>("topic1", "userName", "lc");
//發送記錄
producer.send(record);
producer.close();
}
}運行后，可打開命令行工具，進入Kafka目錄，執行命令查詢服務器的Topic：
bin\windows\kafka-topics.bat --list --zookeeper localhost:2181
結果如下：
2、編寫消費者
本例中，消費者和生產者在同一個項目中，只是使用不同的啟動類。
消費者會為自已指定一個消費者組的標識，每一條發布到Topic的記錄，都會被交付給消費者組的一個消費者實例。
如果多個消費者實例有相同的消費者組，則這些記錄會分配到各個消費者實例上，以達到負載均衡的目錄。
如果所有的消費者有不同的消費者組，則每一條記錄都會廣播到全部的消費者進行處理。
package com.example.rabbittest;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
public class Consumer {
public static void main(String[] args) {
//配置信息
Properties props = new Properties();
//kafka服務器地址
props.put("bootstrap.servers", "localhost:9092");
//必須指定消費者組
props.put("group.id", "test");
//設置數據key和value的序列化處理類
props.put("key.deserializer", StringDeserializer.class);
props.put("value.deserializer", StringDeserializer.class);
//創建消息者實例
KafkaConsumer<String,String> consumer = new KafkaConsumer<>(props);
//訂閱topic1的消息
consumer.subscribe(Arrays.asList("topic1"));
//到服務器中讀取記錄
while (true){
ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(100));
for(ConsumerRecord<String,String> record : records){
System.out.println("key:" + record.key() + "" + ",value:" + record.value());
}
}
}
}運行后，IDEA控制臺其中輸出如下：
看完這篇關于Java如何實現Kafka生產者消費者的文章，如果覺得文章內容寫得不錯的話，可以把它分享出去給更多人看到。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
