【bird-java】分布式服務間的事件總線EventBus
阿新 • • 發佈:2018-02-01
圖片 pic https eight clas 一次 bstr rri eventbus
什麽是EventBus
EventBus是對發布-訂閱模式的一種實現。其以一種非常優雅的方式實現了組件間的解耦與通信,在Android開發、DDD等領域都有非常廣泛的應用。
事件流大致如下:
- Producer向EventBus發送事件。
- EventBus向所有監聽了該事件的Consumer推送事件。
- 監聽了該事件的Consumer消費事件。
註:一個組件即可以是Producer,也可以是Consumer。
分布式服務間的EventBus
在分布式系統中,事件在服務之間的傳遞要比單機EventBus復雜很多。有沒有一種適用於分布式服務之間的,並且事件傳遞就像單機一樣簡單的EventBus呢?在GitHub上搜索了JAVA實現的EventBus,排名前十的幾乎都是用於Android或JAVA的單機事件總線。良久之後...還是自己動手吧。集群環境下的EventBus比單機版需要多考慮一些問題,比如:
- 服務集群部署的情況下,如何保證每個集群均可訂閱該事件,且每個集群只能消費一次該事件。
- 如何實現一個服務內部多個`xxxService`訂閱同一事件。
解決方案:
- 使用`kafka`實現集群間的發布訂閱(基於`topic`),同一集群處於同一個kafka的consumer-group來保證每個集群只會消費一次該事件。
- 服務在啟動時可反射獲得所有實現了`IEventHandler<TEventArg>`的類並緩存,服務消費消息時獲取所有註冊了該消息的handler並調用其`HandleEvent`方法。
部分關鍵源碼
1、事件消息的定義
publicabstract class EventArg implements IEventArg{ private Date eventTime; public EventArg(){ eventTime = new Date(); } public Date getEventTime() { return eventTime; } public void setEventTime(Date eventTime) { this.eventTime = eventTime; } }
事件消息默認記錄創建時間,可擴展其他字段,比如發送時間、標識等。
2、使用spring-kafka發送消息
/** * kafka事件註冊器,向kafka隊列中push消息 */ @Component public class KafkaRegister implements IEventRegister { @Autowired(required = false) private KafkaTemplate<String,IEventArg> kafkaTemplate; /** * 事件註冊 * * @param eventArg 事件參數 */ @Override public void regist(IEventArg eventArg) { kafkaTemplate.send(getTopic(eventArg),eventArg); } /** * 獲取kafka的topic * * * @param eventArg * @return topic */ private String getTopic(IEventArg eventArg){ return eventArg.getClass().getName(); } }
3、消費kafka消息並執行當前服務中所有訂閱了該消息的事件
/** * kafka事件監聽器 */ public class KafkaEventArgListener implements MessageListener<String,EventArg> { @Autowired private IEventHandlerFactory eventHandlerFactory; @Override public void onMessage(ConsumerRecord<String, EventArg> consumerRecord) { if (consumerRecord == null) return; EventArg value = consumerRecord.value(); Set<IEventHandler> handlers = eventHandlerFactory.getHandlers(value); if (handlers == null) return; for (IEventHandler handler : handlers) { handler.HandleEvent(value); } } }
EventBus的使用
1、事件的定義。所有事件均繼承於上文EventArg抽象類,示例如下:
public class TestEventArg extends EventArg{ private String value; public String getValue() { return value; } public void setValue(String value) { this.value = value; } }
2、事件發布。示例代碼:
eventBus.push(new TestEventArg());
3、事件訂閱。一個服務發布事件之後,任何服務中的任何`xxxServiceImpl`類均可訂閱該事件,實現`IEventHandler<TEventArg>`接口即可完成事件的訂閱,示例如下:
public class FormServiceImpl extends AbstractServiceImpl<Form> implements FormService,IEventHandler<TestEventArg> { @Override public void HandleEvent(TestEventArg eventArg) { System.out.println("notify zero======"); } }
整體來說,使用還是很簡單的,EventBus實現與使用示例收錄於bird-java項目中,項目地址:https://github.com/liuxx001/bird-java。
【bird-java】分布式服務間的事件總線EventBus