Yuebon.EventBus.Kafka/EventBusKafka.cs

这里只显示每个文件的开头 60 行。登录后可以解锁完整代码。

微信扫码关注公众号，代码整理好后第一时间通知你。

using Microsoft.Extensions.Options;
using Newtonsoft.Json;

namespace Yuebon.EventBus.Kafka;

/// <summary>
/// 基于Kafka的事件总线
/// </summary>
public class EventBusKafka : IEventBus
{
    private readonly ILogger<EventBusKafka> _logger;
    private readonly IEventBusSubscriptionsManager _subsManager;
    private readonly IKafkaConnectionPool _connectionPool;
    private readonly KafkaOptions _options;
    public EventBusKafka(ILogger<EventBusKafka> logger,
        IEventBusSubscriptionsManager subsManager,
        IKafkaConnectionPool connectionPool,
        IOptions<KafkaOptions> options)
    {
        _logger = logger;
        _subsManager = subsManager;
        _connectionPool = connectionPool;
        _options = options.Value;
    }
    /// <summary>
    /// 发布
    /// </summary>
    public void Publish(IntegrationEvent @event)
    {
        var producer =  _connectionPool.Producer();
        try
        {
            var eventName = @event.GetType().Name;
            var body = Protobuf.Serialize(JsonConvert.SerializeObject(@event));
            DeliveryResult<string, byte[]> result = producer.ProduceAsync(_options.Topic, new Message<string, byte[]>
            {
                Key = eventName,
                Value = body
            }).ConfigureAwait(false).GetAwaiter().GetResult();
        }
        catch (Exception ex)
        {
            _logger.LogWarning($"Could not publish event: {@event.Id.ToString("N")} ({ex.Message});  Message:{ JsonConvert.SerializeObject(@event)}");
        }
        finally
        {
            //放入连接池中
            _connectionPool.Return(producer);
        }
    }

    /// <summary>
    /// 订阅
    /// 动态
    /// </summary>
    /// <typeparam name="TH">事件处理器</typeparam>
    /// <param name="eventName">事件名</param>
    public void SubscribeDynamic<TH>(string eventName)
        where TH : IDynamicIntegrationEventHandler
    {