模塊（類）之間解耦利器：EventPublishSubscribeUtils 事件發佈訂閱工具類
若是熟悉C#語言的小夥伴們通常都會知道委託、事件的好處，只需在某個類中提早定義好公開的委託或事件（委託的特殊表現形式）變量，而後在其它類中就能夠很隨意的訂閱該委託或事件，當委託或事件被觸發執行時，會自動通知全部的訂閱者進行消費處理。（觀察者模式用委託來實現是最好不過了，DDD所提倡的事件驅動其根本理念也是如此），固然我這裏想到的是不須要在每一個類中進行定義委託或事件，而是由一個統一的中介者（即EventPublishSubscribeUtils）來提供事件的訂閱及發佈操做，這樣各模塊之間無需直接依賴，只需經過中介者完成發佈通知與訂閱回調便可，何樂而不爲呢？java
這裏我先借助C#語言獨有的委託類型快速實現了一個簡易的EventPublishSubscribeUtils，代碼以下：json
/// <summary>
/// 自定義事件發佈訂閱回調工具類（業務解藕、關注點分離，避免互相依賴）--演示版
/// EventBus簡化版，觀察者模式
/// author:zuowenjun
/// </summary>
public static class EventPublishSubscribeUtils
{
private static ConcurrentDictionary<Type, EventHandler<object>> EventHandlers { get; } = new ConcurrentDictionary<Type, EventHandler<object>>();
private static void removeRegisters(ref EventHandler<object> srcEvents, EventHandler<object> removeTargetEvents)
{
var evtTypes = removeTargetEvents.GetInvocationList().Select(d => d.GetType());
var registeredEventHandlers = Delegate.Combine(srcEvents.GetInvocationList().Where(ei => evtTypes.Contains(ei.GetType())).ToArray());
srcEvents -= (EventHandler<object>)registeredEventHandlers;
}
public static void Register<T>(EventHandler<object> eventHandlers)
{
EventHandlers.AddOrUpdate(typeof(T), eventHandlers,
(t, e) =>
{
//先根據訂閱委託類型匹匹配過濾掉以前已有的相同訂閱，而後再從新訂閱，防止重複訂閱，屢次執行的狀況。
removeRegisters(ref e, eventHandlers);
e += eventHandlers;
return e;
});
}
public static void UnRegister<T>(EventHandler<object> eventHandlers = null)
{
Type eventMsgType = typeof(T);
if (eventHandlers == null)
{
EventHandlers.TryRemove(eventMsgType, out eventHandlers);
return;
}
var e = EventHandlers[eventMsgType];
removeRegisters(ref e, eventHandlers);
}
public static void PublishEvent<T>(T eventMsg, object sender)
{
Type eventMsgType = eventMsg.GetType();
if (EventHandlers.ContainsKey(eventMsgType))
{
EventHandlers[eventMsgType].Invoke(sender, eventMsg);
}
}
}
而後使用就比較簡單了，咱們只需經過EventPublishSubscribeUtils.Register註冊訂閱事件消息，經過EventPublishSubscribeUtils.PublishEvent發佈事件通知，這樣就可讓兩個甚至多個不相關的模塊（類）可以經過消息類型實現1對多的通信與協同處理。使用示例代碼以下：c#
class EventMessage
{
public string Name { get; set; }
public string Msg { get; set; }
public DateTime CreatedDate { get; set; }
}
class DemoA
{
public DemoA()
{
EventHandler<object> eventHandlers = EventCallback1;
eventHandlers += EventCallback2;
EventPublishSubscribeUtils.Register<EventMessage>(eventHandlers);
}
private void EventCallback1(object sender, object e)
{
string json = JsonConvert.SerializeObject(e);
System.Diagnostics.Debug.WriteLine($"EventCallback1=> sender:{sender},e:{json}");
}
private void EventCallback2(object sender, object e)
{
string json = JsonConvert.SerializeObject(e);
System.Diagnostics.Debug.WriteLine($"EventCallback2=> sender:{sender},e:{json}");
}
}
class DemoB
{
public void ShowMsg(string name, string msg)
{
System.Diagnostics.Debug.WriteLine($"ShowMsg=> name:{name},msg:{msg}");
var eventMsg = new EventMessage
{
Name = name,
Msg = msg,
CreatedDate = DateTime.Now
};
EventPublishSubscribeUtils.PublishEvent(eventMsg, nameof(DemoB.ShowMsg));
}
}
//main方法中使用：
var demoA = new DemoA();
var demoB = new DemoB();
demoB.ShowMsg("夢在旅途", "i love csharp and java!");
從上述示例代碼中能夠看出，DemoA與DemoB各爲獨立，互不依賴，它們都不知道有對方的存在，它們只關心業務的處理，經過執行demoB.ShowMsg方法進而觸發回調demoA.EventCallback1，demoA.EventCallback2方法，是否是比起直接從DemoA中調DemoB更好呢？app
c#有委託類型（方法的引用），那若是是在java中該如何實現呢？異步
其實同理，咱們能夠藉助匿名內部類+匿名實現類的方式（如：函數式接口）實現與C#殊途同歸的效果，一樣能夠實現相似的事件發佈與訂閱功能，以下即是採用java語言的實現EventPublishSubscribeUtils類的代碼：async
這個因項目須要，我特地實現了兩種模式，一種支持1對多的普通方式，另外一種支持1對1的訂閱回調方式，有返回值。函數
/**
* 自定義事件發佈訂閱回調工具類（業務解藕、關注點分離，避免互相依賴）
* EventBus簡化版，觀察者模式
* <pre>
* 支持兩種模式
* 1.無返回值：訂閱事件消費（register）+ 發佈事件消息（publishEvent/publishEventAsync）
* 2.有返回值：監聽回調通知處理（listenCallback）+通知回調（notifyCallback）,經過notifyMessageType+MessageChannel 便可標識惟一的一組通知回調與監聽回調處理
* <pre>
* @author zuowenjun
* @date 20200310
*/
public final class EventPublishSubscribeUtils {
private static final Logger LOGGER = LoggerFactory.getLogger(EventPublishSubscribeUtils.class);
private static final Map<Class<?>, LinkedList<Consumer<Object>>> eventConsumers = new ConcurrentHashMap<>();
private static final Map<Class<?>, ConcurrentHashMap<MessageChannel, Function<Object, Object>>> callbackFuncs = new ConcurrentHashMap<>();
private EventPublishSubscribeUtils() {
}
/**
* 註冊事件回調消費者
* 用法：EventSubscribeConsumeUtils.register(this::xxxx方法） 或lambda表達式
* 注意：若回調方法添加了事務註解，則應指派其代理對象的方法來完成回調，如：
* EventSubscribeConsumeUtils.register((xxxService)SpringUtils.getBean(this.class)::xxxx方法）
*
* @param eventConsumer
*/
public static void register(Class<?> eventMessageType, Consumer<Object> eventConsumer) {
if (eventConsumer == null) {
return;
}
LinkedList<Consumer<Object>> eventConsumerItems = null;
if (!eventConsumers.containsKey(eventMessageType)) {
eventConsumers.putIfAbsent(eventMessageType, new LinkedList<>());
}
eventConsumerItems = eventConsumers.get(eventMessageType);
eventConsumerItems.add(eventConsumer);
}
/**
* 取消訂閱回調
*
* @param eventMessageType
* @param eventConsumer
*/
public static void unRegister(Class<?> eventMessageType, Consumer<Object> eventConsumer) {
if (!eventConsumers.containsKey(eventMessageType)) {
return;
}
LinkedList<Consumer<Object>> eventConsumerItems = eventConsumers.get(eventMessageType);
int eventConsumerIndex = eventConsumerItems.indexOf(eventConsumer);
if (eventConsumerIndex == -1) {
return;
}
eventConsumerItems.remove(eventConsumerIndex);
}
/**
* 發佈事件，同步觸發執行回調事件消費者方法（存在阻塞等待），即事件消息生產者
* 用法：在須要觸發事件消息回調時調用，如：publishEvent(eventMessage);
*
* @param eventMessage
*/
public static <T> void publishEvent(T eventMessage) {
Class<?> eventMessageType = eventMessage.getClass();
if (!eventConsumers.containsKey(eventMessageType)) {
return;
}
LOGGER.info("事件已發佈，正在執行通知消費:{}", JSONObject.toJSONString(eventMessage));
for (Consumer<Object> eventConsumer : eventConsumers.get(eventMessageType)) {
try {
eventConsumer.accept(eventMessage);
} catch (Exception ex) {
LOGGER.error("eventConsumer.accept error:{},eventMessageType:{},eventMessage:{}",
ex, eventMessageType, JSONObject.toJSONString(eventMessage));
}
}
}
/**
* 發佈事件，異步觸發執行回調事件消費者方法（異步非阻塞），即事件消息生產者
* 用法：在須要觸發事件消息回調時調用，如：publishEventAsync(eventMessage);
*
* @param eventMessage
*/
public static <T> void publishEventAsync(final T eventMessage) {
Executor asyncTaskExecutor = (Executor) SpringUtils.getBean("asyncTaskExecutor");
asyncTaskExecutor.execute(() -> {
publishEvent(eventMessage);
});
}
/**
* 監聽回調處理（須要有返回值），即有返回值的回調消費者
*
* @param notifyMessageType
* @param messageChannel
* @param callbackFunc
*/
public static void listenCallback(Class<?> notifyMessageType, MessageChannel messageChannel, Function<Object, Object> callbackFunc) {
if (!callbackFuncs.containsKey(notifyMessageType)) {
callbackFuncs.putIfAbsent(notifyMessageType, new ConcurrentHashMap<>());
}
Map<MessageChannel, Function<Object, Object>> functionMap = callbackFuncs.get(notifyMessageType);
if (!functionMap.containsKey(messageChannel)) {
functionMap.putIfAbsent(messageChannel, callbackFunc);
} else {
LOGGER.error("該通知消息類型：{}+消息通道：{}，已被訂閱監聽，重複訂閱監聽無效！", notifyMessageType.getSimpleName(), messageChannel.getDescription());
}
}
/**
* 通知回調（同步等待獲取監聽回調的處理結果），即生產者
*
* @param notifyMessage
* @param messageChannel
* @param <R>
* @return
*/
@SuppressWarnings("unchecked")
public static <R> R notifyCallback(Object notifyMessage, MessageChannel messageChannel) {
Class<?> notifyMessageType = notifyMessage.getClass();
Map<MessageChannel, Function<Object, Object>> functionMap = callbackFuncs.getOrDefault(notifyMessageType, null);
if (functionMap != null) {
Function<Object, Object> callbackFunction = functionMap.getOrDefault(messageChannel, null);
if (callbackFunction != null) {
LOGGER.info("通知回調消息已發佈，正在執行回調處理:{},messageChannel:[{}]", JSONObject.toJSONString(notifyMessage), messageChannel.getDescription());
Object result = callbackFunction.apply(notifyMessage);
try {
return (R) result;
} catch (ClassCastException castEx) {
throw new ClassCastException(String.format("監聽回調處理後返回值實際類型與發佈通知回調待接收的值預期類型不一致，致使類型轉換失敗：%s," +
"請確保notifyCallback與listenCallback針對通知消息類型：%s+消息通道：%s返回值類型必需一致。",
castEx.getMessage(), notifyMessageType.getSimpleName(), messageChannel.getDescription()));
}
}
}
return null;
}
}
固然若是須要實現1對1的通信，除了指定消息類型外，還須要指定消息通信通道（即：惟一標識），目的是能夠實現同一種消息類型，支持不一樣的點對點的處理。工具
/**
* 自定義消息通道
* 做用：用於識別同一個消息類型下不一樣的監聽回調者（notifyMessage+messageChannel 便可標識惟一的一組通知回調[生產者]與監聽回調[消費者]）
* @author zuowenjun
* @date 2020-03-31
*/
public enum MessageChannel {
None("無效"),
MSG_A("測試消息A"),
;
private String description;
MessageChannel(String description) {
this.description=description;
}
public String getDescription() {
return description;
}
}
使用方法示例代碼以下：測試
@Service
public class DemoAService {
private static final Logger LOGGER = LoggerFactory.getLogger(DemoAService.class);
public void showMsg(String name, String msg) {
System.out.printf("【%1$tF %1$tT.%1$tL】hello!%s,DemoAService showMsg:%s %n", new Date(), name, msg);
EventMessage eventMessage = new EventMessage();
eventMessage.setName("aaa");
eventMessage.setMsg("test");
eventMessage.setCreatedDate(new Date());
EventPublishSubscribeUtils.publishEvent(eventMessage);
String msgJsonStr = EventPublishSubscribeUtils.notifyCallback(eventMessage, MessageChannel.MSG_A);
System.out.printf("【%1$tF %1$tT.%1$tL】DemoAService showMsg notifyCallback json result:%2$s %n", new Date(), msgJsonStr);
}
}
@Service
public class DemoBService {
@Autowired
private DemoAService demoAService;
@PostConstruct
public void init(){
//訂閱消費，無返回值，支持1對多，即：同一個消息類型可同時被多個消費者訂閱
EventPublishSubscribeUtils.register(EventMessage.class,this::showFinishedMsg);
//訂閱監聽回調，有返回值，只能1對1
EventPublishSubscribeUtils.listenCallback(EventMessage.class, MessageChannel.MSG_A,this::getMsgCallbak);
}
private void showFinishedMsg(Object eventMsg){
EventMessage eventMessage=(EventMessage)eventMsg;
System.out.printf("【%1$tF %1$tT.%1$tL】%s,receive msg:%s doing...%n",
eventMessage.getCreatedDate(),eventMessage.getName(),eventMessage.getMsg());
//模擬邏輯處理
try {
Thread.sleep(500L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.printf("【%1$tF %1$tT.%1$tL】%s,do finished!!!%n",new Date(),eventMessage.getName());
}
private String getMsgCallbak(Object eventMsg){
EventMessage eventMessage=(EventMessage)eventMsg;
eventMessage.setMsg(eventMessage.getMsg()+"--callback added!");
eventMessage.setCreatedDate(new Date());
System.out.printf("【%1$tF %1$tT.%1$tL】%s,do msg callback!!!%n",new Date(),eventMessage.getName());
return JSONObject.toJSONString(eventMessage);
}
}
如上代碼所示，咱們藉助於EventPublishSubscribeUtils，解耦了兩個Service Bean之間的依賴，避免了循環依賴的問題，去掉了以前爲了解決循環依賴而使用@Lazy註解的方式，更易於擴展與更改。其實Spring底層也使用了相似的Event機制,說明這種方式仍是有合適的用武之地的。this
這裏我經過簡單的關係圖來對比未引用EventPublishSubscribeUtils前與引用後的區別，你們能夠感覺一下哪一種更方便：
以前：
以後：
最後，關於業務解耦，分清業務邊界，我我的認爲跨進程通信使用MQ，同進程跨多模塊（類，或者說跨多業務邊界）可以使用Event事件驅動思路來解決。你們以爲如何呢？若是有更好的方案歡迎評論交流，謝謝。
相關文章
- 1. 發佈訂閱模型， 事件機制
- 2. 發佈訂閱
- 3. JavaScript之發佈訂閱模式理解
- 4. guava的事件發佈訂閱功能
- 5. RabbitMQ 發佈訂閱
- 6. 設計模式之發佈訂閱模式(2) Redis 發佈/訂閱模式
- 7. Redis 發佈訂閱
- 8. 事件的發佈和訂閱 以及消息隊列
- 9. C#中的事件-訂閱與發佈
- 10. 觀察者模式——解耦利器
- 更多相關文章...
- • Redis發佈訂閱模式 - Redis教程
- • jQuery Mobile 工具欄 - jQuery Mobile 教程
- • PHP開發工具
- • JDK13 GA發佈：5大特性解讀
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. 發佈訂閱模型， 事件機制
- 2. 發佈訂閱
- 3. JavaScript之發佈訂閱模式理解
- 4. guava的事件發佈訂閱功能
- 5. RabbitMQ 發佈訂閱
- 6. 設計模式之發佈訂閱模式(2) Redis 發佈/訂閱模式
- 7. Redis 發佈訂閱
- 8. 事件的發佈和訂閱 以及消息隊列
- 9. C#中的事件-訂閱與發佈
- 10. 觀察者模式——解耦利器