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、普通监听
package com.example.kafkademo; import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component; @Component
public class KafkaMessageListener { /** * 监听Kafka消息 * @param message 接收到的消息 */ @KafkaListener(topics = "test-topic", groupId = "my-group") public void listen(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方法来发布事件。
发布订阅模式的用处
- 解耦:•发布者和订阅者之间是松耦合的。发布者不知道谁会监听它的事件，订阅者也不知道事件是由谁发布的。•这种解耦使得系统更容易维护和扩展。
- 灵活性:•可以动态地添加或删除订阅者，而不需要修改发布者或其他订阅者的代码。•支持多种类型的事件处理策略，如同步处理、异步处理等。
- 扩展性:•新增事件类型或订阅者变得非常简单，只需实现相应的事件类和监听器即可。•易于集成第三方系统，可以通过发布事件来通知外部系统。
- 事件驱动架构:•适用于构建高度动态和响应式的系统，例如实时通知、消息传递、状态变化跟踪等。
- 异步处理:•通过事件机制，可以轻松地实现异步处理，从而提高系统的响应速度和整体性能。
- 简化代码:•通过使用事件发布和订阅机制，可以将关注点分离，使得代码更加清晰和易于理解。典型应用场景
- 状态变化通知:•当系统中某个对象的状态发生变化时，可以发布事件来通知所有关心这一变化的组件。
- 异步消息处理:•发布事件以异步处理消息，例如发送电子邮件或短信通知。
- 分布式系统通信:•在微服务架构中，服务之间可以通过发布事件来相互通信。
- 日志记录和审计:•发布事件来记录重要的操作或状态变化，这对于审计和监控非常有用。
- 用户界面更新:•在Web应用中，当后端状态改变时，可以通过发布事件来更新前端界面。
监听器
package com.dc.bill.syncentry.Listener; import ... // 省略了import语句 @Component
@Slf4j
public class KafkaComponent { @Resource private ApplicationEventPublisher applicationEventPublisher; // 用于发布事件 @Resource private KafkaUtils kafkaUtils; // Kafka工具类，用于数据源切换等 // 使用@KafkaListener注解监听Kafka消息 @KafkaListener(topics = "#{'${dc.topics.bill}'.split(',')}", groupId = "${spring.kafka.groupId}", autoStartup = "${spring.kafka.autoStartup}") public void onMessage(ConsumerRecord<String, String> record, Acknowledgment acknowledgment, @Headers Map<String, Object> headers) throws Exception { 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 = new DcDataCollectBackEvent(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()); } }
}
事件
package com.dc.bill.syncbill.Listener.event; import ... // 省略了import语句 @Getter
@Setter
public class DcDataCollectBackEvent extends ApplicationEvent { private String sid; // 唯一标识 private String businessType; // 业务类型 private String recordValue; // Kafka消息内容 public DcDataCollectBackEvent(Object source, String recordValue) { super(source); this.recordValue = recordValue; }
}
处理器
package com.dc.bill.syncbill.Listener.listener; import ... // 省略了import语句 @Slf4j
@Component
@Async // 异步执行
public class DcDataCollectBackEventListener implements ApplicationListener<DcDataCollectBackEvent> { @Resource private DcDataCollectHandlerService handlerService; // 处理服务 @Override public void onApplicationEvent(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()); } }
}
