kafka监听

1、引入依赖

maven

<dependencies>  
    <!-- Spring Boot Starter -->  
    <dependency>  
        <groupId>org.springframework.boot</groupId>  
        <artifactId>spring-boot-starter</artifactId>  
    </dependency>  
    <!-- Spring Kafka -->  
    <dependency>  
        <groupId>org.springframework.kafka</groupId>  
        <artifactId>spring-kafka</artifactId>  
    </dependency>  
    <!-- Spring Boot Configuration Processor (Optional, for @ConfigurationProperties) -->  
    <dependency>  
        <groupId>org.springframework.boot</groupId>  
        <artifactId>spring-boot-configuration-processor</artifactId>  
        <optional>true</optional>  
    </dependency>  
</dependencies>

gradle

<dependencies>  
    <!-- Spring Boot Starter -->  
    <dependency>  
        <groupId>org.springframework.boot</groupId>  
        <artifactId>spring-boot-starter</artifactId>  
    </dependency>  
    <!-- Spring Kafka -->  
    <dependency>  
        <groupId>org.springframework.kafka</groupId>  
        <artifactId>spring-kafka</artifactId>  
    </dependency>  
    <!-- Spring Boot Configuration Processor (Optional, for @ConfigurationProperties) -->  
    <dependency>  
        <groupId>org.springframework.boot</groupId>  
        <artifactId>spring-boot-configuration-processor</artifactId>  
        <optional>true</optional>  
    </dependency>  
</dependencies>

2、配置文件

spring:  
  kafka:  
    bootstrap-servers: localhost:9092 # Kafka服务器地址  
    consumer:  
      group-id: my-group # 消费者组ID  
      auto-offset-reset: earliest # 如果Kafka中没有初始偏移量或者当前偏移量不存在，则从最早的记录开始消费  
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer # 键的反序列化器  
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer # 值的反序列化器  
    # 其他配置...

3、代码

1、普通监听

packagecom.example.kafkademo;importorg.springframework.kafka.annotation.KafkaListener;importorg.springframework.stereotype.Component;@ComponentpublicclassKafkaMessageListener{/**  
     * 监听Kafka消息  
     * @param message 接收到的消息  
     */@KafkaListener(topics ="test-topic", groupId ="my-group")publicvoidlisten(String message){System.out.println("Received Message: "+ message);// 在这里处理接收到的消息  }}

2、订阅发布模式

Spring框架中的发布订阅模式是一种常用的设计模式，它在事件驱动架构中扮演着重要角色。这种模式通过解耦事件的发送者（发布者）和接收者（订阅者）来提高系统的灵活性和可扩展性。下面我将详细介绍Spring中发布订阅模式的用处及其应用场景。
发布订阅模式的基本概念

    ~~~~~~~ 
   
  
        在发布订阅模式中，有三个主要组成部分：

 
  
   
   
                 
    
   
  
    ~~~~~~~~~~~ 
   
  
            1. **发布者 (Publisher)**：负责发送事件。

 
  
   
   
                 
    
   
  
    ~~~~~~~~~~~ 
   
  
            2. **事件 (Event)**：封装了一些数据或状态变化的通知。

 
  
   
   
                 
    
   
  
    ~~~~~~~~~~~ 
   
  
            3. **订阅者 (Subscriber 或 Listener)**：负责接收并处理事件。

Spring中的实现

Spring通过ApplicationEvent和ApplicationListener接口以及ApplicationEventPublisher接口来支持发布订阅模式。

•ApplicationEvent：这是所有事件的基类，事件对象通常会携带一些数据或状态信息。•ApplicationListener：这是事件监听器接口，定义了一个onApplicationEvent方法，用于处理事件。•ApplicationEventPublisher：这是一个接口，提供了publishEvent方法来发布事件。

发布订阅模式的用处

1. 解耦:•发布者和订阅者之间是松耦合的。发布者不知道谁会监听它的事件，订阅者也不知道事件是由谁发布的。•这种解耦使得系统更容易维护和扩展。
2. 灵活性:•可以动态地添加或删除订阅者，而不需要修改发布者或其他订阅者的代码。•支持多种类型的事件处理策略，如同步处理、异步处理等。
3. 扩展性:•新增事件类型或订阅者变得非常简单，只需实现相应的事件类和监听器即可。•易于集成第三方系统，可以通过发布事件来通知外部系统。
4. 事件驱动架构:•适用于构建高度动态和响应式的系统，例如实时通知、消息传递、状态变化跟踪等。
5. 异步处理:•通过事件机制，可以轻松地实现异步处理，从而提高系统的响应速度和整体性能。
6. 简化代码:•通过使用事件发布和订阅机制，可以将关注点分离，使得代码更加清晰和易于理解。典型应用场景
7. 状态变化通知:•当系统中某个对象的状态发生变化时，可以发布事件来通知所有关心这一变化的组件。
8. 异步消息处理:•发布事件以异步处理消息，例如发送电子邮件或短信通知。
9. 分布式系统通信:•在微服务架构中，服务之间可以通过发布事件来相互通信。
10. 日志记录和审计:•发布事件来记录重要的操作或状态变化，这对于审计和监控非常有用。
11. 用户界面更新:•在Web应用中，当后端状态改变时，可以通过发布事件来更新前端界面。

监听器

packagecom.dc.bill.syncentry.Listener;import...// 省略了import语句  @Component@Slf4jpublicclassKafkaComponent{@ResourceprivateApplicationEventPublisher applicationEventPublisher;// 用于发布事件  @ResourceprivateKafkaUtils kafkaUtils;// Kafka工具类，用于数据源切换等  // 使用@KafkaListener注解监听Kafka消息  @KafkaListener(topics ="#{'${dc.topics.bill}'.split(',')}", groupId ="${spring.kafka.groupId}", autoStartup ="${spring.kafka.autoStartup}")publicvoidonMessage(ConsumerRecord<String,String> record,Acknowledgment acknowledgment,@HeadersMap<String,Object> headers)throwsException{String sid =UUID.randomUUID().toString();// 生成唯一标识  // 记录日志  LoggerUtils.writeInfo(log,String.format("消费消息-报关单ENTRY Thread [{%s}], partition[{%s}],  Kafka message:[{%s}] ",Thread.currentThread().getId(), record.partition(), record.value()), sid);// 切换数据源  
        kafkaUtils.kafkaSwitchDataSource(headers, sid);try{// 创建事件对象  DcDataCollectBackEvent event =newDcDataCollectBackEvent(acknowledgment, record.value());  
            event.setSid(sid);  
            event.setBusinessType(BillEnum.BusinessType.SendEntry.getBusinessType());// 发布事件  
            applicationEventPublisher.publishEvent(event);}catch(Exception e){// 记录错误日志  LoggerUtils.writeError(log, e, sid +" HxChen-Kafka-Bill-Entry-Collect-Application-Event-pub-fail,消息："+ record.toString());}try{// 确认消息已被消费  
            acknowledgment.acknowledge();}catch(Exception e){// 记录错误日志  LoggerUtils.writeError(log, e, sid +" HxChen-Kafka-Bill-Entry-Collect-Acknowledgement-fail,消息："+ record.toString());}}}

事件

packagecom.dc.bill.syncbill.Listener.event;import...// 省略了import语句  @Getter@SetterpublicclassDcDataCollectBackEventextendsApplicationEvent{privateString sid;// 唯一标识  privateString businessType;// 业务类型  privateString recordValue;// Kafka消息内容  publicDcDataCollectBackEvent(Object source,String recordValue){super(source);this.recordValue = recordValue;}}

处理器

packagecom.dc.bill.syncbill.Listener.listener;import...// 省略了import语句  @Slf4j@Component@Async// 异步执行  publicclassDcDataCollectBackEventListenerimplementsApplicationListener<DcDataCollectBackEvent>{@ResourceprivateDcDataCollectHandlerService handlerService;// 处理服务  @OverridepublicvoidonApplicationEvent(DcDataCollectBackEvent event){String sid = event.getSid();try{String businessType = event.getBusinessType();String content = event.getRecordValue();// 记录日志  LoggerUtils.writeInfo(log,"消息落地content并添加pushTask任务", sid);// 处理事件  
            handlerService.handleCollectDataBackEvent(businessType, content, sid);}catch(Exception ex){// 记录错误日志  LoggerUtils.writeError(log, ex, sid +",入任务表失败HxChen-Kafka-Bill-Collect-Application-Event-consume-fail "+ event.getRecordValue());}}}