java kafka producer_Java实现Kafka Producer方法

﹏ヽ暗。殇╰゛Y 2022-11-03 11:15 170阅读 0赞

import org.apache.kafka.clients.producer.\*;

import java.util.Properties;

public class MyProducer \{

private static KafkaProducer producer;

static\{

Properties kfkProperties = new Properties();

kfkProperties.put("bootstrap.servers","slave1:9092");

kfkProperties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");

kfkProperties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");

producer = new KafkaProducer<>(kfkProperties);

\}

/\*\*

\* producer1 发送消息不考虑返回信息

\*/

private static void sendMessageForgotResult()\{

ProducerRecord record = new ProducerRecord<>("kafkatest","name","ForgetResult");

producer.send(record);

producer.close();

\}

/\*\*

\* producer2 发送消息同步等到发送成功

\*/

private static void sendMessageSync() throws Exception\{

ProducerRecord record = new ProducerRecord<>("kafkatest","name","Sync");

RecordMetadata result = producer.send(record).get();

System.out.println("时间戳，主题，分区，位移: " + result.timestamp() + "," + record.topic() + "," + result.partition() + "," + result.offset());

producer.close();

\}

/\*\*

\* producer3 发送消息异步回调返回消息

\*/

private static void sendMessageCallBack()\{

ProducerRecord record ;

while(true)\{

record= new ProducerRecord<>("kafkatest","name","CallBack");

producer.send(record,new MyProducerCallBack());

try\{

Thread.sleep(1000);

\}catch (Exception e)\{

e.printStackTrace();

\}

\}

/producer.close();

\}

private static class MyProducerCallBack implements Callback\{

@Override

public void onCompletion(RecordMetadata recordMetadata, Exception e) \{

if(null != e)\{

e.printStackTrace();

return;

\}

System.out.println("时间戳，主题，分区，位移: " + recordMetadata.timestamp() + ", " + recordMetadata.topic() + "," + recordMetadata.partition() + " " + recordMetadata.offset());

\}

\}

public static void main(String args\[\]) throws Exception\{

// MyProducer.sendMessageForgotResult();

// MyProducer.sendMessageSync();

MyProducer.sendMessageCallBack();

\}

\}

发表评论取消回复

表情：

评论列表（有 0 条评论，170人围观）

还没有评论，来说两句吧...

相关阅读

相关 kafka producer核心流程

引用：[https://blog.csdn.net/weixin\_40650734/article/details/106760541][https_blog.csdn.ne

偏执的太偏执、/ 2023年01月05日 12:15/ 0 赞/ 181 阅读

相关 Kafka Producer 核心知识

Kafka Producer 消息的产生非常简单，但是消息的发送过程还是比较复杂的，如图 ![watermark_type_ZmFuZ3poZW5naGVpdGk_sh

电玩女神/ 2022年11月18日 04:44/ 0 赞/ 246 阅读

相关 java kafka producer_Java实现Kafka Producer方法

import org.apache.kafka.clients.producer.\; import java.util.Properties; public class

﹏ヽ暗。殇╰゛Y/ 2022年11月03日 11:15/ 0 赞/ 171 阅读

相关 12.kafka Producer Example

生产者 Producer类用于为特定主题和可选分区创建新消息。如果使用Java，您需要为Producer和支持类包括几个包： <table style="width

刺骨的言语ヽ痛彻心扉/ 2022年07月08日 15:57/ 0 赞/ 174 阅读

相关 Kafka Producer详解(二)

文章目录一、Producer介绍同步和异步的方式调用send() 二、Producer工作原理介绍

骑猪看日落/ 2022年05月18日 07:58/ 0 赞/ 219 阅读

相关 Kafka：Producer Config 配置

阅读数：10069 Producer Configs 配置 <table> <thead> <tr> <th style="background-co

﹏ヽ暗。殇╰゛Y/ 2022年05月16日 00:46/ 0 赞/ 248 阅读

相关 Kafka Producer 发送数据

Kafka Producer 发送数据（1）生产者概览（1）不同的应用场景对消息有不同的需求，即是否允许消息丢失、重复、延迟以及吞吐量的要求。不同场景对Kafk

女爷i/ 2022年04月18日 02:25/ 0 赞/ 275 阅读

相关 Kafka之Producer

producer 发布消息 1、内容将数据转化为“消息”，并通过网络发送给Broker，每条数据被称为“消息”，每条信息表示为一个三元组，即<t

た入场券/ 2022年02月05日 03:13/ 0 赞/ 201 阅读

相关 kafka-producer参数详解

kafka--producer参数详解主要介绍下kafka的producer配置参数，只取了其中的一部分常用的，后续的有时间，也会补充一些，更多的详细参数，可以参考[《

ゝ一纸荒年。/ 2021年09月28日 07:34/ 0 赞/ 477 阅读

相关 Kafka Producer配置

java生产者配置： <table> <thead> <tr> <th>NAME</th> <th>DESCRIPTION</th> <th

ゞ浴缸里的玫瑰/ 2021年06月22日 15:36/ 0 赞/ 424 阅读