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

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#生产者可用于缓冲等待发送到服务器的记录的内存总字节数，默认值为3355443buffer-memory: 33554432#key的Serializer类，实现类实现了接口org.apache.kafka.common.serialization.Serializerkey-serializer: org.apache.kafka.common.serialization.StringSerializer#value的Serializer类，实现类实现了接口org.apache.kafka.common.serialization.Serializervalue-serializer: org.apache.kafka.common.serialization.StringSerializer#=============== consumer =======================consumer:#用于标识此使用者所属的使用者组的唯一字符串group-id: consumer-group-default#当Kafka中没有初始偏移量或者服务器上不再存在当前偏移量时该怎么办，默认值为latest，表示自动将偏移重置为最新的偏移量#可选的值为latest, earliest, noneauto-offset-reset: earliest#消费者的偏移量将在后台定期提交，默认值为trueenable-auto-commit: true#如果'enable-auto-commit'为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000。auto-commit-interval: 100#密钥的反序列化器类，实现类实现了接口org.apache.kafka.common.serialization.Deserializerkey-deserializer: org.apache.kafka.common.serialization.StringDeserializer#值的反序列化器类，实现类实现了接口org.apache.kafka.common.serialization.Deserializervalue-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 属性指定。