最近花了不少时间对消息总线进行了重构。重构的重点是在消息总线中加入了Guava的EventBus，并应用于以下两个场景： （1）改进广播通知 （2）业务逻辑串联，用事件驱动替代责任链模式 EventBus简介EventBus是Google的开源项目Guava里的一个组件，有兴趣的人可以看我前不久的一篇博文解读。总得来说，EventBus是观察者模型的实现，利用它你既可以实现观察者模型的业务场景，还可以基于它的事件驱动机制来实现应用程序内组件之间的解耦与通信。 改进广播通知广播通知是消息总线提供的功能之一。在重构之前，客户端接收广播通知是通过消息总线客户端SDK的一个API来实现的： public void setNotificationListener(IMessageReceiveListener notificationListener); 这个拓扑结构中有些只发送不接受的“虚拟队列”并不是真实存在的队列。这些消息生产者无法接收消息，这是非常大的一个缺陷。我一直在想办法重新设计它，之前的关注点都集中在RabbitMQ上，想在MQ上找到一种解决方案，但这很难，除非摈弃“虚拟队列”的设计。于是，我将关注点转移到消息总线中另一个可以提供pub/sub的组件上（后称之为pubsuber），该组件目前可以是redis也可以是zookeeper。因为每个client（更准确得说是每个创建client的pool）都会以长连接的方式挂在pubsuber上。所以，它本身就是一个很不错的广播渠道，并且因为它脱离RabbitMQ单独实现，跟虚拟队列的设计不相冲突。 上面的思路没有问题，但语义与实现上并不对等。通知的收发从语义上来说应该是Client API级别的。而PubSuber接收到的广播事件却是Pool级别的，并不依赖client（Pool创建PubSuber以及Client）。我们不应该在Pool层面上接收广播事件。因此这里存在一个事件的截获与二次转发的过程。这是我们针对EventBus的第一个应用场景：用它转发PubSuber接收到的广播通知给client。 PubSuber接收到广播消息之后通过EventBus 作二次转发： public class NotifyHandler implements IPubSubListener { @Override public void onChange(String channel, byte[] data, Map |
|
声明：文章版权归原作者所有 部分文章转自互联网 如有侵权请联系
[邮箱地址] 删除
|
注册
登录
