Apache Kafka-消费端_批量消费消息的核心参数及功能实现
文章目录
- 概述
- 参数设置
- Code
- POM依赖
- 配置文件
- 生产者
- 消费者
- 单元测试
- 测试结果
- 源码地址
概述
kafka提供了一些参数可以用于设置在消费端,用于提高消费的速度。
参数设置
https://kafka.apache.org/24/documentation.html#consumerconfigs
支持的属性 见源码 KafkaProperties#Consumer
spring.kafka.listener.type 默认Single
spring.kafka.consumer.max-poll-records
spring.kafka.consumer.fetch-min-size
spring.kafka.consumer.fetch-max-wait
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.domainfetch-max-wait: 10000 # poll 一次拉取的阻塞的最大时长,单位:毫秒。这里指的是阻塞拉取需要满足至少 fetch-min-size 大小的消息fetch-min-size: 10 # poll 一次消息拉取的最小数据量,单位:字节max-poll-records: 100 # poll 一次消息拉取的最大数量# Kafka Consumer Listener 监听器配置listener:missing-topics-fatal: false # 消费监听接口监听的主题不存在时,默认会报错。所以通过设置为 false ,解决报错type: batch # 监听器类型,默认为SINGLE ,只监听单条消息。这里我们配置 BATCH ,监听多条消息,批量消费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;import java.util.List;/*** @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(List<MessageMock> messageMocks){logger.info("【ArtisanCosumerMock接受到消息][线程:{} 消息大小:{}]", Thread.currentThread().getName(), messageMocks.size());messageMocks.forEach(messageMock -> System.out.println("ArtisanCosumerMock收到的消息:" + messageMock));}}
注意入参参数变为了 List
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;import java.util.List;/*** @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(List<MessageMock> messageMocks){logger.info("【ArtisanCosumerMockDiffConsumeGroup接受到消息][线程:{} 消息大小:{}]", Thread.currentThread().getName(), messageMocks.size());messageMocks.forEach(messageMock -> System.out.println("ArtisanCosumerMockDiffConsumeGroup收到的消息:" + 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);logger.info("发送一条结束...");}// 阻塞等待,保证消费new CountDownLatch(1).await();}}
异步发送2条消息,每次发送消息之间, sleep 5 秒,以便达到配置的 linger.ms 最大等待时长10秒。
测试结果
2021-02-18 12:13:00.201 INFO 8252 --- [ main] c.a.s.produceTest.ProduceMockTest : 开始发送
2021-02-18 12:13:05.426 INFO 8252 --- [ main] c.a.s.produceTest.ProduceMockTest : 发送一条结束...
2021-02-18 12:13:10.429 INFO 8252 --- [ main] c.a.s.produceTest.ProduceMockTest : 发送一条结束...
2021-02-18 12:13:10.442 INFO 8252 --- [ad | producer-1] c.a.s.produceTest.ProduceMockTest : 回调结果 Result = topic:[MOCK_TOPIC] , partition:[0], offset:[34]
2021-02-18 12:13:10.443 INFO 8252 --- [ad | producer-1] c.a.s.produceTest.ProduceMockTest : 回调结果 Result = topic:[MOCK_TOPIC] , partition:[0], offset:[35]
2021-02-18 12:13:10.493 INFO 8252 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【ArtisanCosumerMock接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息大小:2]
2021-02-18 12:13:10.493 INFO 8252 --- [ntainer#1-0-C-1] a.s.c.ArtisanCosumerMockDiffConsumeGroup : 【ArtisanCosumerMockDiffConsumeGroup接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1 消息大小:2]
ArtisanCosumerMockDiffConsumeGroup收到的消息:MessageMock{id=24, name='messageSendByAsync-24'}
ArtisanCosumerMockDiffConsumeGroup收到的消息:MessageMock{id=32, name='messageSendByAsync-32'}
ArtisanCosumerMock收到的消息:MessageMock{id=24, name='messageSendByAsync-24'}
ArtisanCosumerMock收到的消息:MessageMock{id=32, name='messageSendByAsync-32'}
从日志中可以看出,发送的 2条消息被 消费者批量消费了
咦 , 我们把Type改成默认值试试呢?
重新测试
观察日志
2021-02-18 12:17:59.598 INFO 7764 --- [ main] c.a.s.produceTest.ProduceMockTest : 开始发送
2021-02-18 12:18:04.776 INFO 7764 --- [ main] c.a.s.produceTest.ProduceMockTest : 发送一条结束...
2021-02-18 12:18:09.778 INFO 7764 --- [ main] c.a.s.produceTest.ProduceMockTest : 发送一条结束...
2021-02-18 12:18:09.781 INFO 7764 --- [ad | producer-1] c.a.s.produceTest.ProduceMockTest : 回调结果 Result = topic:[MOCK_TOPIC] , partition:[0], offset:[36]
2021-02-18 12:18:09.782 INFO 7764 --- [ad | producer-1] c.a.s.produceTest.ProduceMockTest : 回调结果 Result = topic:[MOCK_TOPIC] , partition:[0], offset:[37]
2021-02-18 12:18:09.837 INFO 7764 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【ArtisanCosumerMock接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息大小:1]
2021-02-18 12:18:09.837 INFO 7764 --- [ntainer#1-0-C-1] a.s.c.ArtisanCosumerMockDiffConsumeGroup : 【ArtisanCosumerMockDiffConsumeGroup接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1 消息大小:1]
ArtisanCosumerMock收到的消息:MessageMock{id=13, name='messageSendByAsync-13'}
ArtisanCosumerMockDiffConsumeGroup收到的消息:MessageMock{id=13, name='messageSendByAsync-13'}
2021-02-18 12:18:09.838 INFO 7764 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【ArtisanCosumerMock接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息大小:1]
ArtisanCosumerMock收到的消息:MessageMock{id=45, name='messageSendByAsync-45'}
2021-02-18 12:18:09.838 INFO 7764 --- [ntainer#1-0-C-1] a.s.c.ArtisanCosumerMockDiffConsumeGroup : 【ArtisanCosumerMockDiffConsumeGroup接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1 消息大小:1]
ArtisanCosumerMockDiffConsumeGroup收到的消息:MessageMock{id=45, name='messageSendByAsync-45'}
源码地址
https://github.com/yangshangwei/boot2/tree/master/springkafkaBatchSend
Apache Kafka-消费端_批量消费消息的核心参数及功能实现相关推荐
- Apache Kafka-生产者_批量发送消息的核心参数及功能实现
文章目录 概述 参数设置 Code POM依赖 配置文件 生产者 消费者 单元测试 测试结果 源码地址 概述 kafka中有个 micro batch 的概念 ,为了提高Producer 发送的性能. ...
- Apache Kafka-消费端_顺序消费的实现
文章目录 概述 Code POM依赖 配置文件 生产者 消费者 单元测试 测试结果 源码地址 概述 一个partition同一个时刻在一个consumer group中只能有一个consumer in ...
- java如何保证mq一定被消费_消费端如何保证消息队列MQ的有序消费
消息无序产生的原因 消息队列,既然是队列就能保证消息在进入队列,以及出队列的时候保证消息的有序性,显然这是在消息的生产端(Producer),但是往往在生产环境中有多个消息的消费端(Consumer) ...
- RocketMQ:消费端的消息消息队列负载均衡与重新发布机制源码解析
文章目录 前言 流程解析 总结 前言 在上一篇博客中我们了解到,PullMessageService线程主要是负责从pullRequestQueue中获得拉取消息请求并进行请求处理的. PullMes ...
- 消息中间件--RabbitMQ --- 消费端限流 -- 非常重要
什么是消费端的·限流? 假设一个场景,首先,我们Rabbitmq服务器有上万条未处理的消息,我们随便打开一个消费者客户端,会出现下面的情况: 巨量的消息瞬间全部推送过来,但是我们单个客户端无法同时处理 ...
- 电商场景下,如何处理消费过程中的重复消息?
摘要:比如一个消费订单消息,统计下单金额的微服务.若不正确处理重复消息,就会出现重复统计.那仅靠MQ能保证消息不重复吗? 本文分享自华为云社区<如何处理消费过程中的重复消息?>,作者:Ja ...
- ActiveMq消费端实现集群部署
1.问题背景 一个事件中心接收网关通过ActiveMq上报的告警事件,处理后持久化到数据库,消息模型为发布订阅模式.为了实现高可用,决定将事件中心进行集群部署,运行两个实例. 但是由于消息模型为发布/ ...
- 消息中间件--RabbitMQ ---高级特性之消费端ACK与重回队列
什么是消费端的ACK和重回队列? 消费端的手工ACK和NACK 消费端进行消费的时候,如果由于业务异常我们可以进行日志的记录,然后进行补偿 如果由于服务器宕机等严重问题,那我们就需要手工进行ACK保障 ...
- RabbitMQ 可靠性、重复消费、顺序性、消息积压解决方案
前言 为什么引入消息队列?引入 MQ 给我们解决了一些问题,但同时又引入了一些复杂的问题,这些问题是大型项目中必须解决的重点,更重要的是,面试也经常问.实际上消息队列可以说是没法百分之百保证可靠性的! ...
最新文章
- 小程序云开发获取手机号完整代码 云函数中网络请求第三方接口
- 目标检测 | 清晰易懂的SSD算法原理综述
- python3 urllib
- eccv2020 目标检测笔记
- 快速开发一个自定义 Spring Boot Starter ,希望你也会
- V-rep对UR3机械臂仿真路径规划
- mysql之case_mysql存储过程之case语句
- java.util.ConcurrentModificationException
- php公众号客服系统,公众号在线客服系统哪个好,主流客服系统评测及推荐
- 基于java语言轻量级实时风控引擎
- h5 video 手机上无法显示_手机镜象投屏到电视上全屏显示
- oracle事务默认自动提交吗_干货|事务
- 华为海思总裁凌晨邮件燃爆全国:多年备胎一夜转“正”,今后要科技自立
- python实现文件编码转换_Python实现批量转换文件编码的方法
- 数学 - 线性代数导论 - #9 Ax=b的解:存在性、解法、解的结构、解的数量
- CentOS 7.4安装postgresql96
- 在自定义类中使用HttpContext和Page等对象的方法
- 枚举smb共享期间出错_大量三星手机黑屏系统崩溃,客服:闰4月计算出错
- 大津阈值分割(OSTU)
- 百度网盘破解版,满速下载不限速【转载,亲测有效】
热门文章
- Nginx 内置 命令启动,停止和重新加载Nginx
- pytho sin(1/x)震荡间断点
- vue 只在父级容器移动_Vue易遗忘的基础复习(二)
- 102. Leetcode 198. 打家劫舍 (动态规划-打家劫舍)
- 文计笔记7:HTML与CSS
- 文巾解题 50. Pow(x, n)
- 数据中台实战(一):以B2B电商亿订为例谈数据埋点(产品经理视角)
- 线性回归实例-鸢尾花数据集
- 【LeetCode从零单排】No 191.Number of 1 Bits(考察位运算)
- 带你看android的内心世界