《SpringBoot2.0 实战》系列-整合kafka实现消息发送、消费

Bertha 。 2023-03-04 11:23 5阅读 0赞

之前写过一篇关于[ActiveMq][]的博文，有兴趣的小伙伴可以点击查看。但是ActiveMq总体性能表现一般，如果对消息队列性能要求较高，业务量较大，那我们需要重新考量。本文就介绍一款性能高的消息队列- kafka。首先看下它们的对比表：

<table> 
 <thead> 
  <tr> 
   <th>对比</th> 
   <th>Kafka</th> 
   <th>ActiveMQ</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>可用性</td> 
   <td>非常高，分布式，多副本备份</td> 
   <td>高，基于主从架构实现的高可用性</td> 
  </tr> 
  <tr> 
   <td>存储接口</td> 
   <td>文件存储，而且这些文件是顺序存储的</td> 
   <td>消息持久化机制有JDBC，KahaDB和LevelDB</td> 
  </tr> 
  <tr> 
   <td>单机吞吐量</td> 
   <td>吞吐量非常大，可以达到10万级</td> 
   <td>吞吐量一般，万级，写入和读取message性能太低</td> 
  </tr> 
  <tr> 
   <td>消息传递模型</td> 
   <td>PUB/SUB（发布/订阅）</td> 
   <td>P2P（点对点），PUB/SUB（发布/订阅）</td> 
  </tr> 
 </tbody> 
</table>

## 基础概念 ##

*  Broker：节点，一个Kafka节点就是一个broker，多个broker可以组成一个Kafka集群。
 *  Topic：一类消息，即主题。
 *  Partition：topic物理上的分组，一个topic可以分为多个partition。
 *  Partition Offset：partition中的每个消息都有一个连续的序列号叫做offset。
 *  Producer : 生产者，生产message发送到topic。
 *  Consumer : 消费者，订阅topic消费message。
 *  Consumer Group：消费组，一个Group包含多个consumer。

## 安装 ##

个人习惯使用docker来安装软件环境，可参考我的博客第5条：  
[https://blog.csdn.net/HXNLYW/article/details/88950291][https_blog.csdn.net_HXNLYW_article_details_88950291]

## 如何整合 ##

添加依赖，版本使用spring-boot-dependencies默认的就行。

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

配置项

# kafka
    spring:  
      kafka:
        # kafka 代理地址
        bootstrap-servers: 47.103.5.190:9092
        producer:
          # 当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。
          batch-size: 16384
          # 设置生产者内存缓冲区的大小。
          buffer-memory: 33554432
          # 发生错误后，消息重发的次数。
          retries: 0
          # 指定消息键和值的序列化方式
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
          # topic主题
          gourd-topic: gourd-topic
        consumer:
          # 指定消费者group id
          group-id: gourd-group
          # 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D
          auto-commit-interval: 100
          # 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
          # latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
          # earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
          auto-offset-reset: earliest
          # 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量
          enable-auto-commit: true
          # 指定消息键和值的反序列化方式
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          # 消费topic主题
          gourd-topic: gourd-topic
        listener:
          # 在侦听器容器中运行的线程数。
          concurrency: 5

增加配置类，根据配置项动态加载

/** * kafka消息队列配置 * @author gour.hu */
    @Configuration
    @ConditionalOnProperty(prefix = "spring.kafka",value = "bootstrap-servers")
    @Import({ KafkaConsumerListener.class})
    public class KafkaConfig { 
    }

封装消息发送工具类

@Slf4j
    public class KafkaUtil { 
        /** * 发送topic消息 * @param topic * @param message */
        public static void sendTopicMessage(String topic, String message){ 
            log.info("发送topic消息体：{}",message);
            KafkaTemplate<String,String> kafkaTemplate = SpringContextHolder.getBean(KafkaTemplate.class);
            ListenableFuture listenableFuture = kafkaTemplate.send(topic,message);
            listenableFuture.addCallback(
                    o -> log.info("消息发送成功,{}", o.toString()),
                    throwable -> log.info("消息发送失败,{}" + throwable.getMessage())
            );
        }
    }

定义消费类逻辑

@Slf4j
    public class KafkaConsumerListener { 
        @KafkaListener(topics = { "${spring.kafka.consumer.gourd-topic}"}, groupId = "${spring.kafka.consumer.group-id}" ,containerFactory = "kafkaListenerContainerFactory")
        public void kafkaConsumerTest(ConsumerRecord<String, String> record) { 
            log.info("消消费消息 topic = {} , content = {}",record.topic(),record.value());
            String messageText = record.value();
            // 业务代码......
        }
    }

测试接口，本人喜欢使用swagger/ postman 测试

@Slf4j
    @RestController
    @RequestMapping("/mq/kafka")
    @Api(tags = "kafka测试API")
    @ConditionalOnBean({ KafkaConfig.class})
    public class KafkaController { 
    
        @Value("${spring.kafka.producer.gourd-topic}")
        private String topic;
    
        @GetMapping("/sendMsg")
        @ApiOperation(value = "发送消息到主题")
        public BaseResponse sendMsg(@RequestParam("msg")String  msg) { 
            KafkaUtil.sendTopicMessage(topic,msg);
            return BaseResponse.ok("success!");
        }
    }

这样基础的整合就好了，下面我们测试下。

## 测试![在这里插入图片描述][20200804110346752.gif]附： ##

> kafka可视化管理工具：kafka tool  
> 下载地址：[http://www.kafkatool.com/download.html][http_www.kafkatool.com_download.html]  
> 使用：[https://www.jianshu.com/p/aa196f24f332][https_www.jianshu.com_p_aa196f24f332]
> 
> **源码：**  
> 代码均已上传至本人的开源项目  
> cloud-plus：[https://blog.csdn.net/HXNLYW/article/details/104635673][https_blog.csdn.net_HXNLYW_article_details_104635673]

[ActiveMq]: https://blog.csdn.net/HXNLYW/article/details/105113133
[https_blog.csdn.net_HXNLYW_article_details_88950291]: https://blog.csdn.net/HXNLYW/article/details/88950291
[20200804110346752.gif]: /images/20230208/8ea59115970e4a65aec262f9c6013b86.png
[http_www.kafkatool.com_download.html]: http://www.kafkatool.com/download.html
[https_www.jianshu.com_p_aa196f24f332]: https://www.jianshu.com/p/aa196f24f332
[https_blog.csdn.net_HXNLYW_article_details_104635673]: https://blog.csdn.net/HXNLYW/article/details/104635673