SpringBoot-Kafka使用（三）

本是古典何须时尚 2022-03-07 06:44 293阅读 0赞

## 一、Kafka的事务 ##

### 为什么要使用Kafka事务 ###

在日常开发中，数据库的事务几乎是必须用到的，事务回滚不一定在于数据增删改异常，可能系统出现特定逻辑判断的时候也需要进行数据回滚，Kafka亦是如此，我们并不希望消息监听器接收到一些错误的或者不需要的消息。SpringBoot使用数据库事务非常简单，只需要在方法上加上@Transactional注解即可，那Kafka如果需要使用事务也可以如此，不过还需修改其他配置才能生效。

### Kafka使用事务的两种方式 ###

（一）配置Kafka事务管理器并使用@Transactional注解  
（二）使用KafkaTemplate的executeInTransaction方法

### 使用@Transactional注解方式 ###

使用注解方式开启事务还是比较方便的，不过首先需要我们配置KafkaTransactionManager，这个类就是Kafka提供给我们的事务管理类，我们需要使用生产者工厂来创建这个事务管理类。需要注意的是，我们需要在producerFactory中开启事务功能，并设置TransactionIdPrefix，TransactionIdPrefix是用来生成Transactional.id的前缀。

@Bean
        public ProducerFactory<Integer, String> producerFactory() {
            DefaultKafkaProducerFactory factory = new DefaultKafkaProducerFactory<>(senderProps());
            factory.transactionCapable();
            factory.setTransactionIdPrefix("tran-");
            return factory;
        }
    
        @Bean
        public KafkaTransactionManager transactionManager(ProducerFactory producerFactory) {
            KafkaTransactionManager manager = new KafkaTransactionManager(producerFactory);
            return manager;
        }

配置Kafka事务还是非常简单的，接下来我们测试一下事务是否能正常使用，在DemoTest类中创建该方法

@Test
        @Transactional
        public void testTransactionalAnnotation() throws InterruptedException {
            kafkaTemplate.send("topic.quick.tran", "test transactional annotation");
            throw new RuntimeException("fail");
        }

运行测试方法后我们可以看到控制台中输出了如下日志，这就代表我们使用事务成功，或者你可以打开Kafka Tool 2工具查看一下测试前后的数据是否有变化

org.apache.kafka.common.KafkaException: Failing batch since transaction was aborted
        at org.apache.kafka.clients.producer.internals.Sender.maybeSendTransactionalRequest(Sender.java:317) [kafka-clients-1.0.2.jar:na]
        at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:214) [kafka-clients-1.0.2.jar:na]
        at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:163) [kafka-clients-1.0.2.jar:na]
        at java.lang.Thread.run(Thread.java:748) [na:1.8.0_161]

### 使用KafkaTemplate.executeInTransaction开启事务 ###

这种方式开启事务是不需要配置事务管理器的，也可以称为本地事务。直接编写测试方法

@Test
        public void testExecuteInTransaction() throws InterruptedException {
            kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback() {
                @Override
                public Object doInOperations(KafkaOperations kafkaOperations) {
                    kafkaOperations.send("topic.quick.tran", "test executeInTransaction");
                    throw new RuntimeException("fail");
                    //return true;
                }
            });
        }

运行测试方法后控制台同样打印出了事务终止的异常，代表可以正常使用事务

## 二、使用非注解方式和@KafkaListener注解方式监听消息 ##

### 消费者订阅监听消息 ###

对于Kafka中Topic的数据消费，我们一般都选择使用消息监听器进行消费，怎么把消息监听器玩出花来呢，那就得看看它所实现的功能了。  
Spring-Kafka中消息监听大致分为两种类型，一种是单条数据消费，一种是批量消费；两者的区别只是在于监听器一次性获取消息的数量。GenericMessageListener是我们实现消息监听的一个接口，向上扩展的接口有非常多，比如：单数据消费的MessageListener、批量消费的BatchMessageListener、还有具备ACK机制的AcknowledgingMessageListener和BatchAcknowledgingMessageListener等等。接下来我们就一一解析一下。

### 先来了解源码中的GenericMessageListener ###

这里可以看到GenericMessageListener使用注解标明这是一个函数式接口，默认实现了三种不同参数的onMessage方法。data就是我们需要接收的数据，Consumer则是消费者类，Acknowledgment则是用来实现Ack机制的类。这里需要注意一下的是，Consumer对象并不是线程安全的。

@FunctionalInterface
    public interface GenericMessageListener<T> {
        void onMessage(T var1);
    
        default void onMessage(T data, Acknowledgment acknowledgment) {
            throw new UnsupportedOperationException("Container should never call this");
        }
    
        default void onMessage(T data, Consumer<?, ?> consumer) {
            throw new UnsupportedOperationException("Container should never call this");
        }
    
        default void onMessage(T data, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
            throw new UnsupportedOperationException("Container should never call this");
        }
    }

接下来先浏览一下继承了GenericMessageListener接口的类。前缀为Batch的接口都是批处理类型的消息监听接口，里面的参数也都讲解过了

public interface MessageListener<K, V> {
        void onMessage(ConsumerRecord<K, V> data);
    }
    
    public interface AcknowledgingMessageListener<K, V> { 
        void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment);
    }
    
    public interface ConsumerAwareMessageListener<K, V> extends MessageListener<K, V> {
        void onMessage(ConsumerRecord<K, V> data, Consumer<?, ?> consumer);
    }
    
    public interface AcknowledgingConsumerAwareMessageListener<K, V> extends MessageListener<K, V> { 
        void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment, Consumer<?, ?> consumer);
    }
    
    public interface BatchMessageListener<K, V> { 
        void onMessage(List<ConsumerRecord<K, V>> data);
    }
    
    public interface BatchAcknowledgingMessageListener<K, V> {
        void onMessage(List<ConsumerRecord<K, V>> data, Acknowledgment acknowledgment);
    }
    
    public interface BatchConsumerAwareMessageListener<K, V> extends BatchMessageListener<K, V> { 
        void onMessage(List<ConsumerRecord<K, V>> data, Consumer<?, ?> consumer);
    }
    
    public interface BatchAcknowledgingConsumerAwareMessageListener<K, V> extends BatchMessageListener<K, V> { 
        void onMessage(List<ConsumerRecord<K, V>> data, Acknowledgment acknowledgment, Consumer<?, ?> consumer);
    }

### 在解析@KafkaListener前，我们先了解怎么使用非注解方式去监听Topic ###

我们在创建监听容器前需要创建一个监听容器工厂，这里只需要配置一下消费者工厂就好了  
，之后我们使用它去创建我们的监听容器。consumerFactory()这个参数在之前就已经定义过了。

//ConcurrentKafkaListenerContainerFactory为创建Kafka监听器的工程类，这里只配置了消费者
        @Bean
        public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory());
            return factory;
        }
    
        //根据consumerProps填写的参数创建消费者工厂
        @Bean
        public ConsumerFactory<Integer, String> consumerFactory() {
            return new DefaultKafkaConsumerFactory<>(consumerProps());
        }

有了监听容器工厂，我们就可以使用它去创建我们的监听容器

Bean方式创建监听容器，在onMessage()方法中实现消息监听

@Bean
        public KafkaMessageListenerContainer demoListenerContainer() {
            ContainerProperties properties = new ContainerProperties("topic.quick.bean");
            
            properties.setGroupId("bean");
            
            properties.setMessageListener(new MessageListener<Integer,String>() {
                private Logger log = LoggerFactory.getLogger(this.getClass());
                @Override
                public void onMessage(ConsumerRecord<Integer, String> record) {
                    log.info("topic.quick.bean receive : " + record.toString());
                }
            });
    
            return new KafkaMessageListenerContainer(consumerFactory(), properties);
        }

启动项目我们可以看一下控制台的日志，监听容器成功分配给某个消费者的结果很清晰的显示出来了，顺便就写个测试方法测试一下监听器能不能正常运行。

2018-09-11 10:36:15.732  INFO 1168 --- [erContainer-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-4, groupId=bean] Successfully joined group with generation 1
    2018-09-11 10:36:15.733  INFO 1168 --- [erContainer-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-4, groupId=bean] Setting newly assigned partitions [topic.quick.bean-0]
    2018-09-11 10:36:15.733  INFO 1168 --- [erContainer-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [topic.quick.bean-0]

@Test
        public void test() {
            kafkaTemplate.send("topic.quick.bean", "send msg to beanListener");
        }

## @KafkaListener参数讲解 ##

用@KafkaListener注解实现监听的代码

@KafkaListener(id = "demo", topics = "topic.quick.demo")
        public void listen(String msgData) {
            log.info("demo receive : "+msgData);
        }

使用@KafkaListener这个注解并不局限于这个监听容器是单条数据消费还是批量消费，区分单数据还是多数据消费只需要配置一下注解的containerFactory属性即可，先讲解一下这个监听方法都能接收写什么参数吧。  
  
**data** ： 对于data值的类型其实并没有限定，根据KafkaTemplate所定义的类型来决定。data为List集合的则是用作批量消费。  
**ConsumerRecord**：具体消费数据类，包含Headers信息、分区信息、时间戳等  
**Acknowledgment**：用作Ack机制的接口  
**Consumer**：消费者类，使用该类我们可以手动提交偏移量、控制消费速率等功能

public void listen1(String data) 
    
        public void listen2(ConsumerRecord<K,V> data) 
    
        public void listen3(ConsumerRecord<K,V> data, Acknowledgment acknowledgment) 
    
        public void listen4(ConsumerRecord<K,V> data, Acknowledgment acknowledgment, Consumer<K,V> consumer) 
    
        public void listen5(List<String> data) 
    
        public void listen6(List<ConsumerRecord<K,V>> data) 
    
        public void listen7(List<ConsumerRecord<K,V>> data, Acknowledgment acknowledgment) 
    
        public void listen8(List<ConsumerRecord<K,V>> data, Acknowledgment acknowledgment, Consumer<K,V> consumer)

接下来在看看@KafkaListener的注解都提供了什么属性。

**id**：消费者的id，当GroupId没有被配置的时候，默认id为GroupId  
**containerFactory**：上面提到了@KafkaListener区分单数据还是多数据消费只需要配置一下注解的containerFactory属性就可以了，这里面配置的是监听容器工厂，也就是ConcurrentKafkaListenerContainerFactory，配置BeanName  
**topics**：需要监听的Topic，可监听多个  
**topicPartitions**：可配置更加详细的监听信息，必须监听某个Topic中的指定分区，或者从offset为200的偏移量开始监听  
**errorHandler**：监听异常处理器，配置BeanName  
**groupId**：消费组ID  
**idIsGroup**：id是否为GroupId  
**clientIdPrefix**：消费者Id前缀  
**beanRef**：真实监听容器的BeanName，需要在 BeanName前加 "\_\_"

public @interface KafkaListener {
        String id() default "";
    
        String containerFactory() default "";
    
        String[] topics() default {};
    
        String topicPattern() default "";
    
        TopicPartition[] topicPartitions() default {};
    
        String containerGroup() default "";
    
        String errorHandler() default "";
    
        String groupId() default "";
    
        boolean idIsGroup() default true;
    
        String clientIdPrefix() default "";
    
        String beanRef() default "__listener";
    }

### 使用ConsumerRecord类消费 ###

用ConsumerRecord类接收的好处是什么呢，ConsumerRecord类里面包含分区信息、消息头、消息体等内容，如果业务需要获取这些参数时，使用ConsumerRecord会是个不错的选择。如果使用具体的类型接收消息体则更加方便，比如说用String类型去接收消息体。

这里我们编写一个consumerListener方法，监听"topic.quick.consumer" Topic，并把ConsumerRecord里面所包含的内容打印到控制台中

@Component
    public class SingleListener {
    
        private static final Logger log = LoggerFactory.getLogger(SingleListener.class);
    
        @KafkaListener(id = "consumer", topics = "topic.quick.consumer")
        public void consumerListener(ConsumerRecord<Integer, String> record) {
            log.info("topic.quick.consumer receive : " + record.toString());
        }
    }

编写测试方法，发送数据到对应的Topic中，运行测试我们可以看到控制台打印的日志，日志里面包含topic、partition、offset等信息，这其实就是完整的消息储存结构。

@Test
        public void testConsumerRecord() {
            kafkaTemplate.send("topic.quick.consumer", "test receive by consumerRecord");
        }

2018-09-11 15:52:13.546  INFO 13644 --- [ consumer-0-C-1] com.viu.kafka.listen.SingleListener      : topic.quick.consumer receive : ConsumerRecord(topic = topic.quick.consumer, partition = 0, offset = 0, CreateTime = 1536652333476, serialized key size = -1, serialized value size = 30, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = test receive by consumerRecord)

### 批量消费案例 ###

1.  重新创建一份新的消费者配置，配置为一次拉取5条消息
2.  创建一个监听容器工厂，设置其为批量消费并设置并发量为5，这个并发量根据分区数决定，必须小于等于分区数，否则会有线程一直处于空闲状态
3.  创建一个分区数为8的Topic
4.  创建监听方法，设置消费id为batch，clientID前缀为batch，监听topic.quick.batch，使用batchContainerFactory工厂创建该监听容器

@Component
    public class BatchListener {
    
        private static final Logger log= LoggerFactory.getLogger(BatchListener.class);
    
        private Map<String, Object> consumerProps() {
            Map<String, Object> props = new HashMap<>();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
            props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
            props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
            //一次拉取消息数量
            props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "5");
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            return props;
        }
    
        @Bean("batchContainerFactory")
        public ConcurrentKafkaListenerContainerFactory listenerContainer() {
            ConcurrentKafkaListenerContainerFactory container = new ConcurrentKafkaListenerContainerFactory();
            container.setConsumerFactory(new DefaultKafkaConsumerFactory(consumerProps()));
            //设置并发量，小于或等于Topic的分区数
            container.setConcurrency(5);
            //设置为批量监听
            container.setBatchListener(true);
            return container;
        }
    
        @Bean
        public NewTopic batchTopic() {
            return new NewTopic("topic.quick.batch", 8, (short) 1);
        }
    
    
        @KafkaListener(id = "batch",clientIdPrefix = "batch",topics = {"topic.quick.batch"},containerFactory = "batchContainerFactory")
        public void batchListener(List<String> data) {
            log.info("topic.quick.batch  receive : ");
            for (String s : data) {
                log.info(  s);
            }
        }
    
    }

紧接着我们启动项目，控制台的日志信息非常完整，我们可以看到batchListener这个监听容器的partition分配信息。我们设置concurrency为5，也就是将会启动5条线程进行监听，那我们创建的topic则是有8个partition，意味着将有3条线程分配到2个partition和2条线程分配到1个partition。我们可以看到这段日志的最后5行，这就是每条线程分配到的partition。

2018-09-11 12:47:49.628  INFO 4708 --- [    batch-2-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=batch-2, groupId=batch] Successfully joined group with generation 98
    2018-09-11 12:47:49.628  INFO 4708 --- [    batch-2-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=batch-2, groupId=batch] Setting newly assigned partitions [topic.quick.batch-4, topic.quick.batch-5]
    2018-09-11 12:47:49.630  INFO 4708 --- [    batch-3-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=batch-3, groupId=batch] Successfully joined group with generation 98
    2018-09-11 12:47:49.630  INFO 4708 --- [    batch-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=batch-0, groupId=batch] Successfully joined group with generation 98
    2018-09-11 12:47:49.630  INFO 4708 --- [    batch-4-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=batch-4, groupId=batch] Successfully joined group with generation 98
    2018-09-11 12:47:49.630  INFO 4708 --- [    batch-3-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=batch-3, groupId=batch] Setting newly assigned partitions [topic.quick.batch-6]
    2018-09-11 12:47:49.630  INFO 4708 --- [    batch-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=batch-0, groupId=batch] Setting newly assigned partitions [topic.quick.batch-0, topic.quick.batch-1]
    2018-09-11 12:47:49.630  INFO 4708 --- [    batch-4-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=batch-4, groupId=batch] Setting newly assigned partitions [topic.quick.batch-7]
    2018-09-11 12:47:49.631  INFO 4708 --- [    batch-1-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=batch-1, groupId=batch] Successfully joined group with generation 98
    2018-09-11 12:47:49.631  INFO 4708 --- [    batch-1-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=batch-1, groupId=batch] Setting newly assigned partitions [topic.quick.batch-2, topic.quick.batch-3]
    2018-09-11 12:47:49.633  INFO 4708 --- [    batch-3-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [topic.quick.batch-6]
    2018-09-11 12:47:49.633  INFO 4708 --- [    batch-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [topic.quick.batch-0, topic.quick.batch-1]
    2018-09-11 12:47:49.633  INFO 4708 --- [    batch-4-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [topic.quick.batch-7]
    2018-09-11 12:47:49.633  INFO 4708 --- [    batch-1-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [topic.quick.batch-2, topic.quick.batch-3]
    2018-09-11 12:47:49.634  INFO 4708 --- [    batch-2-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [topic.quick.batch-4, topic.quick.batch-5]

那我们来编写一下测试方法，在短时间内发送12条消息到topic中，可以看到运行结果，对应的监听方法总共拉取了三次数据，其中两次为5条数据，一次为2条数据，加起来就是我们在测试方法发送的12条数据。证明我们的批量消费方法是按预期进行的。

@Autowired
        private KafkaTemplate kafkaTemplate;
    
        @Test
        public void testBatch() {
            for (int i = 0; i < 12; i++) {
                kafkaTemplate.send("topic.quick.batch", "test batch listener,dataNum-" + i);
            }
        }

2018-09-11 12:08:51.840  INFO 12416 --- [    batch-0-C-1] com.viu.kafka.listen.BatchListener       : topic.quick.batch  receive : 
    2018-09-11 12:08:51.840  INFO 12416 --- [    batch-0-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-5
    2018-09-11 12:08:51.840  INFO 12416 --- [    batch-0-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-2
    2018-09-11 12:08:51.840  INFO 12416 --- [    batch-0-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-10
    2018-09-11 12:08:51.840  INFO 12416 --- [    batch-0-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-6
    2018-09-11 12:08:51.840  INFO 12416 --- [    batch-0-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-3
    2018-09-11 12:08:51.841  INFO 12416 --- [    batch-0-C-1] com.viu.kafka.listen.BatchListener       : topic.quick.batch  receive : 
    2018-09-11 12:08:51.841  INFO 12416 --- [    batch-0-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-11
    2018-09-11 12:08:51.841  INFO 12416 --- [    batch-0-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-0
    2018-09-11 12:08:51.841  INFO 12416 --- [    batch-0-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-8
    2018-09-11 12:08:51.841  INFO 12416 --- [    batch-0-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-7
    2018-09-11 12:08:51.841  INFO 12416 --- [    batch-0-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-4
    2018-09-11 12:08:51.842  INFO 12416 --- [    batch-0-C-1] com.viu.kafka.listen.BatchListener       : topic.quick.batch  receive : 
    2018-09-11 12:08:51.842  INFO 12416 --- [    batch-0-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-1
    2018-09-11 12:08:51.842  INFO 12416 --- [    batch-0-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-9

**注意：设置的并发量不能大于partition的数量，如果需要提高吞吐量，可以通过增加partition的数量达到快速提升吞吐量的效果。**

### 监听Topic中指定的分区 ###

紧接着刚才编写的代码里面编写新的监听器  
这里使用@KafkaListener注解的topicPartitions属性监听不同的partition分区。  
@TopicPartition：topic--需要监听的Topic的名称，partitions --需要监听Topic的分区id，  
partitionOffsets --可以设置从某个偏移量开始监听  
@PartitionOffset：partition --分区Id，非数组，initialOffset --初始偏移量

@Bean
        public NewTopic batchWithPartitionTopic() {
            return new NewTopic("topic.quick.batch.partition", 8, (short) 1);
        }
    
        @KafkaListener(id = "batchWithPartition",clientIdPrefix = "bwp",containerFactory = "batchContainerFactory",
            topicPartitions = {
                @TopicPartition(topic = "topic.quick.batch.partition",partitions = {"1","3"}),
                @TopicPartition(topic = "topic.quick.batch.partition",partitions = {"0","4"},
                        partitionOffsets = @PartitionOffset(partition = "2",initialOffset = "100"))
            }
        )
        public void batchListenerWithPartition(List<String> data) {
            log.info("topic.quick.batch.partition  receive : ");
            for (String s : data) {
                log.info(s);
            }
        }

其实和我们刚才写的批量消费区别只是在注解上多了个属性，启动项目我们仔细搜索一下控制台输出的日志，如果存在该日志则说明成功。同样的我们往这个Topic里面写入一些数据，运行后我们可以看到控制台只监听到一部分消息，这是因为创建的Topic的partition数量为8，而我们只监听了0、1、2、3、4这几个partition，也就是说5 6 7这三个分区的消息我们并没有读取出来。

2018-09-11 14:39:52.045  INFO 12412 --- [Partition-4-C-1] o.a.k.c.consumer.internals.Fetcher       : [Consumer clientId=bwp-4, groupId=batchWithPartition] Fetch offset 100 is out of range for partition topic.quick.batch-2, resetting offset

@Test
        public void testBatch() throws InterruptedException {
            for (int i = 0; i < 12; i++) {
                kafkaTemplate.send("topic.quick.batch.partition", "test batch listener,dataNum-" + i);
            }
        }

2018-09-11 14:51:09.063  INFO 1532 --- [Partition-2-C-1] com.viu.kafka.listen.BatchListener       : topic.quick.batch.partition  receive : 
    2018-09-11 14:51:09.063  INFO 1532 --- [Partition-2-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-4
    2018-09-11 14:51:09.064  INFO 1532 --- [Partition-1-C-1] com.viu.kafka.listen.BatchListener       : topic.quick.batch.partition  receive : 
    2018-09-11 14:51:09.064  INFO 1532 --- [Partition-1-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-2
    2018-09-11 14:51:09.075  INFO 1532 --- [Partition-0-C-1] com.viu.kafka.listen.BatchListener       : topic.quick.batch.partition  receive : 
    2018-09-11 14:51:09.075  INFO 1532 --- [Partition-0-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-1
    2018-09-11 14:51:09.078  INFO 1532 --- [Partition-1-C-1] com.viu.kafka.listen.BatchListener       : topic.quick.batch.partition  receive : 
    2018-09-11 14:51:09.078  INFO 1532 --- [Partition-1-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-10
    2018-09-11 14:51:09.091  INFO 1532 --- [Partition-4-C-1] com.viu.kafka.listen.BatchListener       : topic.quick.batch.partition  receive : 
    2018-09-11 14:51:09.091  INFO 1532 --- [Partition-4-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-5
    2018-09-11 14:51:09.095  INFO 1532 --- [Partition-0-C-1] com.viu.kafka.listen.BatchListener       : topic.quick.batch.partition  receive : 
    2018-09-11 14:51:09.096  INFO 1532 --- [Partition-0-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-9
    2018-09-11 14:51:09.097  INFO 1532 --- [Partition-3-C-1] com.viu.kafka.listen.BatchListener       : topic.quick.batch.partition  receive : 
    2018-09-11 14:51:09.098  INFO 1532 --- [Partition-3-C-1] com.viu.kafka.listen.BatchListener       : test batch listener,dataNum-7

### 注解方式获取消息头及消息体 ###

当你接收的消息包含请求头，以及你监听方法需要获取该消息非常多的字段时可以通过这种方式，毕竟get方法代码量还是稍多点的。这里使用的是默认的监听容器工厂创建的，如果你想使用批量消费，把对应的类型改为List即可，比如List<String> data ， List<Integer> key。

**@Payload**：获取的是消息的消息体，也就是发送内容  
**@Header(KafkaHeaders.RECEIVED\_MESSAGE\_KEY)**：获取发送消息的key  
**@Header(KafkaHeaders.RECEIVED\_PARTITION\_ID)**：获取当前消息是从哪个分区中监听到的  
**@Header(KafkaHeaders.RECEIVED\_TOPIC)**：获取监听的TopicName  
**@Header(KafkaHeaders.RECEIVED\_TIMESTAMP)**：获取时间戳

@KafkaListener(id = "anno", topics = "topic.quick.anno")
        public void annoListener(@Payload String data,
                                 @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) Integer key,
                                 @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                                 @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                 @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts) {
            log.info("topic.quick.anno receive : \n"+
                "data : "+data+"\n"+
                "key : "+key+"\n"+
                "partitionId : "+partition+"\n"+
                "topic : "+topic+"\n"+
                "timestamp : "+ts+"\n"
            );
    
        }

监听容器编写好了，那就写个测试方法测试一下。启动测试后可以看到监听方法成功的把我们所需要的数据提取出来了，说明这段代码也是ojbk的。

@Test
        public void testAnno() throws InterruptedException {
            Map map = new HashMap<>();
            map.put(KafkaHeaders.TOPIC, "topic.quick.anno");
            map.put(KafkaHeaders.MESSAGE_KEY, 0);
            map.put(KafkaHeaders.PARTITION_ID, 0);
            map.put(KafkaHeaders.TIMESTAMP, System.currentTimeMillis());
    
            kafkaTemplate.send(new GenericMessage<>("test anno listener", map));
        }

2018-09-11 15:27:47.108  INFO 7592 --- [     anno-0-C-1] com.viu.kafka.listen.SingleListener      : topic.quick.anno receive : 
    data : test anno listener
    key : 0
    partitionId : 0
    topic : topic.quick.anno
    timestamp : 1536650867015

### 使用Ack机制确认消费 ###

Kafka的Ack机制相对于RabbitMQ的Ack机制差别比较大，刚入门Kafka的时候我也被搞蒙了，不过能弄清楚Kafka是怎么消费消息的就能理解Kafka的Ack机制了

我先说说RabbitMQ的Ack机制，RabbitMQ的消费可以说是一次性的，也就是你确认消费后就立刻从硬盘或内存中删除，而且RabbitMQ粗糙点来说是顺序消费，像排队一样，一个个顺序消费，未被确认的消息则会重新回到队列中，等待监听器再次消费。  
但Kafka不同，Kafka是通过最新保存偏移量进行消息消费的，而且确认消费的消息并不会立刻删除，所以我们可以重复的消费未被删除的数据，当第一条消息未被确认，而第二条消息被确认的时候，Kafka会保存第二条消息的偏移量，也就是说第一条消息再也不会被监听器所获取，除非是根据第一条消息的偏移量手动获取。

使用Kafka的Ack机制比较简单，只需简单的三步即可：

1.  设置ENABLE\_AUTO\_COMMIT\_CONFIG=false，禁止自动提交
2.  设置AckMode=MANUAL\_IMMEDIATE
3.  监听方法加入Acknowledgment ack 参数

怎么**拒绝消息**呢，只要在监听方法中**不调用ack.acknowledge()**即可

@Component
    public class AckListener {
    
        private static final Logger log= LoggerFactory.getLogger(AckListener.class);
    
        private Map<String, Object> consumerProps() {
            Map<String, Object> props = new HashMap<>();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
            props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
            props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            return props;
        }
    
        @Bean("ackContainerFactory")
        public ConcurrentKafkaListenerContainerFactory ackContainerFactory() {
            ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
            factory.setConsumerFactory(new DefaultKafkaConsumerFactory(consumerProps()));
            factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL_IMMEDIATE);
            factory.setConsumerFactory(new DefaultKafkaConsumerFactory(consumerProps()));
            return factory;
        }
    
    
        @KafkaListener(id = "ack", topics = "topic.quick.ack",containerFactory = "ackContainerFactory")
        public void ackListener(ConsumerRecord record, Acknowledgment ack) {
            log.info("topic.quick.ack receive : " + record.value());
            ack.acknowledge();
        }
    }

编写测试方法，运行后可以方法监听方法能收到消息，紧接着**注释ack.acknowledge()**方法，重新测试，同样你会发现监听容器能接收到消息，这个时候如果你重启项目还是可以看到未被确认的那几条消息。

@Test
        public void testAck() throws InterruptedException {
            for (int i = 0; i < 5; i++) {
                kafkaTemplate.send("topic.quick.ack", i+"");
            }
        }

在这段章节开头之初我就讲解了Kafka机制会出现的一些情况，导致没办法重复消费未被Ack的消息，解决办法有如下：

> 重新将消息发送到队列中，这种方式比较简单而且可以使用Headers实现第几次消费的功能，用以下次判断

@KafkaListener(id = "ack", topics = "topic.quick.ack", containerFactory = "ackContainerFactory")
        public void ackListener(ConsumerRecord record, Acknowledgment ack, Consumer consumer) {
            log.info("topic.quick.ack receive : " + record.value());
    
            //如果偏移量为偶数则确认消费，否则拒绝消费
            if (record.offset() % 2 == 0) {
                log.info(record.offset()+"--ack");
                ack.acknowledge();
            } else {
                log.info(record.offset()+"--nack");
                kafkaTemplate.send("topic.quick.ack", record.value());
            }
        }

> 使用Consumer.seek方法，重新回到该未ack消息偏移量的位置重新消费，这种可能会导致死循环，原因出现于业务一直没办法处理这条数据，但还是不停的重新定位到该数据的偏移量上

@KafkaListener(id = "ack", topics = "topic.quick.ack", containerFactory = "ackContainerFactory")
        public void ackListener(ConsumerRecord record, Acknowledgment ack, Consumer consumer) {
            log.info("topic.quick.ack receive : " + record.value());
    
            //如果偏移量为偶数则确认消费，否则拒绝消费
            if (record.offset() % 2 == 0) {
                log.info(record.offset()+"--ack");
                ack.acknowledge();
            } else {
                log.info(record.offset()+"--nack");
                consumer.seek(new TopicPartition("topic.quick.ack",record.partition()),record.offset() );
            }
        }