前言
村里的老人常说:真男人就该懂得遵守“三不原则”——不主动、不拒绝、不负责。
一个复杂的软件系统,其中必然会存在各种各样的“对象”,如果在设计之初没有注意控制好耦合度,导致各个对象甚至是函数之间高度耦合,那对于后期开发和维护将是一个灾难!
在日常开发中,大家不难发现,“面向事件编程”是解耦合的利器,其对应的设计模式便是大家常常会听到的“观察者模式”,而核心思想,就是尽可能令大部分对象都遵守“三不原则”:
- 合理设计事件处理器,等待事件的发生,而不要主动轮询某个临界资源;
- 设置一个安全高效的事件分发中心,无论多大并发都能保证不拒绝服务;
- 事件生产者不必关心事件将由谁来处理、如何处理,亦无需对结果负责。
接下来我将为大家展示如何设计一个优雅的本地消息分发处理系统。
接口设计
首先我们定义一个通用的 ```Notification``` (也可以叫“事件”,或者“消息”),它包含3个基本信息:
- 事件名;
- 发送者;
- 当前信息(上下文)
/// Notification object with name, sender and extra info
class Notification {Notification(this.name, this.sender, this.userInfo);final String name;final dynamic sender;final Map? userInfo;}
然后我们定义“观察者”接口,以便让任务中心正确分发消息:
/// Notification observer
abstract class Observer {Future<void> onReceiveNotification(Notification notification);}
由于消费者处理该事件时有可能会需要花费较长时间,所以这里我们设计为异步接口。
最后我们再实现一个消息分发中心:
/// Singleton
class NotificationCenter {factory NotificationCenter() => _instance;static final NotificationCenter _instance = NotificationCenter._internal();NotificationCenter._internal();BaseCenter center = BaseCenter();/// Add observer with notification name////// @param observer - who will receive notification/// @param name - notification namevoid addObserver(Observer observer, String name) {center.addObserver(observer, name);}/// Remove observer for notification name////// @param observer - who will receive notification/// @param name - notification namevoid removeObserver(Observer observer, [String? name]) {center.removeObserver(observer, name);}/// Post a notification with extra info////// @param name - notification name/// @param sender - who post this notification/// @param info - extra infoFuture<void> postNotification(String name, dynamic sender, [Map? info]) async {await center.postNotification(name, sender, info);}/// Post a notification////// @param notification - notification objectFuture<void> post(Notification notification) async {await center.post(notification);}}
这个事件分发中心主要实现3个功能:
- 将一个观察者以及其关心的事件名称添加到内部等候队列;
- 将一个观察者移出等候队列;
- 提交一个事件(中心内部异步执行分发)。
并且,因为一个应用系统中通常应该只有一个事件分发中心,所以这里的 NotificationCenter 被设计成为单例模式。
这样一个通用的本地消息分发系统就设计完成了。
应用示例
接下来将为你展示这个系统如何使用。
首先我们先定义一个观察者,并且将其添加到事件分发中心:
第一步,实现 Observer 接口:
import 'package:lnc/notification.dart' as lnc;class _ContactListState extends State<ContactListPage> implements lnc.Observer {// ...@overrideFuture<void> onReceiveNotification(lnc.Notification notification) async {// 获取事件名称与相关信息String name = notification.name;Map? userInfo = notification.userInfo;// 根据事件名称处理信息if (name == 'ContactsUpdated') {ID? contact = userInfo?['contact'];Log.info('contact updated: $contact');// ...} else if (name == 'DocumentUpdated') {ID? did = userInfo?['ID'];Log.info('document updated: $did');// ...}}}
第二步,在适当的时机将该观察者添加进事件分发中心 or 从中心删除:
class _ContactListState extends State<ContactListPage> implements lnc.Observer {_ContactListState() {// ...var nc = lnc.NotificationCenter();nc.addObserver(this, 'ContactsUpdated');nc.addObserver(this, 'DocumentUpdated');}@overridevoid dispose() {var nc = lnc.NotificationCenter();nc.removeObserver(this, 'DocumentUpdated');nc.removeObserver(this, 'ContactsUpdated');super.dispose();}// ...}
第三步,在事件发生点提交事件给分发中心:
// post notificationvar nc = NotificationCenter();nc.postNotification('DocumentUpdated', this, {'ID': identifier,'document': doc,});
至此,一个观察者模式的本地事件系统的应用就介绍完了。
下面我们再来深入一下内部,看看这个事件分发中心是如何实现的?
进阶
注意前面提到的事件分发中心(单例) NotificationCenter,里面有一个代理对象 center:
/// Singleton
class NotificationCenter {factory NotificationCenter() => _instance;static final NotificationCenter _instance = NotificationCenter._internal();NotificationCenter._internal();BaseCenter center = BaseCenter(); // 代理对象,内部实现(可替换)// ...}
这里采用代理模式,是为了方便用户根据项目的特殊需要,定义具体的分发逻辑实现以替换之。
下面介绍一下这个默认的分发中心 BaseCenter:
class BaseCenter {// name => WeakSet<Observer>final Map<String, Set<Observer>> _observers = {};/// Add observer with notification name////// @param observer - listener/// @param name - notification namevoid addObserver(Observer observer, String name) {Set<Observer>? listeners = _observers[name];if (listeners == null) {listeners = WeakSet(); // 弱引用集合listeners.add(observer);_observers[name] = listeners;} else {listeners.add(observer);}}/// Remove observer from notification center////// @param observer - listener/// @param name - notification namevoid removeObserver(Observer observer, [String? name]) {if (name == null) {// 1. remove from all observer set_observers.forEach((key, listeners) {listeners.remove(observer);});// 2. remove empty set_observers.removeWhere((key, listeners) => listeners.isEmpty);} else {// 3. get listeners by nameSet<Observer>? listeners = _observers[name];if (listeners != null && listeners.remove(observer)) {// observer removedif (listeners.isEmpty) {_observers.remove(name);}}}}/// Post notification with name////// @param name - notification name/// @param sender - notification sender/// @param info - extra infoFuture<void> postNotification(String name, dynamic sender, [Map? info]) async {return await post(Notification(name, sender, info));}Future<void> post(Notification notification) async {Set<Observer>? listeners = _observers[notification.name]?.toSet();if (listeners == null) {return;}List<Future> tasks = [];for (Observer item in listeners) {tasks.add(item.onReceiveNotification(notification));}// wait all tasks finishedawait Future.wait(tasks);}}
- 首先,它有3个接口函数和 NotificationCenter 一一对应:
- 其次,它的内部有一个 key 为字符串的映射对象 _observers,其中每一个事件名称(字符串)映射向一个弱引用的集合 WeakSet,集合中的元素则是关注该事件名称的所有观察者;
- 当生产者提交事件时,该中心会根据该事件名称从 _observers 中获取对应的观察者集合,并调用其事件接口函数。
这里有两点值得注意:
- 由于低耦合的设计,各个观察者(事件消费者)分别独立处理事件结果,相互之间并无关联,并且也没有前后时序关系的要求,所以这里的 post 函数会采用异步并发的方式来同时调用这些观察者接口;
- 一般而言,观察者的添加与移除是一一对应的,但为了防止异常情况发生,这里的观察者集合仍然采用弱引用的集合,以便某些观察者非正常退出时,即使没有显式调用 removeObserver() 函数,也不会造成泄漏。
(关于弱引用的实现我们留到以后再来讲解)
代码引用
由于我已提交了一个完整的模块代码到 pub.dev,所以在实际应用中,你只需要在项目工程文件 ```pubspec.yaml``` 中添加
dependencies:lnc: ^0.1.2
然后在需要使用的 dart 文件头引入即可:
import 'package:lnc/notification.dart' as lnc;
全部源码
import 'package:object_key/object_key.dart';
import 'package:lnc/log.dart';/// Notification observer
abstract class Observer {Future<void> onReceiveNotification(Notification notification);
}/// Notification object with name, sender and extra info
class Notification {Notification(this.name, this.sender, this.userInfo);final String name;final dynamic sender;final Map? userInfo;@overrideString toString() {Type clazz = runtimeType;return '<$clazz name="$name">\n\t<sender>$sender</sender>\n''\t<info>$userInfo</info>\n</$clazz>';}}/// Notification center
class NotificationCenter {factory NotificationCenter() => _instance;static final NotificationCenter _instance = NotificationCenter._internal();NotificationCenter._internal();BaseCenter center = BaseCenter();/// Add observer with notification name////// @param observer - who will receive notification/// @param name - notification namevoid addObserver(Observer observer, String name) {center.addObserver(observer, name);}/// Remove observer for notification name////// @param observer - who will receive notification/// @param name - notification namevoid removeObserver(Observer observer, [String? name]) {center.removeObserver(observer, name);}/// Post a notification with extra info////// @param name - notification name/// @param sender - who post this notification/// @param info - extra infoFuture<void> postNotification(String name, dynamic sender, [Map? info]) async {await center.postNotification(name, sender, info);}/// Post a notification////// @param notification - notification objectFuture<void> post(Notification notification) async {await center.post(notification);}}class BaseCenter with Logging {// name => WeakSet<Observer>final Map<String, Set<Observer>> _observers = {};/// Add observer with notification name////// @param observer - listener/// @param name - notification namevoid addObserver(Observer observer, String name) {Set<Observer>? listeners = _observers[name];if (listeners == null) {listeners = WeakSet();listeners.add(observer);_observers[name] = listeners;} else {listeners.add(observer);}}/// Remove observer from notification center////// @param observer - listener/// @param name - notification namevoid removeObserver(Observer observer, [String? name]) {if (name == null) {// 1. remove from all observer set_observers.forEach((key, listeners) {listeners.remove(observer);});// 2. remove empty set_observers.removeWhere((key, listeners) => listeners.isEmpty);} else {// 3. get listeners by nameSet<Observer>? listeners = _observers[name];if (listeners != null && listeners.remove(observer)) {// observer removedif (listeners.isEmpty) {_observers.remove(name);}}}}/// Post notification with name////// @param name - notification name/// @param sender - notification sender/// @param info - extra infoFuture<void> postNotification(String name, dynamic sender, [Map? info]) async {return await post(Notification(name, sender, info));}Future<void> post(Notification notification) async {Set<Observer>? listeners = _observers[notification.name]?.toSet();if (listeners == null) {logDebug('no listeners for notification: ${notification.name}');return;}List<Future> tasks = [];for (Observer item in listeners) {try {tasks.add(item.onReceiveNotification(notification).onError((error, st) =>Log.error('observer error: $error, $st, $notification')));} catch (ex, stackTrace) {logError('observer error: $ex, $stackTrace, $notification');}}// wait all tasks finishedawait Future.wait(tasks);}}
GitHub 地址:
https://github.com/dimchat/sdk-dart/blob/main/lnc/lib/src/notification.dart
结语
这里展示了一个基由观察者模式设计的本地事件通知分发系统,其中包含了“观察者模式”、“单例模式”、“代理模式”等设计思想,希望对你有帮助。
如有其他问题,可以下载登录 Tarsier 与我交流(默认通讯录里找 Albert Moky)