【kafka系列】kafka之生产者发送消息实践
目录
- 一、准备工作
- 二、终端命令
- 生产者命令
- 消费者命令
- 三、Java实践
- 搭建项目
- 异步发送-无回调
- 异步发送-有回调
- 同步发送
一、准备工作
进入实战之前先熟悉一下topic
的相关命令,使用终端命令查询创建一个新topic,用于后期实战;
特别注意:以下命令全部依据kafka文件目录中操作;
如果尚未安装kafka,请移步《centos7系统安装kafka》
- 查看操作主题命令参数
命令:./bin/kafka-topics.sh
日常工作中常用的命令如下表:
参数 | 描述 |
---|---|
–bootstrap-server <String: server toconnect to> | 连接的 Kafka Broker 主机名称和端口号。 |
–topic <String: topic> | 操作的 topic 名称。 |
–create | 创建主题 |
–delete | 删除主题 |
–alter | 修改主题 |
–list | 查看所有主题 |
–describe | 查看主题详细描述 |
–partitions <Integer: # of partitions> | 设置分区数。 |
–replication-factor<Integer: replication factor> | 设置分区副本。 |
–config <String: name=value> | 更新系统默认的配置。 |
查看当前服务器中的所有topic
命令:bin/kafka-topics.sh --list --bootstrap-server localhost:9092
创建topic
命令:bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --partitions 1 --replication-factor 1 --topic kafka-test
选项说明:
- –topic 定义 topic 名
- –replication-factor 定义副本数
- –partitions 定义分区数
查看topic详情
命令:bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic kafka-test --describe
修改topic分区
命令:bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic kafka-test --partitions 3
注意:kafka一旦增加分区号,不支持减少分区数;
二、终端命令
生产者命令
- 查看操作者命令参数
命令:bin/kafka-console-producer.sh
参数 | 描述 |
---|---|
–bootstrap-server <String: server toconnect to> | 连接的 Kafka Broker 主机名称和端口号。 |
–topic <String: topic> | 操作的 topic 名称。 |
- 生产者发送消息
命令:bin/kafka-console-producer.sh --broker-list localhost:9092 --topic kafka-test
消费者命令
- 查看操作消费者命令参数
命令:bin/kafka-console-consumer.sh
参数 | 描述 |
---|---|
–bootstrap-server <String: server toconnect to> | 连接的 Kafka Broker 主机名称和端口号。 |
–topic <String: topic> | 操作的 topic 名称。 |
–from-beginning | 从头开始消费。 |
–group <String: consumer group id> | 指定消费者组名称。 |
消费者消费消息
- 消费主题中的消息
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafka-test
- 主题中所有的数据都读取出来
包括历史数据
。
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic kafka-test
执行结果如图:
三、Java实践
正式进入生产者代码实践之前,首先列举出生产者方大致的参数列表如下:
参数 | 解释说明 |
---|---|
bootstrap.servers | 生产者连接集群所需的 broker 地 址 清 单 。 例 如hadoop102:9092,hadoop103:9092,hadoop104:9092,可以设置 1 个或者多个,中间用逗号隔开。注意这里并非需要所有的 broker 地址,因为生产者从给定的 broker里查找到其他 broker 信息。 |
key.serializer 和 value.serializer | 指定发送消息的 key 和 value 的序列化类型。一定要写全类名。 |
buffer.memory | RecordAccumulator 缓冲区总大小,默认 32m。 |
batch.size | 缓冲区一批数据最大值,默认 16k。适当增加该值,可以提高吞吐量,但是如果该值设置太大,会导致数据传输延迟增加。 |
linger.ms | 如果数据迟迟未达到 batch.size,sender 等待 linger.time之后就会发送数据。单位 ms,默认值是 0ms,表示没有延迟。生产环境建议该值大小为 5-100ms 之间。 |
acks | 0:生产者发送过来的数据,不需要等数据落盘应答。1:生产者发送过来的数据,Leader 收到数据后应答。-1(all):生产者发送过来的数据,Leader+和 isr 队列里面的所有节点收齐数据后应答。默认值是-1,-1 和all 是等价的。 |
max.in.flight.requests.per.connection | 允许最多没有返回 ack 的次数,默认为 5,开启幂等性要保证该值是 1-5 的数字。 |
retries | 当消息发送出现错误的时候,系统会重发消息。retries表示重试次数。默认是 int 最大值,2147483647。如果设置了重试,还想保证消息的有序性,需要设置MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1否则在重试此失败消息的时候,其他的消息可能发送成功了。 |
retry.backoff.ms | 两次重试之间的时间间隔,默认是 100ms。 |
enable.idempotence | 是否开启幂等性,默认 true,开启幂等性。 |
compression.type | 生产者发送的所有数据的压缩方式。默认是 none,也就是不压缩。支持压缩类型:none、gzip、snappy、lz4 和 zstd。 |
搭建项目
- 1、引入kafka依赖
<dependencies><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.0.0</version></dependency>
</dependencies>
- 2、打开消费者监听topic命令
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafka-test
异步发送-无回调
代码如下:
public static void main(String[] args) {// 1. 创建kafka生产者配置对象Properties properties = new Properties();// 2. 给 kafka 配置对象添加配置信息:bootstrap.serversproperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "XXXXXXX:9092");// key,value 序列化(必须):key.serializer,value.serializerproperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");// 3. 创建 kafka 生产者对象KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);// 4. 调用 send 方法,异步发送消息,无回调接口for (int i = 0; i < 5; i++) {kafkaProducer.send(new ProducerRecord<>("kafka-test", "producerAsyncMessage-" + i));}// 5. 关闭资源kafkaProducer.close();}
执行结果如图:
异步发送-有回调
public static void main(String[] args) throws InterruptedException {// 1. 创建Kafka生产者的配置对象Properties properties = new Properties();// 2. 给kafka配置对象添加配置信息properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "xxxxxx:9092");// 3. key 序列化 key.serializer,value.serializerproperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 4. value 序列化 value.serializerproperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 5. 创建kafka生产者对象KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String, String>(properties);for (int i = 0; i < 5; i++) {kafkaProducer.send(new ProducerRecord<>("kafka-test", "asyncCallbackMessage-" + i), new Callback() {// 该方法在Producer 收到 ack 时调用,为异步调用@Overridepublic void onCompletion(RecordMetadata recordMetadata, Exception e) {if (e == null) {// 没有异常,输出信息到控制台System.out.println("topic:" + recordMetadata.topic() + " -> " + " 分区 " + recordMetadata.partition());}else {e.printStackTrace();}}});// 延迟发送是否会发送到不同分区Thread.sleep(2);}// 5. 关闭资源kafkaProducer.close();}
消费者接收消息结果
生产者接收回调结果
同步发送
public static void main(String[] args) throws ExecutionException, InterruptedException {// 1. 创建kafka生产者配置对象Properties properties = new Properties();// 2. 给 kafka 配置对象添加配置信息:bootstrap.serversproperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "xxxxxx:9092");// key,value 序列化(必须):key.serializer,value.serializerproperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");// 3. 创建 kafka 生产者对象KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);// 4. 调用 send 方法,发送消息for (int i = 0; i < 5; i++) {RecordMetadata recordMetadata = kafkaProducer.send(new ProducerRecord<>("kafka-test", "syncMessage-" + i)).get();System.out.println("send response result,topic:" + recordMetadata.topic() + ", " + " 分区:" + recordMetadata.partition());}// 5. 关闭资源kafkaProducer.close();}
- 消费者接收消息结果
[root@VM-4-8-centos kafka]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafka-test
producerAsyncMessage-0
producerAsyncMessage-1
producerAsyncMessage-2
producerAsyncMessage-3
producerAsyncMessage-4
asyncCallbackMessage-0
asyncCallbackMessage-1
asyncCallbackMessage-2
asyncCallbackMessage-3
asyncCallbackMessage-4
syncMessage-0
syncMessage-1
syncMessage-2
syncMessage-3
syncMessage-4
- 生产者接收发送结果
【kafka系列】kafka之生产者发送消息实践相关推荐
- Kafka系列 —— Kafka监控
Kafka系列文章: Kafka系列 -- 入门及应用场景 & 部署 & 简单测试 Kafka系列 -- Kafka核心概念 Kafka系列 -- Kafka常用命令 常见Kafka监 ...
- Kafka生产者发送消息的三种方式
Kafka是一种分布式的基于发布/订阅的消息系统,它的高吞吐量.灵活的offset是其它消息系统所没有的. Kafka发送消息主要有三种方式: 1.发送并忘记 2.同步发送 3.异步发送+回调函数 下 ...
- 线上问题-kafka生产者发送消息总是失败
问题解决过程 线上一个界面发现老是没有数据,排查下来时生产者没有成功发送消息所致,报错如下: org.springframework.kafka.core.KafkaProducerException ...
- 10 kafka生产者发送消息的原理
1.发送原理: 在消息发送的过程中,涉及到了两个线程--main 线程和 Sender 线程.在 main 线程 中创建了一个双端队列 RecordAccumulator.main 线程将消息发送给 ...
- kafka通过控制台模拟消息发送和消息接收正常,但是通过javaAPI操作生产者发送消息不成功 消费者接收不到数据解决方案?
通过命令行工具(kafka-console-producer.sh和kafka-console-consumer.sh)是能够相互通信的,producer发布的信息consumer能够接收到. 但是j ...
- 生产者发送消息的过程?
1.Producer先连接到Broker,建立连接Connection,开启一个信道(Channel). 2.Producer声明一个交换器并设置好相关属性. 3.Producer声明一个队列并设置好 ...
- RocketMQ3.2.2生产者发送消息自动创建Topic队列数无法超过4个
问题现象 RocketMQ3.2.2版本,测试时尝试发送消息时自动创建Topic,设置了队列数量为8: producer.setDefaultTopicQueueNums(8); 同时设置broker ...
- 【Kafka消息队列】生产者发送消息流程
如何描述一条消息? 如何描述一条消息,就是在问这条消息的数据结构是什么? public class ProducerRecord<K, V> {private final String t ...
- Kafka实战 - 02 Kafka生产者发送消息至topic实现数据上报
文章目录 1. 项目背景 2. 依赖和配置 3. 生产者配置 KafkaConfiguration 4. 同步数据Topic枚举 SyncDataTopicEnum 5. 请求体 DataSyncQo ...
最新文章
- 【简洁+注释】剑指 Offer 32 - II. 从上到下打印二叉树 II
- php中mysql和mysqli_PHP中mysql和mysqli的区别
- 从零开始徒手撸一个vue的toast弹窗组件
- paip.提升性能---协程“微线程”的使用.
- sql获得当前时间 与 SQL 比较时间大小
- 采矿工具的全球与中国市场2022-2028年:技术、参与者、趋势、市场规模及占有率研究报告
- aspack脱壳脚本的编写
- 微信小程序 收藏功能实现
- 机器学习——回归模型
- 以“人”为本,精准营销
- 一文了解linux 网络协议栈(链路层)
- 哪里可以下载Holer软件包
- 实验:网络常见的9个命令
- 使用绝对定位来实现拉手网上的效果
- 什么是反射,反射能干嘛?
- 扯淡-20220918
- zjfc-1108,hdu-1222 wolf and rabbits
- 软件安装管家(2021年1月8日更新)
- 真正秒懂增量表、全量表和拉链表
- ImportError: cannot import name ‘XXXXX‘ from ‘XX‘解决方案