關(guān)于kafka發(fā)送消息的三種方式總結(jié)
更新時(shí)間：2023年04月06日 09:44:18 作者：大佬喝可樂(lè)丶
這篇文章主要介紹了關(guān)于kafka發(fā)送消息的三種方式總結(jié)，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教
kafka發(fā)送消息的方式
package com.zl.kafkademo;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.quartz.*;
import org.quartz.impl.StdSchedulerFactory;
import java.util.Properties;
/**
* @Auther: le
* @Date: 2019/4/23 22:05
* @Description:
*/
public class MyProducer implements Job {
private static KafkaProducer<String,String> producer;
static {
Properties properties = new Properties();
properties.put("bootstrap.servers","127.0.0.1:9092");
properties.put("key.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
properties.put("value.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
producer = new KafkaProducer<String, String>(properties);
}
/**
* 第一種直接發(fā)送，不管結(jié)果
*/
private static void sendMessageForgetResult(){
ProducerRecord<String,String> record = new ProducerRecord<String,String>(
"kafka-study","name","Forget_result"
);
producer.send(record);
producer.close();
}
/**
* 第二種同步發(fā)送，等待執(zhí)行結(jié)果
* @return
* @throws Exception
*/
private static RecordMetadata sendMessageSync() throws Exception{
ProducerRecord<String,String> record = new ProducerRecord<String,String>(
"kafka-study","name","sync"
);
RecordMetadata result = producer.send(record).get();
System.out.println(result.topic());
System.out.println(result.partition());
System.out.println(result.offset());
return result;
}
/**
* 第三種執(zhí)行回調(diào)函數(shù)
*/
private static void sendMessageCallback(){
ProducerRecord<String,String> record = new ProducerRecord<String,String>(
"kafka-study","name","callback"
);
producer.send(record,new MyProducerCallback());
}
//定時(shí)任務(wù)
@Override
public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
try {
sendMessageSync();
}catch (Exception e){
System.out.println("error:"+e);
}
}
private static class MyProducerCallback implements Callback{
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (e !=null){
e.printStackTrace();
return;
}
System.out.println(recordMetadata.topic());
System.out.println(recordMetadata.partition());
System.out.println(recordMetadata.offset());
System.out.println("Coming in MyProducerCallback");
}
}
public static void main(String[] args){
//sendMessageForgetResult();
//sendMessageCallback();
JobDetail job = JobBuilder.newJob(MyProducer.class).build();
Trigger trigger = TriggerBuilder.newTrigger()
.withSchedule(SimpleScheduleBuilder.repeatSecondlyForever()).build();
try {
Scheduler scheduler = StdSchedulerFactory.getDefaultScheduler();
scheduler.scheduleJob(job,trigger);
scheduler.start();
}catch (SchedulerException e){
e.printStackTrace();
} catch (Exception e) {
e.printStackTrace();
}
}
}需要引入文件
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.10.0.1</version>
</dependency>
<dependency>
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
<version>2.3.0</version>
</dependency>測(cè)試方法
MAC下操作指令
1、創(chuàng)建主題：
./kafka-topics.sh --create --topic kafka-study --zookeeper 127.0.0.1:2181 --config max.message.bytes=12800000 --config flush.messages=1 --partitions 5 --replication-factor 1
2、運(yùn)行上述程序，執(zhí)行定時(shí)任務(wù)
3、查看消費(fèi)情況
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafka-study --from-beginning
windows操作指令
1、進(jìn)入 D:\zookeeper-3.4.14\bin 打開(kāi)新的cmd，輸入“zkServer“，運(yùn)行Zookeeper
2、進(jìn)入 D:\kafka_2.11-0.11.0.0 運(yùn)行cmd
.\bin\windows\kafka-server-start.bat .\config\server.properties
3、 創(chuàng)建主題
進(jìn)入D:\kafka_2.11-0.11.0.0運(yùn)行cmd，輸入：
.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
查看已創(chuàng)建主題：
.\bin\windows\kafka-topics.bat --list --zookeeper localhost:2181
查看指定主題的詳細(xì)信息：
.\bin\windows\kafka-topics.bat --describe --zookeeper localhost:2181 --topic test
查看主題消費(fèi)詳情：
.\bin\windows\kafka-console-consumer.bat --zookeeper localhost:2181 --topic kafka-study --from-beginning
總結(jié)
以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。
相關(guān)文章
在Java編程中,獲取堆棧信息對(duì)于調(diào)試和故障排除非常重要,Java提供了多種方式來(lái)獲取當(dāng)前線程的堆棧信息,下面就跟隨小編一起學(xué)習(xí)一下常用的三種吧2024-03-03
java編程中實(shí)現(xiàn)調(diào)用js方法分析
這篇文章主要介紹了java編程中實(shí)現(xiàn)調(diào)用js方法,結(jié)合具體實(shí)例形式較為詳細(xì)的分析了java編程中調(diào)用js方法的常用操作技巧與注意事項(xiàng),需要的朋友可以參考下2017-09-09
Java實(shí)現(xiàn)一個(gè)簡(jiǎn)易版的多級(jí)菜單功能
這篇文章主要給大家介紹了關(guān)于Java如何實(shí)現(xiàn)一個(gè)簡(jiǎn)易版的多級(jí)菜單功能的相關(guān)資料,文中通過(guò)實(shí)例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2022-01-01
大家好，本篇文章主要講的是java基礎(chǔ)之注解示例詳解，感興趣的同學(xué)趕快來(lái)看一看吧，對(duì)你有幫助的話記得收藏一下，方便下次瀏覽2021-12-12
解決在Gradle/IDEA中無(wú)法正常使用readLine的問(wèn)題原因
這篇文章主要介紹了在Gradle/IDEA中無(wú)法正常使用readLine的解決方法,原因是由于Gradle的標(biāo)準(zhǔn)輸入默認(rèn)并不與系統(tǒng)標(biāo)準(zhǔn)輸入綁定,需手動(dòng)設(shè)置，需要的朋友可以參考下2021-12-12
下面小編就為大家?guī)?lái)一篇JAVA利用泛型返回類型不同的對(duì)象方法。小編覺(jué)得挺不錯(cuò)的，現(xiàn)在就分享給大家，也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧2017-02-02
Spring內(nèi)置定時(shí)任務(wù)調(diào)度@Scheduled使用詳解
這篇文章主要介紹了Spring內(nèi)置定時(shí)任務(wù)調(diào)度@Scheduled使用詳解,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-12-12
Java實(shí)現(xiàn)字符編碼轉(zhuǎn)換(utf-8/gbk)
這篇文章主要為大家詳細(xì)介紹了如何使用Java實(shí)現(xiàn)字符編碼轉(zhuǎn)換工具,主要針對(duì)UTF-8和GBK兩種編碼格式,文中的示例代碼講解詳,需要的可以了解下2025-03-03
這篇文章主要介紹了Java CPU性能分析工具代碼實(shí)例,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-01-01
SpringBoot基于AbstractRoutingDataSource實(shí)現(xiàn)多數(shù)據(jù)源動(dòng)態(tài)切換
本文主要介紹了SpringBoot基于AbstractRoutingDataSource實(shí)現(xiàn)多數(shù)據(jù)源動(dòng)態(tài)切換，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2022-05-05