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

您的位置:首頁(yè)技術(shù)文章
文章詳情頁(yè)

Java實(shí)現(xiàn)Kafka生產(chǎn)者消費(fèi)者代碼實(shí)例

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

Kafka的結(jié)構(gòu)與RabbitMQ類(lèi)似,消息生產(chǎn)者向Kafka服務(wù)器發(fā)送消息,Kafka接收消息后,再投遞給消費(fèi)者。生產(chǎn)者的消費(fèi)會(huì)被發(fā)送到Topic中,Topic中保存著各類(lèi)數(shù)據(jù),每一條數(shù)據(jù)都使用鍵、值進(jìn)行保存。每一個(gè)Topic中都包含一個(gè)或多個(gè)物理分區(qū)(Partition),分區(qū)維護(hù)著消息的內(nèi)容和索引,它們有可能被保存在不同服務(wù)器。

新建一個(gè)Maven項(xiàng)目,pom.xml 加入依賴(lài):

<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.3.0</version> </dependency>

1、編寫(xiě)生產(chǎn)者

將消息投遞到Kafka服務(wù)器的名稱(chēng)為“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的序列化處理類(lèi) props.put('key.serializer', StringSerializer.class); props.put('value.serializer', StringSerializer.class); //創(chuàng)建生產(chǎn)者實(shí)例 KafkaProducer<String,String> producer = new KafkaProducer<>(props); ProducerRecord record = new ProducerRecord<String, String>('topic1', 'userName', 'lc'); //發(fā)送記錄 producer.send(record); producer.close(); }}

運(yùn)行后,可打開(kāi)命令行工具,進(jìn)入Kafka目錄,執(zhí)行命令查詢(xún)服務(wù)器的Topic:

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

結(jié)果如下:

Java實(shí)現(xiàn)Kafka生產(chǎn)者消費(fèi)者代碼實(shí)例

2、編寫(xiě)消費(fèi)者

本例中,消費(fèi)者和生產(chǎn)者在同一個(gè)項(xiàng)目中,只是使用不同的啟動(dòng)類(lèi)。

消費(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的序列化處理類(lèi) props.put('key.deserializer', StringDeserializer.class); props.put('value.deserializer', StringDeserializer.class); //創(chuàng)建消息者實(shí)例 KafkaConsumer<String,String> consumer = new KafkaConsumer<>(props); //訂閱topic1的消息 consumer.subscribe(Arrays.asList('topic1')); //到服務(wù)器中讀取記錄 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()); } } }}

運(yùn)行后,IDEA控制臺(tái)其中輸出如下:

Java實(shí)現(xiàn)Kafka生產(chǎn)者消費(fèi)者代碼實(shí)例

以上就是本文的全部?jī)?nèi)容,希望對(duì)大家的學(xué)習(xí)有所幫助,也希望大家多多支持好吧啦網(wǎng)。

標(biāo)簽: Java
相關(guān)文章:
主站蜘蛛池模板: 中文字幕第5页 | 91中文字幕在线 | 色播视频在线观看 | 欧美日韩国产免费 | 二区三区视频 | 亚洲第一天堂无码专区 | 欧美日韩综合一区 | 福利网址 | 久久精品国产一区二区 | 九九久久精品 | 狠狠操电影| 蜜桃视频在线观看免费视频网站www | 日韩中文字幕网 | 欧美日韩亚洲系列 | 日韩中文字幕在线不卡 | 成人一区二区三区视频 | 国产一级在线 | 天天躁日日躁狠狠躁2018小说 | aaa在线观看 | 69av片| 亚洲黄色视屏 | 中文在线播放 | 在线观看涩涩视频 | 成av在线 | 成人三区| 美日韩中文字幕 | 国产精品精品视频 | 国产激情91久久精品导航 | 99福利视频导航 | 日韩精品成人 | 免费在线观看av片 | 超碰男人天堂 | 99精品久久99久久久久 | 亚洲成人精品 | 国产欧美日韩精品在线观看 | 国产精品视频偷伦精品视频 | 亚洲精品久久久久中文字幕二区 | 精品伊人 | 免费视频一区二区 | 黄网站在线播放 | 精品视频在线观看 |