首页 > 其他分享> 文章详细

Apache Kafka-消费端_批量消费消息的核心参数及功能实现

2021-06-04 14:52:18 阅读：143 来源： 互联网

标签：消费 18 springframework kafka 2021 import Apache org Kafka

文章目录

概述
参数设置
Code
源码地址

概述

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 # 每次批量发送消息的最大数量   单位 字节  默认 16K
      buffer-memory: 33554432 # 每次批量发送消息的最大内存 单位 字节  默认 32M
      properties:
        linger:
          ms: 10000 # 批处理延迟时间上限。[实际不会配这么长，这里用于测速]这里配置为 10 * 1000 ms 过后，不管是否消息数量是否到达 batch-size 或者消息大小到达 buffer-memory 后，都直接发送一次请求。
    # Kafka Consumer 配置项
    consumer:
      auto-offset-reset: earliest # 设置消费者分组最初的消费进度为 earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring:
          json:
            trusted:
              packages: com.artisan.springkafka.domain
      fetch-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-kafka
      apache:
        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 {


    @Autowired
    private 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());


    @Autowired
    private ArtisanProducerMock artisanProducerMock;



    @Test
    public void testAsynSend() throws ExecutionException, InterruptedException {
        logger.info("开始发送");

        for (int i = 0; i < 2; i++) {
            artisanProducerMock.sendMsgASync().addCallback(new ListenableFutureCallback<SendResult<Object, Object>>() {
                @Override
                public void onFailure(Throwable throwable) {
                    logger.info(" 发送异常{}]]", throwable);

                }
                @Override
                public 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:  10000
            TimeUnit.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

标签：消费,18,springframework,kafka,2021,import,Apache,org,Kafka
来源： https://blog.51cto.com/u_15239532/2858257

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

Apache Kafka-消费端_批量消费消息的核心参数及功能实现

文章目录

POM依赖

配置文件

生产者

消费者

单元测试

测试结果