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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

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

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

异常处理

创建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;


/**
 * @author king
 * 异常处理，该消息会被接受，防止消息被需要放置在别的地方
 */
@Component
public class ErrorListener {


    @Bean
    public ConsumerAwareListenerErrorHandler listenerErrorHandler() {
        return (message, e, consumer) -> {
            MessageHeaders headers = message.getHeaders();
           /*  // 批量时获取值
            List topics = headers.get(KafkaHeaders.RECEIVED_TOPIC, List.class);
            List partitions = headers.get(KafkaHeaders.RECEIVED_PARTITION_ID, List.class);
            List offsets = headers.get(KafkaHeaders.OFFSET, List.class);*/
            System.out.println(message);
            System.out.println(e);
            System.out.println(consumer);
            return null;
        };
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

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

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

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

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

消息转发

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

 @KafkaListener(id = "rollback_test", topics = {"topic.quick.default"})
    @SendTo("topic.quick.initial")
    public Object receiveSkMessageInfo(ConsumerRecord<String, String> 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();
    
1
2
3
4
5
6
7
8
9

批量接收

在其他老版本，基本上百度搜到的都是需要创建消费工厂
现在已经不需要了，都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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

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

    /**
     * 批量消费
     * 手动提交的时候，默认是自动提交的，报错会自动回滚
     * @param records
     * @param ack
     */
    @KafkaListener(id = "rollback_default_test", topics = {"topic.quick.default"})
    public void receiveS(List<ConsumerRecord> records, Acknowledgment ack) {
        System.out.println(records);
        System.out.println("我收到了批量消息");
        System.out.println(ack);
        ack.acknowledge();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13

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

指定分区

KafkaListener监听指定区的消息

/**
     * 监听指定分区的消息
     * 

     * TopicPartition： 消息分区，partitions：指定区
     * partitionOffsets：偏移量，partitionOffsets偏移组，partition指定一个区，initialOffset偏移量
     * 例： 当前的为：1. 鉴定0,1两个区，
     * 		2. 监听2,3两个区，并设置2号区从第100条以后开始
     */
@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<ConsumerRecord> records, Acknowledgment ack) {
        System.out.println(records);
        System.out.println("我收到了批量消息");
        System.out.println(ack);
        ack.acknowledge();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

定时接收

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;

/**
 * @author king
 */

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

修改KafkaTopicConfig

将代码添加进KafkaTopicConfig

    @Resource
    private ConsumerFactory consumerFactory;
    @Bean
    public ConcurrentKafkaListenerContainerFactory delayContainerFactory() {
        ConcurrentKafkaListenerContainerFactory container = new ConcurrentKafkaListenerContainerFactory();
        container.setConsumerFactory(consumerFactory);
        //禁止自动启动
        container.setAutoStartup(false);
        return container;
    }
1
2
3
4
5
6
7
8
9
10