doc/kafka-consumer.md · 叉子同学/boot-plus

# bootplus-kafka-consumer

#### 介绍
虽然有KafkaListener，但是有时候自己控制consumer会更好。

#### 使用方法

1.  增加依赖

<dependency>
            <groupId>online.chazi.bootplus</groupId>
            <artifactId>bootplus-kafka-consumer</artifactId>
            <version>2.2.0</version>
        </dependency>

2.  增加配置文件

```
    spring:
      kafka:
        consumer:
          key-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
          value-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
          #用于标识此使用者所属的使用者组的唯一字符串。
          group-id: My-Consumer
          client-id: My-client-c
          #如果为true，则消费者的偏移量将在后台定期提交，默认值为true
          enable-auto-commit: false
          #如果'enable.auto.commit'为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000。
          auto-commit-interval: 1000
          #当Kafka中没有初始偏移量或者服务器上不再存在当前偏移量时该怎么办，默认值为latest，表示自动将偏移重置为最新的偏移量
          #可选的值为 latest, earliest, none
          auto-offset-reset: earliest
          #一次调用poll()操作时返回的最大记录数，默认值为500
          max-poll-records: 100
          #如果没有足够的数据立即满足“fetch.min.bytes”给出的要求，服务器在回答获取请求之前将阻塞的最长时间（以毫秒为单位）
          #默认是500
          fetch-max-wait: 500
          #服务器应以字节为单位返回获取请求的最小数据量，默认值为1，对应的kafka的参数为fetch.min.bytes。
          fetch-min-size: 30
          servers: 172.16.119.130:9092,172.16.119.131:9092,172.16.119.132:9092
    ```
    
    * servers 是新增的，必须。

3.  配置类

```java
    @EnableKafka
    @EnableKafkaConsumer
    @Configuration
    public class KafkaConfig {
    }
    ```

* 主要是要写上 @EnableKafkaConsumer 注解

4.  配置类生产Executor

```java
    @Configuration
    public class CommonConfig {
    
        @Bean
        public PooledBlockExecutor<String> pooledBlockExecutor() {
            return new PooledBlockExecutor<>();
        }
    
    }
    ```

5.  编写服务类

```java
    @Slf4j
    @Service
    @AllArgsConstructor
    public class KafkaReceiver {
    
        private KafkaConsumerPoller<byte[], byte[]> poller;
        private KafkaConsumerOffsetService kafkaConsumerOffsetService;
        private PooledBlockExecutor<String> pooledBlockExecutor;
    
        @PreDestroy
        public void stop() {
            poller.stop();
        }
    
        public void start() {
    
            poller.runWithHandler(new SimpleKafkaConsumerHandler<byte[], byte[]>() {
    
                @Override
                public void consume(String topic, int partition, long offset, byte[] bytes) {
                    if (bytes.length > 20)
                        try {
                            String data = new String(bytes);
                            log.debug("消费到数据:{},topic={},partition={},offset={}", data, topic, partition, offset);
                            pooledBlockExecutor.execute(new MessageRepeatWorker(data));
                        } catch (Exception e) {
                            log.error(e.getMessage(), e);
                        }
                }
    
                @Override
                public KafkaConsumerOffset seekOffset(String consumerId, String topic, int partition) {
                    return kafkaConsumerOffsetService.getOffset(
                            consumerId,
                            topic,
                            partition);
                }
    
                @Override
                public void rebalanceOffset(List<KafkaConsumerOffset> list) {
                    kafkaConsumerOffsetService.saveAll(list);
                }
    
                @Override
                public boolean goAhead() {
                    return SessionHolder.size() > 0;
                }
    
                @Override
                public void consumeCount(String topic, int partition, int count) {
                    log.debug("消费到 {} 条数据。topic={}, partition={}", count, topic, partition);
                }
    
            }, Collections.singletonList("你要消费的主题"));
    
        }
    
    }
    ```
    
    * SimpleKafkaConsumerHandler 可以自己写一个，实现 KafkaConsumerHandler<K, V> 接口即可。
    * KafkaConsumerPoller 是starter默认提供的，直接@Resource进来即可。

6.  使用redis来保存当前offset
    
    ```java
    @Service
    @AllArgsConstructor
    public class KafkaConsumerOffsetService {
    
        private RedisTemplate<String, Object> redisTemplate;
    
        private final static String OFFSET_MAP_KEY = "h:offsets";
    
        private String member(String consumerId, String topic, Integer partition) {
            return consumerId + "-" + topic + "-" + partition;
        }
    
        public void saveAll(List<KafkaConsumerOffset> offsets) {
            for (KafkaConsumerOffset offset : offsets) {
                String member = member(offset.getConsumerId(), offset.getTopic(), offset.getPartition());
                redisTemplate.opsForHash().put(OFFSET_MAP_KEY, member, offset);
            }
        }
    
        public KafkaConsumerOffset getOffset(String consumerId, String topic, Integer partition) {
            return (KafkaConsumerOffset) redisTemplate.opsForHash().get(OFFSET_MAP_KEY, member(consumerId, topic, partition));
        }
    
    }
    ```

7.  MessageRepeatWorker具体处理类
    
    ```java
    //只要继承ExecuteWorker类即可
    public class MessageRepeatWorker extends ExecuteWorker<String> {
        public MessageRepeatWorker(String data) {
            super(data);
        }
        @Override
        public void run() {
            System.out.println("消费到数据:" + getData());    
            //todo 写你的逻辑
        }
    }
    ```

叉子同学/boot-plus

About

Releases

Contributors

Activities

叉子同学/boot-plus .gitee-modal { width: 500px !important; }

About

Releases

The Open Source Evaluation Index is derived from the OSS Compass evaluation system, which evaluates projects around the following three dimensions

Contributors

Activities

Search

叉子同学/boot-plus