分布式计算中的事件通知机制解析

### 分布式计算中的事件通知机制解析 #### 1. 引言 在分布式计算中，向感兴趣的各方通知数据变化是一项常见需求。例如，股票行情程序需通知客户股价变化，计算机监控程序要告知管理员系统状态，病毒检测程序在检测到病毒时需警告用户，医疗监控程序在患者需要紧急关注时要呼叫医生等。 为便于讨论，我们将感兴趣接收信息的程序称为订阅者（subscribers），提供信息的程序称为发布者（publishers）。当发布者通知订阅者数据变化时，COM 客户端和 COM 服务器的传统角色会暂时反转。 订阅者了解数据变化的一种简单方法是定期轮询发布者，类似定期刷新网页获取最新股票报价。以下代码展示了这种轮询方式： ```cpp while(true) { bool bMarketClosed = spStockWatcher->IsMarketClosed(); if (bMarketClosed) { break; // time for dinner. } currentQuote = spStockWatcher->GetQuote("MSFT"); ... } ``` 然而，这种轮询策略存在明显缺点： - **资源利用效率低**：若数据变化不频繁，订阅者会浪费大量 CPU 周期请求相同数据，发布者也会花费大量时间回复相同答案。若发布者位于远程机器，还会浪费网络带宽。 - **事件发生与数据接收存在时间延迟**：轮询在数据变化发生和订阅者轮询之间存在不可避免的延迟，且该延迟是不确定的。 更好的方法是让发布者在检测到数据变化时主动发起通知，即触发事件（firing an event）。COM 为实现事件触发和接收提供了不同程度的支持，我们可根据需求选择合适的技术。 #### 2. 紧密耦合事件（Tightly Coupled Events，TCEs） 在紧密耦合事件机制中，订阅者明确知道向哪个发布者请求通知。运行时，订阅者向发布者注册以接收事件，不再感兴趣时取消注册。 为使该机制生效，发布者和订阅者需就用于通信的预定义接口达成一致。订阅者向发布者提供实现该接口的对象，发布者在数据变化时调用该对象的方法。 以股票价格更新为例，我们定义一个源接口 `IStockPriceUpdate` 用于通知订阅者股票价格变化： ```cpp interface IStockPriceUpdate : IUnknown { HRESULT NewQuote([in] BSTR bsSymbol, [in] double dPrice); }; ``` 每次股票价格变化时，发布者应调用该接口的 `NewQuote` 方法，并传入股票符号和当前价格。 #### 3. 连接点（Connection Points） 连接点技术也称为可连接对象技术，广泛应用于许多基于 Microsoft COM 的技术中，如 ActiveX 控件、ActiveX 脚本引擎和 VB 类模块。 为便于理解连接点技术，我们以股票交易为例。假设你是股票交易者，买了寻呼机并希望接收股票报价。不同经纪公司提供信息的方式不同，你需确保所选经纪公司提供寻呼服务。以下是你与经纪公司的对话示例： - 你：你们有通知客户股票价格变化的服务吗？ - 经纪公司：有。 - 你：你们提供寻呼服务吗？ - 经纪公司：有。 - 你：好的，这是我的寻呼机号码。把我加入列表，并给我一个跟踪号码，以便我以后想取消寻呼服务时使用。 在 COM 中，连接点的工作方式类似。订阅者通过查询标准 COM 接口 `IConnectionPointContainer` 询问发布者是否支持连接点机制： ```cpp CComPtr<IConnectionPointContainer> spCPC; hr = spMyBroker->QueryInterface( __uuidof(IConnectionPointContainer), (void**) &spCPC); ``` `IConnectionPointContainer` 接口支持 `FindConnectionPoint` 方法，用于检查对象是否支持特定接口类型的连接点。成功时，该方法返回指向另一个标准接口 `IConnectionPoint` 的指针： ```cpp CComPtr<IConnectionPoint> spCP; hr = spCPC->FindConnectionPoint( __uuidof(IStockPriceUpdate), &spCP); ``` 此时，订阅者可创建接收器对象。以下代码展示了 `IStockPriceUpdate` 接口的实现： ```cpp class CMySink : public IStockPriceUpdate, public CComObjectRoot { public: BEGIN_COM_MAP(CMySink) COM_INTERFACE_ENTRY(IStockPriceUpdate) END_COM_MAP() STDMETHODIMP NewQuote(BSTR bsSymbol, double dPrice) { ... } }; ``` `IConnectionPoint` 接口支持 `Advise` 方法，用于注册接收器对象。成功时，该方法返回一个 cookie（跟踪号码）： ```cpp CComPtr<CComObject<CMySink> > spSink; hr = CComObject<CMySink>::CreateInstance(&spSink); spSink->InternalAddRef(); DWORD dwCookie; hr = spCP->Advise(spSink, &dwCookie); ``` 订阅者现在可以接收数据了。每当股票价格变化时，发布者会为所有注册的对象调用 `NewQuote` 方法。 若订阅者不再想接收数据，可调用 `IConnectionPoint` 接口的 `Unadvise` 方法，并传入 cookie 作为参数： ```cpp hr = spCP->Unadvise(dwCookie); ``` 调用 `Advise` 后，可释放 `IConnectionPointContainer` 和 `IConnectionPoint` 的指针，需要时可重新创建。 需要注意的是，连接点不是一种高效的机制，注册和取消注册接收器对象可能需要多达五次往返。此外，ATL 生成的连接点代码不是线程安全的，对于非 STA 对象或需要从工作线程触发事件的情况，需提供线程安全逻辑并将接收器接口封送到工作线程，使用 GIT 是实现线程安全和封送的好方法。 对于我们定义的 `IStockPriceUpdate` 接口，它适用于强类型语言。对于 VBScript 等弱类型语言，纯调度接口（dispinterface）更适合作为输出接口。 #### 4. MSMQ 事件接收消息 之前的监听器示例是同步接收消息，`IMSMQQueue` 接口的 `Receive` 方法是阻塞调用，直到收到消息或超时才返回。这