# EventBus ʹÓÃÊÖ²á ±¾ÎĵµÃæÏò NewLife.Core µÄʼþ×ÜÏßÄÜÁ¦£¬º­¸Ç½Ó¿ÚÄ£ÐÍ¡¢Ä¬ÈÏʵÏÖ `EventBus<TEvent>` µÄ½ø³ÌÄÚ·Ö·¢¡¢»ùÓÚÖ÷ÌâµÄ `EventHub<TEvent>` ·ÓÉ£¬ÒÔ¼°»ùÓÚ¶ÓÁÐµÄ `QueueEventBus<TEvent>`¡£ > ´úÂëλÖãº`NewLife.Core\Messaging\IEventBus.cs`¡¢`NewLife.Core\Messaging\EventHub.cs`¡¢`NewLife.Core\Caching\QueueEventBus.cs` --- ## 1. Éè¼Æ¸ÅÀÀ ### 1.1 ÊÊÓó¡¾° - **½ø³ÌÄÚʼþ**£ºÍ¬Ò»½ø³ÌÄÚ·¢²¼/¶©ÔÄ£¬µÍÑÓ³Ù¡¢Î޳־û¯£¨`EventBus<TEvent>`£©¡£ - **¶àÖ÷Ìâʼþ·ÓÉ**£º°Ñ´ø `topic` µÄÍøÂçÏûϢ·Óɵ½¶ÔӦʼþ×ÜÏß»ò»Øµ÷£¨`EventHub<TEvent>`£©¡£ - **½èÖú»º´æ¶ÓÁеġ°¿ç½ø³Ì¡±Í¶µÝ**£º·¢²¼½ø¶ÓÁС¢Óɺǫ́Ïû·ÑÑ­»·À­È¡²¢·Ö·¢µ½±¾µØ¶©ÔÄÕߣ¨`QueueEventBus<TEvent>`£©¡£ ### 1.2 ºËÐÄÌصã - **·¢²¼Òì²½¡¢¶©ÔÄͬ²½**£º·¢²¼/´¦ÀíʹÓà `PublishAsync`/`HandleAsync`£»¶©ÔÄÓëÈ¡ÏûʹÓà `Subscribe`/`Unsubscribe`¡£ - **Ãݵȶ©ÔÄ**£ºÍ¬Ò» `clientId` Öظ´¶©ÔĻḲ¸ÇÉÏÒ»´Î¡£ - **×î¼ÑŬÁ¦·Ö·¢**£ºÄ¬Èϵ¥¸ö´¦ÀíÆ÷Òì³£²»»áÓ°ÏìÆäËü´¦ÀíÆ÷£»¿Éͨ¹ý `ThrowOnHandlerError` ¸ÄΪÑϸñģʽ¡£ - **ÉÏÏÂÎÄ´«µÝ**£ºÍ¨¹ý `IEventContext`£¨Ä¬ÈÏʵÏÖ `EventContext`£©ÔÚ·¢²¼Õß¡¢¶©ÔÄÕß¼°Öмä²ãÖ®¼ä´«µÝÊý¾Ý¡£ - **Á´Â·×·×Ù**£ºÈôʼþʵÏÖ `ITraceMessage` ÇÒ `TraceId` Ϊ¿Õ£¬`EventBus<TEvent>` ·¢²¼Ê±»á×Ô¶¯Ð´È뵱ǰÂñµãµÄ TraceId¡£ --- ## 2. ½Ó¿ÚÄ£ÐÍ ### 2.1 `IEventBus`£¨·Ç·ºÐÍ£© ÓÃÓÚͳһ³ÖÓв»Í¬Ê¼þÀàÐ͵Ä×ÜÏßÒýÓã¨ÀýÈç·ÅÈë `IEventContext.EventBus`£©¡£ - `Task<Int32> PublishAsync(Object event, IEventContext? context = null, CancellationToken cancellationToken = default)` > ˵Ã÷£ºÄ¬ÈÏʵÏÖͨ³£»á°Ñ `Object` ǿתΪʵ¼ÊµÄ `TEvent`¡£ ### 2.2 `IEventBus<TEvent>`£¨·ºÐÍ×ÜÏߣ© - `Task<Int32> PublishAsync(TEvent event, IEventContext? context = null, CancellationToken cancellationToken = default)` - `Boolean Subscribe(IEventHandler<TEvent> handler, String clientId = "")` - `Boolean Unsubscribe(String clientId = "")` #### `clientId` µÄÓïÒå - ÓÃÓÚʶ±ð¶©ÔÄÕß¡£ - Ïàͬ `clientId` Öظ´¶©ÔĻḲ¸Ç¾É¶©ÔÄ£¨Ãݵȣ©¡£ - ÔÚijЩʵÏÖÀï¿ÉÓÃÓÚ¡°Ïû·Ñ×é/·Ö×顱ÓïÒå¡£ ### 2.3 `IAsyncEventBus<TEvent>`£¨Òì²½¶©ÔÄ/È¡Ïû£© ÊÊÓÃÓÚ¶©ÔÄÐèÒªÍøÂçÍù·µ»òÆäËüÒì²½¶¯×÷µÄ³¡¾°¡£ - `Task<Boolean> SubscribeAsync(IEventHandler<TEvent> handler, String clientId = "", CancellationToken cancellationToken = default)` - `Task<Boolean> UnsubscribeAsync(String clientId = "", CancellationToken cancellationToken = default)` ### 2.4 `IEventHandler<TEvent>`£¨Ê¼þ´¦ÀíÆ÷£© - `Task HandleAsync(TEvent event, IEventContext? context, CancellationToken cancellationToken)` ½¨Ò飺´¦ÀíÆ÷¾¡Á¿Ãݵȣ¬²¢×ðÖØ `cancellationToken`¡£ ### 2.5 `IEventContext` / `EventContext` - `IEventBus EventBus { get; }` `EventContext` »¹Ìṩ£º - `String? Topic`£º¶à²ã´Îʼþ¼Ü¹¹£¨ÀýÈç EventHub£©ÖÐʹÓᣠ- `String? ClientId`£º·¢ËÍ·½±êʶ£¬ÓÃÓÚ·Ö·¢Ê±¡°²»Òª·Ö·¢¸ø×Ô¼º¡±¡£ - `IDictionary<String, Object?> Items` ÓëË÷ÒýÆ÷£ºÐ¯´øÀ©Õ¹Êý¾Ý¡£ > `EventBus<TEvent>` ÄÚ²¿»á³Ø»¯ `EventContext`£ºµ±·¢²¼Ê±Î´´«Èë `context`£¬»á´Ó¶ÔÏó³Ø»ñÈ¡²¢ÔÚ·Ö·¢ºó¹é»¹¡£ --- ## 3. ĬÈÏʼþ×ÜÏß `EventBus<TEvent>` ### 3.1 ÐÐΪÓïÒå - **¼´Ê±·Ö·¢£¬²»´æ´¢**£º²»ÔÚÏߵĶ©ÔÄÕßÊÕ²»µ½ÀúÊ·ÏûÏ¢¡£ - **˳Ðòµ÷Óô¦ÀíÆ÷**£º¶Ôµ±Ç°¶©ÔÄ¿ìÕÕÖð¸öµ÷Óà `HandleAsync`¡£ - **Òì³£²ßÂÔ**£º - `ThrowOnHandlerError = false`£¨Ä¬ÈÏ£©£º¼Ç¼´íÎóÈÕÖ¾£¬¼ÌÐø·Ö·¢¡£ - `ThrowOnHandlerError = true`£ºÓöµ½µÚÒ»¸ö´¦ÀíÆ÷Òì³£Á¢¼´Å׳ö£¬ÖжϷַ¢¡£ - **Åųý·¢ËÍ·½**£ºÈç¹û `context` ÊÇ `EventContext` ÇÒÉèÖÃÁË `ClientId`£¬·Ö·¢Ê±»áÌø¹ý `clientId` ÏàͬµÄ¶©ÔÄÕß¡£ ### 3.2 ¿ìËÙ¿ªÊ¼£¨½ø³ÌÄÚ£© 1) ¶¨ÒåʼþÀàÐÍ£º - ½¨ÒéʹÓÃÇáÁ¿ DTO£¨class/record ¾ù¿É£©¡£ 2) ¶©ÔÄ£º - ͨ¹ýʵÏÖ `IEventHandler<TEvent>`£¬»òʹÓÃÀ©Õ¹·½·¨Ö±½Ó¶©ÔÄίÍС£ 3) ·¢²¼£º - µ÷Óà `PublishAsync`£¬·µ»ØÒѳɹ¦´¦Àí¸ÃʼþµÄ´¦ÀíÆ÷ÊýÁ¿¡£ ʾÀý£¨Î¯Íж©ÔÄ£©£º - `bus.Subscribe(e => Console.WriteLine(e));` - `await bus.PublishAsync(myEvent);` > À©Õ¹·½·¨Î»ÓÚ `EventBusExtensions`£¬»á°ÑίÍаüװΪ `DelegateEventHandler<TEvent>`¡£ ### 3.3 ʹÓÃÉÏÏÂÎÄ£¨´«µÝ¸½¼ÓÊý¾Ý£© Äã¿ÉÒÔÔÚ·¢²¼Ê±´«Èë `EventContext`£¬ÓÃÓÚ£º - ÉèÖà `Topic`/`ClientId`£¨ÓÈÆäÔÚ `EventHub<TEvent>` ³¡¾°£©£» - ͨ¹ý `Items` ±£´æ×Ô¶¨ÒåÊý¾Ý£¨ÀýÈç `ext["Raw"]` ´øԭʼ±¨ÎÄ£©£» - ÔÚ´¦ÀíÆ÷ÖжÁÈ¡ÉÏÏÂÎÄÒÔʵÏÖЭ×÷Âß¼­¡£ ×¢ÒâÊÂÏ - Èô´«ÈëµÄ `context` Ϊ `null`£¬`EventBus<TEvent>` ¿ÉÄÜ´Ó¶ÔÏó³Ø´´½¨ÉÏÏÂÎÄ£¬·Ö·¢ºó»áµ÷Óà `Reset()` ²¢¹é»¹£»´¦ÀíÆ÷²»Ó¦±£´æ¸ÃÉÏÏÂÎÄÒýÓõ½Òì²½ÉúÃüÖÜÆÚÖ®Íâ¡£ ### 3.4 ¶©ÔÄ/È¡Ïû¶©ÔÄ - `Subscribe(handler, clientId)`£º¸²¸Çͬ `clientId` µÄ¾É´¦ÀíÆ÷¡£ - `Unsubscribe(clientId)`£ºÒƳý¶ÔÓ¦¶©ÔÄ¡£ ½¨Ò飺 - Ϊ³¤ÆÚ¶©ÔÄÕßÖ¸¶¨Îȶ¨µÄ `clientId`£¬±ãÓÚÖØÁ¬¸²¸ÇÓëÈ¡Ïû¶©ÔÄ¡£ --- ## 4. `EventHub<TEvent>`£º°´Ö÷Ìâ·ÓɵÄʼþÊàŦ `EventHub<TEvent>` µÄÖ°ÔðÊǽ«´øÖ÷ÌâµÄÊäÈëÏûÏ¢·Ö·¢µ½¶ÔÓ¦µÄʼþ×ÜÏß»ò»Øµ÷¡£ ### 4.1 ÏûÏ¢¸ñʽ ½ö´¦ÀíÒÔ `event#` ¿ªÍ·µÄÏûÏ¢£º - `event#topic#clientId#message` ×Ö¶Î˵Ã÷£º - `topic`£ºÖ÷ÌâÃû³Æ¡£ - `clientId`£º·¢ËÍ·½±êʶ/¶©ÔÄ·Ö×é¡£ - `message`£º - ʼþ JSON£¨Í¨³£Îª `TEvent` µÄ JSON£©£» - »ò¿ØÖÆÖ¸Á`subscribe` / `unsubscribe`¡£ ### 4.2 ×¢²á·½Ê½ - `Add(topic, IEventBus<TEvent> bus)`£º°Ñij¸ö×ÜÏ߹̶¨°ó¶¨µ½Ö÷Ìâ¡£ - `Add(topic, IEventHandler<TEvent> dispatcher)`£º°Ñij¸ö´¦ÀíÆ÷/»Øµ÷°óµ½Ö÷Ì⣨²»¾­¹ý×ÜÏߣ©¡£ - `GetEventBus(topic, clientId)`£ºÍ¨¹ý `Factory` ÑÓ³Ù´´½¨²¢»º´æÖ÷Ìâ×ÜÏߣ»Èç¹ûδÉèÖà `Factory`£¬Ä¬ÈÏ´´½¨ `EventBus<TEvent>`¡£ ### 4.3 ¶©ÔÄ/È¡Ïû¶©ÔÄ£¨¿ØÖÆÖ¸Á µ±ÊÕµ½£º - `event#topic#clientId#subscribe` `EventHub<TEvent>` »á£º - ÒªÇó `context` ÖÐÌṩ `Handler`£º`(context as IExtend)?["Handler"] is IEventHandler<TEvent>`¡£ - `GetEventBus(topic, clientId)` »ñÈ¡Ö÷Ìâ×ÜÏß¡£ - `bus.Subscribe(handler, clientId)` °ó¶¨¶©ÔÄ¡£ µ±ÊÕµ½£º - `event#topic#clientId#unsubscribe` »á£º - ÕÒµ½Ö÷Ìâ×ÜÏß²¢ `Unsubscribe(clientId)`¡£ - Èô×ÜÏßΪ `EventBus<TEvent>` ÇÒûÓÐÈκζ©ÔÄÕߣ¬Ôò´ÓÊàŦÖÐÒƳý¸ÃÖ÷ÌâµÄ×ÜÏßÓë·Ö·¢Æ÷£¨±ÜÃâÖ÷ÌⳤÆÚÕ¼ÓÃÄڴ棩¡£ ### 4.4 ·Ö·¢Â·¾¶Óë·µ»ØÖµ - ÃüÖÐÖ÷Ìâ×ÜÏߣº`bus.PublishAsync(event, context)`£¬·µ»Ø¸Ã×ÜÏߵĴ¦ÀíÆ÷¼ÆÊý¡£ - δÃüÖÐ×ÜÏßµ«ÃüÖзַ¢Æ÷£ºµ÷Óà `dispatcher.HandleAsync`£¬·µ»Ø `1`¡£ - ²»Æ¥Åä/½âÎöʧ°Ü/δע²á£º·µ»Ø `0`¡£ ### 4.5 ÉÏÏÂÎÄдÈë ÔÚ `DispatchAsync(topic, clientId, ...)` ÖУº - Èç¹û `context` ÊÇ `EventContext`£ºÐ´Èë `Topic` / `ClientId`¡£ - ·ñÔòÈô `context` Ö§³Ö `IExtend`£ºÐ´Èë `ext["Topic"]` / `ext["ClientId"]`¡£ ÔÚ `HandleAsync` ÊÕµ½ÍøÂçÏûϢʱ£º - »á°ÑԭʼÊäÈë±£´æµ½ `context["Raw"]`£¨Èô `context` Ö§³Ö `IExtend`£©£¬±ãÓÚ¶©ÔÄÕßÁ㿽±´×ª·¢/Õï¶Ï¡£ --- ## 5. `QueueEventBus<TEvent>`£º»ùÓÚ¶ÓÁеÄʼþ×ÜÏß `QueueEventBus<TEvent>` ¼Ì³Ð×Ô `EventBus<TEvent>`£¬µ«¸Ä±äÁË¡°·¢²¼¡±µÄÓïÒ壺 - `PublishAsync` ²»ÔÙ½ø³ÌÄÚÖ±½Ó·Ö·¢£¬¶øÊÇ **дÈë¶ÓÁÐ**¡£ - ¶©ÔÄʱÆô¶¯Ò»¸öºǫ́Ïû·ÑÑ­»·£¬´Ó¶ÓÁÐÀ­È¡ÏûÏ¢²¢µ÷ÓûùÀàµÄ `DispatchAsync` ·Ö·¢µ½±¾µØ¶©ÔÄÕß¡£ ### 5.1 ʹÓ÷½Ê½ - ´´½¨£º`new QueueEventBus<TEvent>(cache, topic)` - ¶©ÔÄ£ºÊ״ζ©ÔÄ»áÆô¶¯ºǫ́ LongRunning Ïû·ÑÈÎÎñ¡£ - ·¢²¼£ºÐ´Èë¶ÓÁУ»·µ»ØֵΪ¶ÓÁÐ `Add` µÄ½á¹û£¨Í¨³£Îª 1£©¡£ - ÊÍ·Å£ºµ÷Óà `Dispose()` »áÈ¡ÏûÏû·ÑÑ­»·²¢µÈ´ýºǫ́ÈÎÎñÍ˳ö¡£ ### 5.2 È¡ÏûÓë¹Ø±Õ - `Dispose()` »á£º - È¡ÏûÄÚ²¿ `CancellationTokenSource`£» - µÈ´ýºǫ́ÈÎÎñ×î¶àÔ¼ 3 Ã룻 - È»ºóÊÍ·Å CTS¡£ ×¢Ò⣺ - ÊͷźóÔÙ·¢²¼ÏûÏ¢£¬»á¼ÌÐøдÈë¶ÓÁУ¨¶ÓÁÐÊôÓÚÍⲿ `ICache`£©£¬µ«±¾ÊµÀý²»ÔÙÏû·Ñ¡£ --- ## 6. ίÍж©ÔÄ£º`EventBusExtensions` Óë `DelegateEventHandler<TEvent>` ### 6.1 ³£Óö©ÔÄÐÎʽ `EventBusExtensions` Ìṩ¶àÖÖ `Subscribe`/`SubscribeAsync` ±ã½ÝÀ©Õ¹£º - `Action<TEvent>` - `Action<TEvent, IEventContext>` - `Func<TEvent, Task>` - `Func<TEvent, IEventContext, CancellationToken, Task>` ×¢Ò⣺ - ÄÚ²¿Í¨¹ý `DelegateEventHandler<TEvent>` ÊÊÅäµ½ `IEventHandler<TEvent>`¡£ ### 6.2 È¡ÏûÁîÅÆ Ö»ÓÐ×îºóÒ»ÖÖίÍÐÇ©Ãû¿ÉÒÔÖ±½ÓÄõ½ `CancellationToken`¡£ --- ## 7. Ḭ̈߳²È«Óë²¢·¢ÓïÒå - `EventBus<TEvent>`£º - ¶©Ôļ¯ºÏʹÓà `ConcurrentDictionary<String, IEventHandler<TEvent>>`¡£ - ·Ö·¢Ê±Ã¶¾Ù×ÖµäÊÇ¿ìÕÕÓïÒ壺·Ö·¢¹ý³ÌÖж©Ôı仯²»±£Ö¤ÊµÊ±¿É¼û¡£ - `EventHub<TEvent>`£º - `_eventBuses` / `_dispatchers` ¾ùΪ `ConcurrentDictionary`¡£ - `GetEventBus` ²¢·¢Ï¿ÉÄܶà´Î´´½¨£¬µ«×îÖÕ½ö»º´æÒ»·ÝʵÀý¡£ --- ## 8. ´íÎó´¦ÀíÓë×î¼Ñʵ¼ù ### 8.1 ´¦ÀíÆ÷Òì³£ - ĬÈÏ£º¼Ç¼ÈÕÖ¾²¢¼ÌÐø¡£ - ÐèҪǿһÖÂ/Ñϸñʧ°Ü£ºÉèÖà `EventBus<TEvent>.ThrowOnHandlerError = true`¡£ ### 8.2 ÃݵÈÐÔ - ʼþ´¦ÀíÆ÷½¨ÒéÃݵȣ¬±ÜÃâÖظ´Í¶µÝ´øÀ´µÄ¸±×÷Óᣠ### 8.3 ²»Òª³ÖÓгػ¯ÉÏÏÂÎÄ - µ± `context` ÓÉ `EventBus<TEvent>` ×Ô¶¯´´½¨Ê±£¬ËüÀ´×Ô¶ÔÏó³Ø£¬·Ö·¢½áÊø»á±»ÖØÖò¢¸´Óᣠ- ´¦ÀíÆ÷ÄÚÈçÐ賤ÆÚ±£´æÐÅÏ¢£¬Ó¦¸´ÖÆËùÐè×Ö¶Î/Êý¾Ý£¬¶ø²»ÊDZ£´æ `context` ÒýÓᣠ### 8.4 `clientId` µÄʹÓý¨Òé - ¿Í»§¶Ë¶©ÔÄ£ºÊ¹ÓÃÎȶ¨µÄ `clientId`£¬±ãÓÚ¸²¸Ç¾É¶©ÔÄ¡£ - ·¢²¼ÕߣºÔÚ `EventHub<TEvent>` ³¡¾°À`clientId` »á±»ÓÃÓÚ±ÜÃâ¡°·Ö·¢¸ø×Ô¼º¡±¡£ --- ## 9. ³£¼ûÓ÷¨×éºÏ ### 9.1 ½ø³ÌÄÚ£ºÒ»¸ö·¢²¼Õß + ¶à¸ö¶©ÔÄÕß - ʹÓà `EventBus<TEvent>`¡£ - ²»ÐèÒª `EventHub<TEvent>`¡£ ### 9.2 ÍøÂ糡¾°£º°´ topic ¶©ÔÄ/·¢²¼ - ʹÓà `EventHub<TEvent>` ×÷ΪͳһÈë¿Ú£º - ÊäÈ룺ÊÕµ½ÍøÂç×Ö·û´®»ò `IPacket`¡£ - Êä³ö£º·Ö·¢µ½ topic ¶ÔÓ¦µÄ `IEventBus<TEvent>` »ò»Øµ÷¡£ - ÈôÐèÒª°´ topic ´´½¨×ÜÏߣºÌṩ `IEventBusFactory`£¬ÈÃÊàŦ°´Ðè´´½¨¡£ ### 9.3 Àà MQ ³¡¾°£ºÊ¹Óûº´æ¶ÓÁÐ - ʹÓà `QueueEventBus<TEvent>`£º - ·¢²¼Ð´Èë¶ÓÁУ» - ±¾µØ¶©ÔÄÕßÓɺǫ́Ïû·ÑÑ­»·À­È¡¶ÓÁÐÔÙ·Ö·¢¡£ --- ## 10. Ïà¹Ø²âÊÔÓÃÀý£¨¿É²Î¿¼£© - `XUnitTest.Core\Messaging\EventBusTests.cs` - `XUnitTest.Core\Messaging\EventHubTests.cs` - `XUnitTest.Core\Caching\QueueEventBusTests.cs` --- ## 11. FAQ ### Q1£º`PublishAsync` ·µ»ØÖµ´ú±íʲô£¿ ĬÈÏʵÏÖ£¨`EventBus<TEvent>`£©£º·µ»Ø³É¹¦Ö´ÐÐ `HandleAsync` µÄ´¦ÀíÆ÷ÊýÁ¿£¨´¦ÀíÆ÷Å×Òì³£ÇÒ `ThrowOnHandlerError=false` Ôò²»¼ÆÈë³É¹¦£©¡£ ### Q2£ºÎªÊ²Ã´ `EventBus<TEvent>` ÓзǷºÐÍ `IEventBus`£¿ ÓÃÓÚÔÚ²»¹ØÐÄʼþ¾ßÌåÀàÐÍʱ£¨ÀýÈçͳһÉÏÏÂÎÄ»òÖмä¼þ¹ÜµÀ£©³ÖÓÐÒ»¸ö×ÜÏßÒýÓᣠ### Q3£ºÈçºÎÔÚ `EventHub<TEvent>` µÄ subscribe Ö¸ÁîÖÐÌṩ´¦ÀíÆ÷£¿ ¹¹Ôì `EventContext` ²¢Ð´Èë `context["Handler"] = myHandler`£¬È»ºóµ÷Óà `HandleAsync("event#...#subscribe", context)`¡£ --- ## 12. °æ±¾Óë¼æÈÝÐÔ - ±¾Ä£¿éÃæÏò¶àÄ¿±ê¿ò¼Ü£¨`net45` ÖÁ `net10`£©²¢Ê¹ÓÃÒì²½ API¡£ - ÔÚ½ÏÀÏ¿ò¼ÜÏ£¬²¿·Ö `Task` Ïà¹Ø API »áʹÓüæÈÝʵÏÖ£¨ÀýÈç `TaskEx`£©¡£