.NET 分布式事件總線 Jaina v1.0.0 發布

Jaina

.NET 事件總線，簡化項目、類庫、線程、服務等之間的通信，代碼更少，質量更好。‎

特性

• 簡化組件之間通信
  • 支持事件監視器
  • 支持動作執行器
  • 支持自定義消息存儲組件
  • 支持自定義策略執行
  • 支持單消費、多消費消息
  • 支持消息冪等性處理
• 高內聚，低耦合，使代碼更簡單
• 非常快速，每秒可處理 30000 + 消息
• 很小，僅 10KB
• 無第三方依賴
• 可在 Windows/Linux/MacOS 守護進程部署
• 支持分布式、集群
• 高質量代碼和良好單元測試

安裝

• Package Manager

Install-Package Jaina

• .NET CLI

dotnet add package Jaina

快速入門

我們在主頁上有不少例子，這是讓您入門的第一個：

1. 定義事件訂閱者 ToDoEventSubscriber：

// 實現 IEventSubscriber 接口
public class ToDoEventSubscriber : IEventSubscriber
{
    private readonly ILogger<ToDoEventSubscriber> _logger;
    public ToDoEventSubscriber(ILogger<ToDoEventSubscriber> logger)
    {
        _logger = logger;
    }

    // 標記 [EventSubscribe(事件 Id)] 特性
    [EventSubscribe("ToDo:Create")]
    // [EventSubscribe("ToDo:CreateOrUpdate")] // 支持多個
    public async Task CreateToDo(EventHandlerExecutingContext context)
    {
        var todo = context.Source;
        _logger.LogInformation("創建一個 ToDo：{Name}", todo.Payload);
        await Task.CompletedTask;
    }
}

2. 創建控制器 ToDoController，依賴注入 IEventPublisher 服務：

[Route("api/[controller]/[action]")]
[ApiController]
public class ToDoController : ControllerBase
{
    // 依賴注入事件發布者 IEventPublisher
    private readonly IEventPublisher _eventPublisher;
    public ToDoController(IEventPublisher eventPublisher)
    {
        _eventPublisher = eventPublisher;
    }

    // 發布 ToDo:Create 消息
    [HttpPost]
    public async Task CreateDoTo(string name)
    {
        await _eventPublisher.PublishAsync(new ChannelEventSource("ToDo:Create", name));
    }
}

3. 在 Startup.cs 注冊 EventBus 服務：

public class Startup
{
    public void ConfigureServices(IServiceCollection services)
    {
        // 注冊 EventBus 服務
        services.AddEventBus(buidler =>
        {
            // 注冊 ToDo 事件訂閱者
            buidler.AddSubscriber<ToDoEventSubscriber>();
        });
        // ....
    }

    public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
    {
        // ....
    }
}

4. 運行項目：

info: Jaina.Samples.ToDoEventSubscriber[0]
      創建一個 ToDo：Jaina

高級指南

1. 自定義事件源 IEventSource

Jaina 使用 IEventSource 作為消息載體，任何實現該接口的類都可以充當消息載體。

如需自定義，只需實現 IEventSource 接口即可：

public class ToDoEventSource : IEventSource
{
    public ToDoEventSource(string eventId, string todoName)
    {
        EventId = eventId;
        ToDoName = todoName;
    }

    // 自定義屬性
    public string ToDoName { get; }

    /// <summary>
    /// 事件 Id
    /// </summary>
    public string EventId { get; }

    /// <summary>
    /// 事件承載（攜帶）數據
    /// </summary>
    public object Payload { get; }

    /// <summary>
    /// 取消任務 Token
    /// </summary>
    /// <remarks>用於取消本次消息處理</remarks>
    public CancellationToken CancellationToken { get; }

    /// <summary>
    /// 事件創建時間
    /// </summary>
    public DateTime CreatedTime { get; } = DateTime.UtcNow;
}

使用：

await _eventPublisher.PublishAsync(new ToDoEventSource ("ToDo:Create", "我的 ToDo Name"));

2. 自定義事件源存儲器 IEventSourceStorer

Jaina 默認采用 Channel 作為事件源 IEventSource 存儲器，開發者可以使用任何消息隊列組件進行替換，如 Kafka、RabbitMQ、ActiveMQ 等，也可以使用部分數據庫 Redis、SQL Server、MySql 實現。

如需自定義，只需實現 IEventSourceStorer 接口即可：

public class RedisEventSourceStorer : IEventSourceStorer
{
    private readonly IRedisClient _redisClient;

    public RedisEventSourceStorer(IRedisClient redisClient)
    {
        _redisClient = redisClient;
    }

    // 往 Redis 中寫入一條
    public async ValueTask WriteAsync(IEventSource eventSource, CancellationToken cancellationToken)
    {
        await _redisClient.WriteAsync(...., cancellationToken);
    }

    // 從 Redis 中讀取一條
    public async ValueTask<IEventSource> ReadAsync(CancellationToken cancellationToken)
    {
       return await _redisClient.ReadAsync(...., cancellationToken);
    }
}

最后，在注冊 EventBus 服務中替換默認 IEventSourceStorer：

services.AddEventBus(buidler =>
{
    // 替換事件源存儲器
    buidler.ReplaceStorer(serviceProvider =>
    {
        var redisClient = serviceProvider.GetService<IRedisClient>();
        return new RedisEventSourceStorer(redisClient);
    });
});

3. 自定義事件發布者 IEventPublisher

Jaina 默認內置基於 Channel 的事件發布者 ChannelEventPublisher。

如需自定義，只需實現 IEventPublisher 接口即可：

public class ToDoEventPublisher : IEventPublisher
{
    private readonly IEventSourceStorer _eventSourceStorer;

    public ChannelEventPublisher(IEventSourceStorer eventSourceStorer)
    {
        _eventSourceStorer = eventSourceStorer;
    }

    public async Task PublishAsync(IEventSource eventSource)
    {
        await _eventSourceStorer.WriteAsync(eventSource, eventSource.CancellationToken);
    }
}

最后，在注冊 EventBus 服務中替換默認 IEventPublisher：

services.AddEventBus(buidler =>
{
    // 替換事件源存儲器
    buidler.ReplacePublisher<ToDoEventPublisher>();
});

4. 添加事件執行監視器 IEventHandlerMonitor

Jaina 提供了 IEventHandlerMonitor 監視器接口，實現該接口可以監視所有訂閱事件，包括 執行之前、執行之后，執行異常，共享上下文數據。

如添加 ToDoEventHandlerMonitor：

public class ToDoEventHandlerMonitor : IEventHandlerMonitor
{
    private readonly ILogger<ToDoEventHandlerMonitor> _logger;
    public ToDoEventHandlerMonitor(ILogger<ToDoEventHandlerMonitor> logger)
    {
        _logger = logger;
    }

    public Task OnExecutingAsync(EventHandlerExecutingContext context)
    {
        _logger.LogInformation("執行之前：{EventId}", context.Source.EventId);
        return Task.CompletedTask;
    }

    public Task OnExecutedAsync(EventHandlerExecutedContext context)
    {
        _logger.LogInformation("執行之后：{EventId}", context.Source.EventId);

        if (context.Exception != null)
        {
            _logger.LogError(context.Exception, "執行出錯啦：{EventId}", context.Source.EventId);
        }

        return Task.CompletedTask;
    }
}

最后，在注冊 EventBus 服務中注冊 ToDoEventHandlerMonitor：

services.AddEventBus(buidler =>
{
    // 添加事件執行監視器
    buidler.AddMonitor<ToDoEventHandlerMonitor>();
});

5. 自定義事件處理程序執行器 IEventHandlerExecutor

Jaina 提供了 IEventHandlerExecutor 執行器接口，可以讓開發者自定義事件處理函數執行策略，如 超時控制，失敗重試、熔斷等等。

如添加 RetryEventHandlerExecutor：

public class RetryEventHandlerExecutor : IEventHandlerExecutor
{
    public async Task ExecuteAsync(EventHandlerExecutingContext context, Func<EventHandlerExecutingContext, Task> handler)
    {
        // 如果執行失敗，每隔 1s 重試，最多三次
        await Retry(async () => {
            await handler(context);
        }, 3, 1000);
    }
}

最后，在注冊 EventBus 服務中注冊 RetryEventHandlerExecutor：

services.AddEventBus(buidler =>
{
    // 添加事件執行器
    buidler.AddExecutor<RetryEventHandlerExecutor>();
});

6. 訂閱 EventBus 服務執行任務意外異常

services.AddEventBus(buidler =>
{
    // 訂閱 EventBus 未捕獲異常
    buidler.UnobservedTaskExceptionHandler = (obj, args) =>
    {
        // ....
    };
});

7. EventBusOptionsBuilder 說明

EventBusOptionsBuilder 是 AddEventBus 構建服務選項，該選項包含以下屬性和方法：

• 屬性
  • ChannelCapacity：默認內存通道容量
  • UnobservedTaskExceptionHandler：訂閱執行任務未察覺異常
• 方法
  • AddSubscriber<TEventSubscriber>：添加訂閱者
  • ReplacePublisher<TEventPublisher>：替換發布者
  • ReplaceStorer(Func<IServiceProvider, IEventSourceStorer>)：替換存儲器
  • AddMonitor<TEventHandlerMonitor>：添加監視器
  • AddExecutor<TEventHandlerExecutor>：添加執行器

文檔

您可以在主頁找到 Jaina 文檔。

貢獻

該存儲庫的主要目的是繼續發展 Jaina 核心，使其更快、更易於使用。Jaina 的開發在 Gitee 上公開進行，我們感謝社區貢獻錯誤修復和改進。

許可證

Jaina 采用 MulanPSL-2.0 開源許可證。

Copyright (c) 2020-2021 百小僧, Baiqian Co.,Ltd.
Jaina is licensed under Mulan PSL v2.
You can use this software according to the terms andconditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
            https://gitee.com/dotnetchina/Jaina/blob/master/LICENSE
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUTWARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.