Springboot 整合 kafka 实现消息的发布和订阅

我不是女神ヾ 2022-09-15 11:59 309阅读 0赞

Kafka 是一个分布式、高吞吐量、可持久性和自动负载均衡的消息队列。它在实现了传统意义上的 MQ 功能的同时，也可以作为大数据的流处理平台。

简单来说，Kafka 就是一个高吞吐量的分布式发布订阅消息系统。

Kafka 的用法跟 RabbitMQ 用法相同，都是作为一个消息中间件收发消息，下面介绍的是 Springboot 微服务集成 Kafka，已经简单的用法说明。

### 依赖 ###

Spring 有专门支持 Kafka 的依赖，引入 Spring 对应版本支持的 Kafka 依赖即可，如下

<dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka</artifactId>
            </dependency>

### 配置 ###

spring:
      kafka:
        bootstrap-servers: ${ KAFKA_HOST:192.168.0.105:9092}
        #=============== producer =======================
        producer:
          #如果该值大于零时，表示启用重试失败的发送次数
          retries: 0
          #每当多个记录被发送到同一分区时，生产者将尝试将记录一起批量处理为更少的请求，默认值为16384(单位字节)
          batch-size: 16384
          #生产者可用于缓冲等待发送到服务器的记录的内存总字节数，默认值为3355443
          buffer-memory: 33554432
          #key的Serializer类，实现类实现了接口org.apache.kafka.common.serialization.Serializer
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          #value的Serializer类，实现类实现了接口org.apache.kafka.common.serialization.Serializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
        #=============== consumer =======================
        consumer:
          #用于标识此使用者所属的使用者组的唯一字符串
          group-id: consumer-group-default
          #当Kafka中没有初始偏移量或者服务器上不再存在当前偏移量时该怎么办，默认值为latest，表示自动将偏移重置为最新的偏移量
          #可选的值为latest, earliest, none
          auto-offset-reset: earliest
          #消费者的偏移量将在后台定期提交，默认值为true
          enable-auto-commit: true
          #如果'enable-auto-commit'为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000。
          auto-commit-interval: 100
          #密钥的反序列化器类，实现类实现了接口org.apache.kafka.common.serialization.Deserializer
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          #值的反序列化器类，实现类实现了接口org.apache.kafka.common.serialization.Deserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

默认 value-serializer 使用 org.apache.kafka.common.serialization.StringSerializer ，只支持文本消息。自定义 org.springframework.kafka.support.serializer.JsonSerializer 可以让消息支持其他类型。

### 使用示例 ###

新建消息实体类

public class Message { 
        private Long id;
        private String content;
        private Date sendTime;
    	
    	// constructor、getter、setter...
    }

消息生产者控制器

@RestController
    @RequestMapping("/kafka/producer")
    public class KafkaProducerController { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProducerController.class);
    
        private static final String TOPIC = "topic-test";
    
        private KafkaTemplate kafkaTemplate;
    
        public KafkaProducerController(KafkaTemplate kafkaTemplate) { 
            this.kafkaTemplate = kafkaTemplate;
        }
    
        @PostMapping("/push")
        public ResponseEntity<String> pushMessage(@RequestBody Message message) { 
            Date time = new Date();
            message.setSendTime(time);
            kafkaTemplate.send(TOPIC, JSON.toJSONString(message)).addCallback(success
                    -> LOGGER.info("{}-生产者发送消息成功:{},时间:{}", TOPIC, success, time), failure
                    -> LOGGER.error("{}-生产者发送消息失败:{}", failure.getMessage()));
            return new ResponseEntity<>("success", HttpStatus.OK);
        }
    }

消息消费者监听

@Component
    public class KafkaConsumer { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(KafkaConsumer.class);
    
        private static final String TOPIC = "topic-test";
    
        @KafkaListener(topics = { TOPIC})
        public void testConsumer(String body) { 
            LOGGER.info("消费时间: {}", new Date());
            Message message = JSON.parseObject(body, Message.class);
            LOGGER.info("topic: {}, 消费消息内容: {}", TOPIC, message);
        }
    }

上边的示例是生产者发送消息到 topic-test，消费者以默认组 consumer-group-default 身份监听 topic-test 消费消息，监听器用 @KafkaListener 注解，topics 属性表示监听的topic，支持同时监听多个，用英文逗号分隔，如果需要使用指定组身份消费消息，可通过注解中的 groupId 属性指定。