Apache Kafka-生产者_批量发送消息的核心参数及功能实现
文章目录
- 概述
- 参数设置
- Code
- POM依赖
- 配置文件
- 生产者
- 消费者
- 单元测试
- 测试结果
- 源码地址
概述
kafka中有个 micro batch 的概念 ,为了提高Producer 发送的性能。
不同于RocketMQ 提供了一个可以批量发送多条消息的 API 。 Kafka 的做法是:提供了一个 RecordAccumulator 消息收集器,将发送给相同 Topic 的相同 Partition 分区的消息们,缓冲一下,当满足条件时候,一次性批量将缓冲的消息提交给 Kafka Broker 。
参数设置
https://kafka.apache.org/24/documentation.html#producerconfigs
主要涉及的参数 ,三个条件,满足任一即会批量发送:
- batch-size :超过收集的消息数量的最大量。默认16KB
- buffer-memory :超过收集的消息占用的最大内存 , 默认32M
- linger.ms :超过收集的时间的最大等待时长,单位:毫秒。
Code
POM依赖
<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- 引入 Spring-Kafka 依赖 --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><scope>test</scope></dependency></dependencies>
配置文件
spring:# Kafka 配置项,对应 KafkaProperties 配置类kafka:bootstrap-servers: 192.168.126.140:9092 # 指定 Kafka Broker 地址,可以设置多个,以逗号分隔# Kafka Producer 配置项producer:acks: 1 # 0-不应答。1-leader 应答。all-所有 leader 和 follower 应答。retries: 3 # 发送失败时,重试发送的次数key-serializer: org.apache.kafka.common.serialization.StringSerializer # 消息的 key 的序列化value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 消息的 value 的序列化batch-size: 16384 # 每次批量发送消息的最大数量 单位 字节 默认 16Kbuffer-memory: 33554432 # 每次批量发送消息的最大内存 单位 字节 默认 32Mproperties:linger:ms: 10000 # 批处理延迟时间上限。[实际不会配这么长,这里用于测速]这里配置为 10 * 1000 ms 过后,不管是否消息数量是否到达 batch-size 或者消息大小到达 buffer-memory 后,都直接发送一次请求。# Kafka Consumer 配置项consumer:auto-offset-reset: earliest # 设置消费者分组最初的消费进度为 earliestkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.springframework.kafka.support.serializer.JsonDeserializerproperties:spring:json:trusted:packages: com.artisan.springkafka.domain# Kafka Consumer Listener 监听器配置listener:missing-topics-fatal: false # 消费监听接口监听的主题不存在时,默认会报错。所以通过设置为 false ,解决报错logging:level:org:springframework:kafka: ERROR # spring-kafkaapache:kafka: ERROR # kafka
生产者
package com.artisan.springkafka.producer;import com.artisan.springkafka.constants.TOPIC;
import com.artisan.springkafka.domain.MessageMock;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;import java.util.Random;
import java.util.concurrent.ExecutionException;/*** @author 小工匠* @version 1.0* @description: TODO* @date 2021/2/17 22:25* @mark: show me the code , change the world*/@Component
public class ArtisanProducerMock {@Autowiredprivate KafkaTemplate<Object,Object> kafkaTemplate ;/*** 同步发送* @return* @throws ExecutionException* @throws InterruptedException*/public SendResult sendMsgSync() throws ExecutionException, InterruptedException {// 模拟发送的消息Integer id = new Random().nextInt(100);MessageMock messageMock = new MessageMock(id,"artisanTestMessage-" + id);// 同步等待return kafkaTemplate.send(TOPIC.TOPIC, messageMock).get();}public ListenableFuture<SendResult<Object, Object>> sendMsgASync() throws ExecutionException, InterruptedException {// 模拟发送的消息Integer id = new Random().nextInt(100);MessageMock messageMock = new MessageMock(id,"messageSendByAsync-" + id);// 异步发送消息ListenableFuture<SendResult<Object, Object>> result = kafkaTemplate.send(TOPIC.TOPIC, messageMock);return result ;}}
消费者
package com.artisan.springkafka.consumer;import com.artisan.springkafka.domain.MessageMock;
import com.artisan.springkafka.constants.TOPIC;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;/*** @author 小工匠* @version 1.0* @description: TODO* @date 2021/2/17 22:33* @mark: show me the code , change the world*/@Component
public class ArtisanCosumerMock {private Logger logger = LoggerFactory.getLogger(getClass());private static final String CONSUMER_GROUP_PREFIX = "MOCK-A" ;@KafkaListener(topics = TOPIC.TOPIC ,groupId = CONSUMER_GROUP_PREFIX + TOPIC.TOPIC)public void onMessage(MessageMock messageMock){logger.info("【接受到消息][线程:{} 消息内容:{}]", Thread.currentThread().getName(), messageMock);}}
package com.artisan.springkafka.consumer;import com.artisan.springkafka.domain.MessageMock;
import com.artisan.springkafka.constants.TOPIC;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;/*** @author 小工匠* @version 1.0* @description: TODO* @date 2021/2/17 22:33* @mark: show me the code , change the world*/@Component
public class ArtisanCosumerMockDiffConsumeGroup {private Logger logger = LoggerFactory.getLogger(getClass());private static final String CONSUMER_GROUP_PREFIX = "MOCK-B" ;@KafkaListener(topics = TOPIC.TOPIC ,groupId = CONSUMER_GROUP_PREFIX + TOPIC.TOPIC)public void onMessage(MessageMock messageMock){logger.info("【接受到消息][线程:{} 消息内容:{}]", Thread.currentThread().getName(), messageMock);}}
单元测试
package com.artisan.springkafka.produceTest;import com.artisan.springkafka.SpringkafkaApplication;
import com.artisan.springkafka.producer.ArtisanProducerMock;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.support.SendResult;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;/*** @author 小工匠* * @version 1.0* @description: TODO* @date 2021/2/17 22:40* @mark: show me the code , change the world*/@RunWith(SpringRunner.class)
@SpringBootTest(classes = SpringkafkaApplication.class)
public class ProduceMockTest {private Logger logger = LoggerFactory.getLogger(getClass());@Autowiredprivate ArtisanProducerMock artisanProducerMock;@Testpublic void testAsynSend() throws ExecutionException, InterruptedException {logger.info("开始发送");for (int i = 0; i < 2; i++) {artisanProducerMock.sendMsgASync().addCallback(new ListenableFutureCallback<SendResult<Object, Object>>() {@Overridepublic void onFailure(Throwable throwable) {logger.info(" 发送异常{}]]", throwable);}@Overridepublic void onSuccess(SendResult<Object, Object> objectObjectSendResult) {logger.info("回调结果 Result = topic:[{}] , partition:[{}], offset:[{}]",objectObjectSendResult.getRecordMetadata().topic(),objectObjectSendResult.getRecordMetadata().partition(),objectObjectSendResult.getRecordMetadata().offset());}});// 发送2次 每次间隔5秒, 凑够我们配置的 linger: ms: 10000TimeUnit.SECONDS.sleep(5);}// 阻塞等待,保证消费new CountDownLatch(1).await();}}
异步发送2条消息,每次发送消息之间, sleep 5 秒,以便达到配置的 linger.ms 最大等待时长10秒。
测试结果
2021-02-18 10:58:53.360 INFO 24736 --- [ main] c.a.s.produceTest.ProduceMockTest : 开始发送
2021-02-18 10:59:03.555 INFO 24736 --- [ad | producer-1] c.a.s.produceTest.ProduceMockTest : 回调结果 Result = topic:[MOCK_TOPIC] , partition:[0], offset:[30]
2021-02-18 10:59:03.556 INFO 24736 --- [ad | producer-1] c.a.s.produceTest.ProduceMockTest : 回调结果 Result = topic:[MOCK_TOPIC] , partition:[0], offset:[31]
2021-02-18 10:59:03.595 INFO 24736 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容:MessageMock{id=6, name='messageSendByAsync-6'}]
2021-02-18 10:59:03.595 INFO 24736 --- [ntainer#1-0-C-1] a.s.c.ArtisanCosumerMockDiffConsumeGroup : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1 消息内容:MessageMock{id=6, name='messageSendByAsync-6'}]
2021-02-18 10:59:03.595 INFO 24736 --- [ntainer#1-0-C-1] a.s.c.ArtisanCosumerMockDiffConsumeGroup : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1 消息内容:MessageMock{id=94, name='messageSendByAsync-94'}]
2021-02-18 10:59:03.595 INFO 24736 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容:MessageMock{id=94, name='messageSendByAsync-94'}]
10 秒后,满足批量消息的最大等待时长,所以 2 条消息被 Producer 批量发送。同时我们配置的是 acks=1 ,需要等待发送成功后,才会回调 ListenableFutureCallback 的方法。
当然了,我们这里都是为了测试,设置的这么长的间隔,实际中需要根据具体的业务场景设置一个合理的值。
源码地址
https://github.com/yangshangwei/boot2/tree/master/springkafkaBatchSend
Apache Kafka-生产者_批量发送消息的核心参数及功能实现相关推荐
- Apache Kafka-消费端_批量消费消息的核心参数及功能实现
文章目录 概述 参数设置 Code POM依赖 配置文件 生产者 消费者 单元测试 测试结果 源码地址 概述 kafka提供了一些参数可以用于设置在消费端,用于提高消费的速度. 参数设置 https: ...
- Kafka生产者是如何发送消息的?
(一)生产者的原理 当有数据要从生产者发往消费者的时候,在kafka底层有这样一套流程.首先生产者调用send方法发送消息后,会先经过一层拦截器,接着进入序列化器.序列化器主要用于对消息的Key和Va ...
- java/kafka的使用,生产者 ,发布/发送消息,自定义数据/json数据,实战。
kafka(卡夫卡生产者)的使用. 发送JSON数据请查看:kafka发送JSON消息,接受JSON消息 一,创建生产者 MyProducer.java . package com.kuxingsen ...
- python微信自动群发脚本_python 微信批量发送消息脚本
通过python 利用微信公众号,批量发送消息 但是,因为免费的,一天一个用户,只能发90条消息... 可以用于zabbix啊,.等监控 #!/usr/bin/env python # -*- cod ...
- RabbitMq消息中心_延迟发送消息
Rabbitmq延迟发送消息解决方案 一. 什么是延迟队列: 延迟队列存储的对象肯定是对应的延时消息,所谓"延时消息"是指当消息被发送以后,并不想让消费者立即拿到消息,而是等待指定 ...
- Android发送消息的核心代码,Android 抖音 发消息Call 调用 实现群发消息 代码hook源码...
* 注意:发送消息前,需要先获取[通讯录好友列表]再调用发送消息Call //hook版 public void sendMsg( String userid){ try { //cl_onCreat ...
- Python实现微信批量发送消息
注意需要先打开微信 import win32clipboard as w import win32con import win32api import win32gui import time# 读取 ...
- 1 partitions have leader brokers without a matching listener, including [baidd-0] (org.apache.kafka.
一 故障描述 9月22日,全国kafka集群中的其中一台kafka因磁盘空间不足宕机后,业务会受到影响,无法生产与消费消息.程序报错: WARN [Consumer clientId=consumer ...
- kafka 发布-订阅模式_使用Apache Kafka作为消息系统的发布-订阅通信中的微服务,并通过集成测试进行了验证...
kafka 发布-订阅模式 发布-订阅消息系统在任何企业体系结构中都起着重要作用,因为它可以实现可靠的集成而无需紧密耦合应用程序. 在解耦的系统之间共享数据的能力并不是一个容易解决的问题. 考虑一个企 ...
最新文章
- Java调用Python遇到的一系列问题与解决方案
- 简单的 H5 视频推流解决方案
- Django路由介绍
- Qt Creator管理数据收集设置
- B程序员:讲述三年计算机学习辛酸史
- 曲线相似度衡量——曲线距离计算Fréchet distance详解与python计算
- java程序设计_Java程序设计-Object类(笔记)
- @Target({ElementType.METHOD, ElementType.TYPE})
- Class的三种构造方法
- 2B or 2c : 做2019的幸存者
- 矩阵分析与应用(一)
- python按键精灵是哪个库_按键精灵命令库的介绍和使用
- django 序列化组件Serializer
- 阅读这篇文章,假设你不知道的傅里叶变换,然后来掐死我
- 从数据库导出数据到EXCEL换行的问题解决方法(即数据库字段太长自动换行)
- ios微信组件跳转_在iOS9中突破微信App跳转的限制
- html5新标签 figure 和 figcaption
- 336高校毕业设计选题
- 条码应用与企业ERP无缝集成
- 你写注释她帮你写代码