java生产者消费者代码_Java实现Kafka生产者消费者代码实例
Kafka的結(jié)構(gòu)與RabbitMQ類似,消息生產(chǎn)者向Kafka服務(wù)器發(fā)送消息,Kafka接收消息后,再投遞給消費者。
生產(chǎn)者的消費會被發(fā)送到Topic中,Topic中保存著各類數(shù)據(jù),每一條數(shù)據(jù)都使用鍵、值進(jìn)行保存。
每一個Topic中都包含一個或多個物理分區(qū)(Partition),分區(qū)維護(hù)著消息的內(nèi)容和索引,它們有可能被保存在不同服務(wù)器。
新建一個Maven項目,pom.xml 加入依賴:
org.apache.kafka
kafka-clients
2.3.0
1、編寫生產(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)者實例
KafkaProducer producer = new KafkaProducer<>(props);
ProducerRecord record = new ProducerRecord("topic1", "userName", "lc");
//發(fā)送記錄
producer.send(record);
producer.close();
}
}
運行后,可打開命令行工具,進(jìn)入Kafka目錄,執(zhí)行命令查詢服務(wù)器的Topic:
bin\windows\kafka-topics.bat --list --zookeeper localhost:2181
結(jié)果如下:
2、編寫消費者
本例中,消費者和生產(chǎn)者在同一個項目中,只是使用不同的啟動類。
消費者會為自已指定一個消費者組的標(biāo)識,每一條發(fā)布到Topic的記錄,都會被交付給消費者組的一個消費者實例。
如果多個消費者實例有相同的消費者組,則這些記錄會分配到各個消費者實例上,以達(dá)到負(fù)載均衡的目錄。
如果所有的消費者有不同的消費者組,則每一條記錄都會廣播到全部的消費者進(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");
//必須指定消費者組
props.put("group.id", "test");
//設(shè)置數(shù)據(jù)key和value的序列化處理類
props.put("key.deserializer", StringDeserializer.class);
props.put("value.deserializer", StringDeserializer.class);
//創(chuàng)建消息者實例
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());
}
}
}
}
運行后,IDEA控制臺其中輸出如下:
以上就是本文的全部內(nèi)容,希望對大家的學(xué)習(xí)有所幫助,也希望大家多多支持腳本之家。
總結(jié)
以上是生活随笔為你收集整理的java生产者消费者代码_Java实现Kafka生产者消费者代码实例的全部內(nèi)容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: 支付宝理财产品靠谱吗?放余额宝里面的钱安
- 下一篇: java await signal_Ja