av一区二区在线观看_亚洲男人的天堂网站_日韩亚洲视频_在线成人免费_欧美日韩精品免费观看视频_久草视

您的位置:首頁技術文章
文章詳情頁

Java實現Kafka生產者消費者代碼實例

瀏覽:4日期:2022-09-02 14:47:11

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:

binwindowskafka-topics.bat --list --zookeeper localhost:2181

結果如下:

Java實現Kafka生產者消費者代碼實例

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生產者消費者代碼實例

以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支持好吧啦網。

標簽: Java
相關文章:
主站蜘蛛池模板: 91精品久久久久久久久久入口 | 国产1区2区3区 | 麻豆av网| 中文字幕一区二区三区四区五区 | 在线看无码的免费网站 | av手机在线免费观看 | 亚洲最大av网站 | 欧美一区二区视频 | 91久久| 视频一区在线 | 欧美一区二区三区在线看 | 久久久99精品免费观看 | 天天操天天拍 | 国产精品久久久久久久久久免费看 | 久久在线免费 | 操到爽 | 91av视频在线 | 日韩欧美一区二区三区免费观看 | 亚洲精品一区二区三区蜜桃久 | 91精品国产91 | 玖玖国产 | 黄色一级毛片免费看 | 成人av一区 | 欧美性一区二区三区 | 欧美日韩在线观看一区 | 久久久久久综合 | 久久精品久久久久久 | 欧美电影免费网站 | 国产在线精品一区二区三区 | 97国产爽爽爽久久久 | 亚洲一区成人 | 99re在线视频| 亚洲午夜精品久久久久久app | 欧美精品福利视频 | 免费人成在线观看网站 | 日韩精品视频一区二区三区 | 免费看国产精品视频 | 国产精品视频一区二区三区四蜜臂 | 国产一级毛片视频 | 国产福利资源在线 | 一级爱爱片 |