关于KafkaTemplate与 @KafkaListener生产者与消费者功能的实现

1.前言：

1.1关于生产者与消费者的详细介绍请查看另一篇文章：

使用JavaApi实现模拟Kafka的消息生产者与发送者http://t.csdnimg.cn/ukNSU

1.2 本文使用 KafkaTemplate与 @KafkaListener实现生产者与消费者功能：

    Kafka 是一个流行的分布式流处理平台，广泛用于构建实时数据管道和流应用程序。在 Java 应用程序中，Spring Framework 提供了对 Kafka 的集成支持，通过 

spring-kafka

模块实现。

KafkaTemplate

和

@KafkaListener

是 Spring Kafka 中的两个重要组件，它们分别用于发送消息和接收消息。

所需依赖：

!-- spring-kafka --> 
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.6.0</version>
</dependency>
<!-- kafka-clients --> 
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.6.0</version>
</dependency>

2. KafkaTemplate生产者发送消息到kafka

    KafkaTemplate 是Spring Kafka提供的一个高级抽象，用于简化Kafka消息的生产。它封装了KafkaProducer生产者客户端的复杂性，并提供了一系列发送消息的方法。

2.1 主要特性：

1. 线程安全：KafkaTemplate 是线程安全的，可以在多个线程中共享使用。
2. 消息发送：支持同步和异步发送消息。
3. 消息类型：支持发送键值对（Key-Value）消息和仅值消息。
4. 事务管理：支持在事务中发送消息。

2.2 常用方法：

• send(String topic, V data)：发送一个仅值消息到指定主题。
• send(String topic, K key, V data)：发送一个键值对消息到指定主题。
• send(Message<?> message)：发送一个Spring Kafka Message对象。
• send(String topic, Integer partition, K key, V data)：发送消息到指定主题的指定分区。

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

public void sendMessage(String topic, String message) {
    kafkaTemplate.send(topic, message);
}

3.@KafkaListener消费者从kafka中获取消息进行消费

    @KafkaListener 是Spring Kafka提供的一个注解，它简化了消息的接收和处理。用于标记方法作为 Kafka 消费者来监听特定的主题。当消息到达时，被标记的方法将会被自动调用来处理这些消息。这使得消息的处理能够以事件驱动的方式进行，无需轮询或显式拉取。

3.1主要特性：

1. 消息监听：可以监听一个或多个主题的消息。
2. 分区监听：可以指定监听特定分区的消息。
3. 消息类型：支持接收键值对（Key-Value）消息和仅值消息。
4. 消息转换：支持将接收到的消息转换为具体的对象。

3.2常用属性：

• topics：监听的Kafka主题，可以是多个。
• topicPattern：监听的Kafka主题模式，支持正则表达式。
• groupId：消费者组ID。
• containerFactory：指定KafkaListenerContainerFactory，用于自定义消费者配置。

@KafkaListener(topics = "myTopic", groupId = "myGroup")
public void listen(String message) {
    System.out.println("Received message: " + message);
}

    在这个例子中，

listen

方法将被自动调用，每当 Kafka 消费者从 "myTopic" 主题的 "myGroup" 消费者组接收到消息时。

4. 代码举例

    最近有个需求，记录不同系统之间的交互日志，使用拦截器将日志拦截，进行前置与后置处理，最终组装成json发送到kafka中，消费者消费kafka中的数据，最终将日志数据进行入库操作，前端进行展示。

    下面只将KafkaTemplate与 @KafkaListener举例出来，其他不过多展示。

4.1 生产者KafkaTemplate

@Slf4j
@Tag(name = "运营平台-系统日志管理")
@RestController
@RequestMapping("/log")
public class SystemInteractionLogController {

    @Autowired
    private SystemInteractionLogService systemInteractionLogService;
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    @PostMapping("/page")
    @Operation(summary = "日志分页查询")
    //分组设计  根据messageId分类
    public Result<PageData<SystemInteractionLogVO>> page(@RequestBody LogQueryPageDTO dto) {
        return Result.success(systemInteractionLogService.queryPage(dto));
    }

    @PostMapping("/queryInfoById/{id}")
    @Operation(summary = "日志详情查询")
    public Result<SystemInteractionLogInfoVO> queryInfoById(@PathVariable Long id) {
        Validator.validateNotNull(id, "id不能为空");
        return Result.success(systemInteractionLogService.queryInfoById(id));
    }

    @PostMapping ("/testSendLogToKafka")
    @Operation(summary = "(测试使用)往kafka中发送数据")
    public void sendMessage(@RequestBody SystemInteractionLog systemInteractionLog) {
        String topicName = "system-interaction-logs";
        if(ObjectUtils.isEmpty(systemInteractionLog)){
            log.info("数据为空，不操作");
            return;
        }
        try {
            String message = JSONUtil.toJsonStr(systemInteractionLog);
            log.info("消息为：{}",message);
            kafkaTemplate.send(topicName, message);
        }catch (Exception e){
            log.error("消息发送失败：{}",e.getMessage());
        }
    }

}

4.2 消费者@KafkaListener

/**跨系统交互日志监听
 * @author ZhaoShuhao
 * @data 2024/7/13 9:46
 */
@Slf4j
@Component
public class SystemInteractionLogListener {
    @Resource
    private OperateApi operateApi;
 @KafkaListener(topics="system-interaction-logs",groupId="system-interaction-logs-groupId")
 public void receiveTask(String message, Acknowledgment ack) {
     try {
         log.info("接收到消息：{}", message);
         if(StringUtils.isNoneBlank(message)){
             SystemInteractionLogDto logDto  = JSONUtil.toBean(message, SystemInteractionLogDto.class);
              //保存入库操作
             Result<Boolean> result  = operateApi.saveLog(logDto);
             if(!ObjectUtils.isEmpty(result)){
                 if(result.getCode() == 200){
                     log.info("消息处理成功：{}", logDto);
                     //只有保存成功后才会确认接收到消息
                     ack.acknowledge();
                 }else {
                     log.info("消息保存失败：{}", logDto);
                 }
             }
         }
     }catch (Exception e) {
         //异常数据不做处理
         log.info("消息保存失败，参数为{}", message);
         log.error(ExceptionUtils.getStackTrace(e));
     }
 }

}