spring kafka使用（三）

传送门：
spring kafka使用（一）
spring kafka使用（二）

本篇文章只介绍接收方手动提交

kafka在配置文件中可以配置手动提交还是自动提交
将enable-auto-commit 改为false
添加listener.ack-mode= MANUAL_IMMEDIATE

  kafka:
    bootstrap-servers: 192.168.2.91:9090,192.168.2.91:9091,192.168.2.91:9092 # 集群的地址
	consumer:
      group-id: default
      enable-auto-commit: false  # 自动提交
      auto-commit-interval: 100 # 自动提交次数
      auto-offset-reset: earliest   #当默认的消费组启动的时候，会从默认的第一个消费组开始消费。
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    listener:
      #        RECORD, 每条记录被监听
      #        BATCH, 批量
      #        TIME,
      #        COUNT,
      #        COUNT_TIME,
      #        MANUAL, 手动通知
      #        MANUAL_IMMEDIATE; 立即手动通知
      ack-mode: MANUAL_IMMEDIATE

文件KafkaReceiver中的消息接收，新增Acknowledgment 接收字段

 @KafkaListener(id = "rollback_default_test", topics = {"topic.quick.default"})
    public void receiveSk(ConsumerRecord record, Acknowledgment ack) {
        System.out.println(record);
        System.out.println("我收到了普通消息");
        // 手动确认消息被 消费
        ack.acknowledge();
        // ack.nack(1000); 拒收当前消息，并睡眠10秒钟后再重新接收消息
        // ack.nack(100,1000); 拒收当前消息，并睡眠10秒钟后接收第100条之后的消息
    }

异常处理

创建ErrorListener

package com.kofan.server.kafkaMq.component;

import org.springframework.context.annotation.Bean;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.MessageHeaders;
import org.springframework.stereotype.Component;

import java.util.List;



@Component
public class ErrorListener {


    @Bean
    public ConsumerAwareListenerErrorHandler listenerErrorHandler() {
        return (message, e, consumer) -> {
            MessageHeaders headers = message.getHeaders();
           
            System.out.println(message);
            System.out.println(e);
            System.out.println(consumer);
            return null;
        };
    }
}

接收消息
在接收消息KafkaListener里添加 errorHandler = “listenerErrorHandler”
listenerErrorHandler就是ErrorListener文件中的异常处理方法名

    @KafkaListener(id = "rollback_default_test", topics = {"topic.quick.default"}, errorHandler = "listenerErrorHandler")
    public void receiveSk(ConsumerRecord record, Acknowledgment ack) {
        System.out.println(record);
        System.out.println("我收到了普通消息");
        ack.acknowledge();
//         ack.nack(1000);
    }

注意！创建异常方法之后，如果接收方法报错，但是该条信息还是被消费了，需要在异常方法中额外记录该消息，否则该消息可能丢失

如果没有该异常方法，则该KafkaListener会一直接收当前消息，并一直报错

消息转发

消息转发只需方法@SendTo指向topic即可

 @KafkaListener(id = "rollback_test", topics = {"topic.quick.default"})
    @SendTo("topic.quick.initial")
    public Object receiveSkMessageInfo(ConsumerRecord record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Acknowledgment ack) {
        System.out.println(record);
        System.out.println(topic);
        System.out.println(ack);
        ack.acknowledge();
        return record.value();

批量接收

在其他老版本，基本上百度搜到的都是需要创建消费工厂
现在已经不需要了，都2202年了，创建什么工厂？spring-kafka已经帮你创建了，只需要配置就可以了
配置文件：
设置max-poll-records=500，一次最多拉500条消息，设置listener.ack-mode= MANUAL，设置listener.type = batch

kafka:
    bootstrap-servers: 192.168.2.91:9090,192.168.2.91:9091,192.168.2.91:9092 # 集群的地址
    consumer:
      group-id: default
      enable-auto-commit: true  # 自动提交
      auto-commit-interval: 100 # 自动提交次数
      auto-offset-reset: earliest   #当默认的消费组启动的时候，会从默认的第一个消费组开始消费。
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      max-poll-records: 500  #一次最多拉500条消息。
    listener:
      #        RECORD, 每条记录被监听
      #        BATCH, 批量
      #        TIME,
      #        COUNT,
      #        COUNT_TIME,
      #        MANUAL, 手动通知
      #        MANUAL_IMMEDIATE; 立即手动通知
      ack-mode: MANUAL
      # batch 批量 single 单一
      type: batch

这样就可以接受批量消息了
接收方：

    
    @KafkaListener(id = "rollback_default_test", topics = {"topic.quick.default"})
    public void receiveS(List records, Acknowledgment ack) {
        System.out.println(records);
        System.out.println("我收到了批量消息");
        System.out.println(ack);
        ack.acknowledge();
    }

注意：批量消费的时候，项目中不能出现单一消费，单一消费的时候不能出现批量消费。批量消费的时候手动提交状态影响的只是是否拒收消息，如果不出现报错，就算没有 ack.acknowledge();该批消息依然是被消费了。指定分区

KafkaListener监听指定区的消息

@KafkaListener(id = "rollback_default_test", topicPartitions = {
            @TopicPartition(topic = "topic.quick.default", partitions = {"0", "1"}),
            @TopicPartition(topic = "", partitions = {"2", "3"}, partitionOffsets = {
                    @PartitionOffset(partition = "2", initialOffset = "100")}
            )
    })
    public void receiveZ(List records, Acknowledgment ack) {
        System.out.println(records);
        System.out.println("我收到了批量消息");
        System.out.println(ack);
        ack.acknowledge();
    }

定时接收

package com.kofan.server.kafkaMq.component;

import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;



@Component
@EnableScheduling
public class KafkaListenerComponent {
    @Resource
    private KafkaListenerEndpointRegistry endpointRegistry;
    //定时器,开启kafka监听
    @Scheduled(cron = "0 45 10 * * ?")
    public void startListener() {
        //判断监听容器是否启动，未启动则将其启动
        if (!endpointRegistry.getListenerContainer("rollback_default_test").isRunning()) {
            endpointRegistry.getListenerContainer("rollback_default_test").start();
        }
        endpointRegistry.getListenerContainer("rollback_default_test").resume();
    }
    //定时器，关闭kafka监听
    @Scheduled(cron = "0 50 10 * * ?")
    public void shutDownListener() {
        endpointRegistry.getListenerContainer("rollback_default_test").pause();
    }
}

修改KafkaTopicConfig

将代码添加进KafkaTopicConfig

    @Resource
    private ConsumerFactory consumerFactory;
    @Bean
    public ConcurrentKafkaListenerContainerFactory delayContainerFactory() {
        ConcurrentKafkaListenerContainerFactory container = new ConcurrentKafkaListenerContainerFactory();
        container.setConsumerFactory(consumerFactory);
        //禁止自动启动
        container.setAutoStartup(false);
        return container;
    }

spring kafka使用（三）

Java相关栏目本月热门文章