首页 > 应用开发 > Java > 设计模式之订阅发布模式
订阅发布模式(Publish-Subscribe Pattern)是一种行之有效的解耦框架与业务逻辑的方式,也是一种常见的观察者设计模式,它被广泛应用于事件驱动架构中。
在这个模式中,发布者(或者说是主题)并不直接发送消息给订阅者,而是通过调度中心(或者叫消息代理)来传递消息。 发布者(或者说是主题)并不知道订阅者的存在,而订阅者也不知道发布者的存在。他们彼此的关系就是在调度中心注册成为订阅者或者发布者。
当一个发布者有新消息时,就将这个消息发布到调度中心。调度中心就会将这个消息通知给所有订阅者。这就实现了发布者和订阅者之间的解耦,发布者和订阅者不再直接依赖于彼此,他们可以独立地扩展自己。
在具体的实现中,可以通过消息队列、事件总线等机制来实现调度中心,不同语言和平台都有实现的库和框架,例如 Java 中的 ActiveMQ、RabbitMQ、Kafka等。
订阅发布模式有以下优点:
interface
Subscriber {
void
update
(String message);
}
class
Publisher {
private Map<String, List<Subscriber>> subscribers =
new HashMap<>();
public
void
subscribe
(String topic, Subscriber subscriber) {
List<Subscriber> subscriberList = subscribers.get(topic);
if (subscriberList ==
null) {
subscriberList =
new ArrayList<>();
subscribers.put(topic, subscriberList);
}
subscriberList.add(subscriber);
}
public
void
unsubscribe
(String topic, Subscriber subscriber) {
List<Subscriber> subscriberList = subscribers.get(topic);
if (subscriberList !=
null) {
subscriberList.remove(subscriber);
}
}
public
void
publish
(String topic, String message) {
List<Subscriber> subscriberList = subscribers.get(topic);
if (subscriberList !=
null) {
for (Subscriber subscriber : subscriberList) {
subscriber.update(message);
}
}
}
}
class
EmailSubscriber
implements
Subscriber {
private String email;
public
EmailSubscriber
(String email) {
this.email = email;
}
public
void
update
(String message) {
System.out.println(
"Send email to " + email +
": " + message);
}
}
class
SMSSubscriber
implements
Subscriber {
private String phoneNumber;
public
SMSSubscriber
(String phoneNumber) {
this.phoneNumber = phoneNumber;
}
public
void
update
(String message) {
System.out.println(
"Send SMS to " + phoneNumber +
": " + message);
}
}
public
class
Main {
public
static
void
main
(String[] args) {
Publisher publisher =
new Publisher();
Subscriber emailSubscriber1 =
new EmailSubscriber(
"foo@example.com");
Subscriber smsSubscriber1 =
new SMSSubscriber(
"1234567890");
publisher.subscribe(
"news", emailSubscriber1);
publisher.subscribe(
"news", smsSubscriber1);
publisher.publish(
"news",
"发布新消息1");
publisher.unsubscribe(
"news", smsSubscriber1);
publisher.publish(
"news",
"发布新消息2");
}
}
打印输出如下:
Send email to foo@example.com: 发布新消息1
Send SMS to 1234567890: 发布新消息1
Send email to foo@example.com: 发布新消息2
Spring的订阅发布模式是通过发布事件、事件监听器和事件发布器3个部分来完成的
这里我们通过 newbee-mall-pro 项目中已经实现订阅发布模式的下单流程给大家讲解,项目地址:https://github.com/wayn111/newbee-mall-pro
public
class
OrderEvent
extends
ApplicationEvent {
void
onApplicationEvent
(Object event) {
...
}
}
@Component
public
class
OrderListener
implements
ApplicationListener<
OrderEvent> {
@Override
public
void
onApplicationEvent
(OrderEvent event) {
// 生成订单、删除购物车、扣减库存
...
}
}
@Resource
private ApplicationEventPublisher applicationEventPublisher;
private
void
saveOrder
(MallUserVO mallUserVO, Long couponUserId, List<ShopCatVO> shopcatVOList, String orderNo) {
// 订单检查
...
// 生成订单号
String orderNo = NumberUtil.genOrderNo();
// 发布订单事件,在事件监听中处理下单逻辑
applicationEventPublisher.publishEvent(
new OrderEvent(orderNo, mallUserVO, couponUserId, shopcatVOList));
// 所有操作成功后,将订单号返回
return orderNo;
...
}
通过事件监听机制,我们将下单逻辑拆分成如下步骤:
订单检查 生成订单号 发布订单事件,在事件监听中处理订单保存逻辑 所有操作成功后,将订单号返回 每个步骤都是各自独立不互相影响
如上的代码已经实现了订阅发布模式,成功解耦了下单逻辑。但是在性能上还没有得到优化,因为 Spring Boot 项目中,默认情况下事件监听器是同步处理的,也就是说这里下单流程会等待事件监听器处理完毕才返回,最终影响接口响应时长。
Spring Boot 项目中事件监听发布类是由
SimpleApplicationEventMulticaster 这个类实现的,源码中通知订阅者代码如下:
可以看到,代码里是有判断
getTaskExecutor() 方法返回不为空的话,就交由 executor 执行,负责同步执行。这个时候大家就要问了,这里不是有线程池在异步通知订阅者吗?
不急,博主带大家继续查看源码。
可以看到
getTaskExecutor() 方法返回一个成员属性,这个成员属性在
SimpleApplicationEventMulticaster 类中是通过
setTaskExecutor(@Nullable Executor taskExecutor) 方法设置的。我们通过
ctrl + f7 查一下
setTaskExecutor(...) 方法在哪里被调用过,
Ok,到此水落石出,
SimpleApplicationEventMulticaster 类的 taskExecutor 成员属性一直为 null,所以在通过订阅者的时候一直是同步处理,等待订阅者处理完毕。
对于异步处理,我们可以从2个方面入手:
onApplicationEvent 上加上
@Async注解,表示该方法异步执行。taskExecutor属性,通过源码可知,也可以解决。这里博主给大家介绍下怎么修改事件监听发布类来解决。
/**
* 系统启动时執行
*/
@Component
public
class
SpringBeanStartupRunner
implements
ApplicationRunner {
@Override
public
void
run
(ApplicationArguments args)
throws Exception {
// 设置spring默认的事件监听为异步执行
SimpleApplicationEventMulticaster multicaster = SpringContextUtil.getBean(SimpleApplicationEventMulticaster
.
class);
ThreadPoolExecutor threadPoolExecutor =
new ThreadPoolExecutor(
5,
10,
60L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<>(
500),
new CustomizableThreadFactory(
"newbee—event-task"),
new ThreadPoolExecutor.CallerRunsPolicy()
);
multicaster.setTaskExecutor(threadPoolExecutor);
}
}
在系统启动时反射修改
SimpleApplicationEventMulticaster类的
taskExecutor属性,从而让
SimpleApplicationEventMulticaster类支持异步事件通知。
建议大家在日常开发中多加思考哪些业务流程可以适用,例如微服务项目中订单支付成功后需要通知用户、商品、活动等多个服务时,可以考虑使用订阅发布模式。解耦发布者和订阅者,发布者只管发布消息,不需要知道有哪些订阅者,也不需要知道订阅者的具体实现。订阅者只需要关注自己感兴趣的消息即可。这种松耦合的设计使得系统更容易扩展和维护。
关注公众号【waynblog】每周分享技术干货、开源项目、实战经验、高效开发工具等,您的关注将是我的更新动力!
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/70029997/viewspace-2954976/ ,如需转载,请注明出处,否则将追究法律责任。