小編這次要給大家分享的是Java如何實(shí)現(xiàn)Kafka生產(chǎn)者消費(fèi)者，文章內(nèi)容豐富，感興趣的小伙伴可以來(lái)了解一下，希望大家閱讀完這篇文章之后能夠有所收獲。
創(chuàng)新互聯(lián)公司服務(wù)項(xiàng)目包括宣恩網(wǎng)站建設(shè)、宣恩網(wǎng)站制作、宣恩網(wǎng)頁(yè)制作以及宣恩網(wǎng)絡(luò)營(yíng)銷策劃等。多年來(lái)，我們專注于互聯(lián)網(wǎng)行業(yè)，利用自身積累的技術(shù)優(yōu)勢(shì)、行業(yè)經(jīng)驗(yàn)、深度合作伙伴關(guān)系等，向廣大中小型企業(yè)、政府機(jī)構(gòu)等提供互聯(lián)網(wǎng)行業(yè)的解決方案，宣恩網(wǎng)站推廣取得了明顯的社會(huì)效益與經(jīng)濟(jì)效益。目前，我們服務(wù)的客戶以成都為中心已經(jīng)輻射到宣恩省份的部分城市，未來(lái)相信會(huì)繼續(xù)擴(kuò)大服務(wù)區(qū)域并繼續(xù)獲得客戶的支持與信任！
Kafka的結(jié)構(gòu)與RabbitMQ類似，消息生產(chǎn)者向Kafka服務(wù)器發(fā)送消息，Kafka接收消息后，再投遞給消費(fèi)者。
生產(chǎn)者的消費(fèi)會(huì)被發(fā)送到Topic中，Topic中保存著各類數(shù)據(jù)，每一條數(shù)據(jù)都使用鍵、值進(jìn)行保存。
每一個(gè)Topic中都包含一個(gè)或多個(gè)物理分區(qū)（Partition），分區(qū)維護(hù)著消息的內(nèi)容和索引，它們有可能被保存在不同服務(wù)器。
新建一個(gè)Maven項(xiàng)目，pom.xml 加入依賴：
org.apache.kafka kafka-clients 2.3.0
1、編寫(xiě)生產(chǎn)者
將消息投遞到Kafka服務(wù)器的名稱為“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服務(wù)器地址
props.put("bootstrap.servers", "localhost:9092");
//設(shè)置數(shù)據(jù)key和value的序列化處理類
props.put("key.serializer", StringSerializer.class);
props.put("value.serializer", StringSerializer.class);
//創(chuàng)建生產(chǎn)者實(shí)例
KafkaProducer producer = new KafkaProducer<>(props);
ProducerRecord record = new ProducerRecord("topic1", "userName", "lc");
//發(fā)送記錄
producer.send(record);
producer.close();
}
} 運(yùn)行后，可打開(kāi)命令行工具，進(jìn)入Kafka目錄，執(zhí)行命令查詢服務(wù)器的Topic：
bin\windows\kafka-topics.bat --list --zookeeper localhost:2181
結(jié)果如下：
2、編寫(xiě)消費(fèi)者
本例中，消費(fèi)者和生產(chǎn)者在同一個(gè)項(xiàng)目中，只是使用不同的啟動(dòng)類。
消費(fèi)者會(huì)為自已指定一個(gè)消費(fèi)者組的標(biāo)識(shí)，每一條發(fā)布到Topic的記錄，都會(huì)被交付給消費(fèi)者組的一個(gè)消費(fèi)者實(shí)例。
如果多個(gè)消費(fèi)者實(shí)例有相同的消費(fèi)者組，則這些記錄會(huì)分配到各個(gè)消費(fèi)者實(shí)例上，以達(dá)到負(fù)載均衡的目錄。
如果所有的消費(fèi)者有不同的消費(fèi)者組，則每一條記錄都會(huì)廣播到全部的消費(fèi)者進(jìn)行處理。
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服務(wù)器地址
props.put("bootstrap.servers", "localhost:9092");
//必須指定消費(fèi)者組
props.put("group.id", "test");
//設(shè)置數(shù)據(jù)key和value的序列化處理類
props.put("key.deserializer", StringDeserializer.class);
props.put("value.deserializer", StringDeserializer.class);
//創(chuàng)建消息者實(shí)例
KafkaConsumer consumer = new KafkaConsumer<>(props);
//訂閱topic1的消息
consumer.subscribe(Arrays.asList("topic1"));
//到服務(wù)器中讀取記錄
while (true){
ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
for(ConsumerRecord record : records){
System.out.println("key:" + record.key() + "" + ",value:" + record.value());
}
}
}
} 運(yùn)行后，IDEA控制臺(tái)其中輸出如下：
看完這篇關(guān)于Java如何實(shí)現(xiàn)Kafka生產(chǎn)者消費(fèi)者的文章，如果覺(jué)得文章內(nèi)容寫(xiě)得不錯(cuò)的話，可以把它分享出去給更多人看到。
