init

2025-11-04 21:09:16 +08:00
parent 8260e293c7
commit bb90a020dc
592 changed files with 61749 additions and 27 deletions
--- a/DouyinApi.EventBus/EventBusKafka/EventBusKafka.cs
+++ b/DouyinApi.EventBus/EventBusKafka/EventBusKafka.cs
@@ -0,0 +1,118 @@
+using DouyinApi.Common.Extensions;
+using Confluent.Kafka;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using Newtonsoft.Json;
+using System;
+
+namespace DouyinApi.EventBus
+{
+    /// <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
+        {
+            _logger.LogInformation("Subscribing to dynamic event {EventName} with {EventHandler}", eventName, typeof(TH).GetGenericTypeName());
+
+            _subsManager.AddDynamicSubscription<TH>(eventName);
+        }
+
+        /// <summary>
+        /// 订阅
+        /// </summary>
+        /// <typeparam name="T">约束：事件模型</typeparam>
+        /// <typeparam name="TH">约束：事件处理器<事件模型></typeparam>
+        public void Subscribe<T, TH>()
+            where T : IntegrationEvent
+            where TH : IIntegrationEventHandler<T>
+        {
+            var eventName = _subsManager.GetEventKey<T>();
+
+            _logger.LogInformation("Subscribing to event {EventName} with {EventHandler}", eventName, typeof(TH).GetGenericTypeName());
+
+            _subsManager.AddSubscription<T, TH>();
+        }
+
+        /// <summary>
+        /// 取消订阅
+        /// </summary>
+        /// <typeparam name="T"></typeparam>
+        /// <typeparam name="TH"></typeparam>
+        public void Unsubscribe<T, TH>()
+            where T : IntegrationEvent
+            where TH : IIntegrationEventHandler<T>
+        {
+            var eventName = _subsManager.GetEventKey<T>();
+
+            _logger.LogInformation("Unsubscribing from event {EventName}", eventName);
+
+            _subsManager.RemoveSubscription<T, TH>();
+        }
+
+        public void UnsubscribeDynamic<TH>(string eventName)
+            where TH : IDynamicIntegrationEventHandler
+        {
+            _subsManager.RemoveDynamicSubscription<TH>(eventName);
+        }
+
+        public void Dispose()
+        {
+            if (_connectionPool != null)
+            {
+                _connectionPool.Dispose();
+            }
+            _subsManager.Clear();
+        }
+
+    }
+}
--- a/DouyinApi.EventBus/EventBusKafka/IKafkaConnectionPool.cs
+++ b/DouyinApi.EventBus/EventBusKafka/IKafkaConnectionPool.cs
@@ -0,0 +1,25 @@
+using Confluent.Kafka;
+using System;
+
+
+namespace DouyinApi.EventBus
+{
+    /// <summary>
+    /// Kafka连接池
+    /// </summary>
+   public interface IKafkaConnectionPool:IDisposable
+    {
+        /// <summary>
+        /// 取对象
+        /// </summary>
+        /// <returns></returns>
+        IProducer<string, byte[]> Producer();
+
+        /// <summary>
+        /// 将对象放入连接池
+        /// </summary>
+        /// <param name="producer"></param>
+        /// <returns></returns>
+        bool Return(IProducer<string, byte[]> producer);
+    }
+}
--- a/DouyinApi.EventBus/EventBusKafka/KafkaConnectionPool.cs
+++ b/DouyinApi.EventBus/EventBusKafka/KafkaConnectionPool.cs
@@ -0,0 +1,79 @@
+using Confluent.Kafka;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using System.Collections.Concurrent;
+using System.Threading;
+
+
+namespace DouyinApi.EventBus
+{
+    /// <summary>
+    /// Kafka producer 连接池管理
+    /// 可以使用微软官方的对象池进行构造ObjectPool
+    /// </summary>
+    public class KafkaConnectionPool : IKafkaConnectionPool
+    {
+        private readonly KafkaOptions _options;
+        private ConcurrentQueue<IProducer<string, byte[]>> _producerPool = new();
+        private int _currentCount;
+        private int _maxSize;
+        public KafkaConnectionPool(IOptions<KafkaOptions> options)
+        {
+            _options = options.Value;
+            _maxSize = _options.ConnectionPoolSize;
+        }
+
+        /// <summary>
+        /// 取对象
+        /// </summary>
+        /// <returns></returns>
+        public IProducer<string,byte[]> Producer()
+        {
+            if (_producerPool.TryDequeue(out var producer))
+            {
+                Interlocked.Decrement(ref _currentCount);
+                return producer;
+            }
+
+            var config = new ProducerConfig()
+            {
+                BootstrapServers = _options.Servers,
+                QueueBufferingMaxMessages = 10,
+                MessageTimeoutMs = 5000,
+                RequestTimeoutMs = 3000
+            };
+
+            producer = new ProducerBuilder<string, byte[]>(config)
+                .Build();
+            return producer;
+        }
+        /// <summary>
+        /// 将对象放入连接池
+        /// </summary>
+        /// <param name="producer"></param>
+        /// <returns></returns>
+        public bool Return(IProducer<string, byte[]> producer)
+        {
+            if (Interlocked.Increment(ref _currentCount) <= _maxSize)
+            {
+                _producerPool.Enqueue(producer);
+                return true;
+            }
+
+            producer.Dispose();
+            Interlocked.Decrement(ref _currentCount);
+
+            return false;
+        }
+        public void Dispose()
+        {
+            _maxSize = 0;
+            _currentCount = 0;
+            while (_producerPool.TryDequeue(out var context))
+            {
+                context?.Dispose();
+            }
+        }
+
+    }
+}
--- a/DouyinApi.EventBus/EventBusKafka/KafkaConsumerHostService.cs
+++ b/DouyinApi.EventBus/EventBusKafka/KafkaConsumerHostService.cs
@@ -0,0 +1,162 @@
+using Autofac;
+using Confluent.Kafka;
+using Confluent.Kafka.Admin;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using Newtonsoft.Json;
+using Newtonsoft.Json.Linq;
+using System;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DouyinApi.EventBus
+{
+    /// <summary>
+    /// Kafka consumer 监听服务
+    /// </summary>
+    public class KafkaConsumerHostService : BackgroundService
+    {
+        private readonly string AUTOFAC_SCOPE_NAME = "blogcore_event_bus";
+        private readonly ILogger<KafkaConsumerHostService> _logger;
+        private readonly IConsumer<string, byte[]> _consumer;
+        private readonly KafkaOptions _options;
+        private readonly IEventBusSubscriptionsManager _subsManager;
+        private readonly ILifetimeScope _autofac;
+        private CancellationTokenSource cts = new();
+        public KafkaConsumerHostService(ILogger<KafkaConsumerHostService> logger,
+            IOptions<KafkaOptions> options,
+            IEventBusSubscriptionsManager eventBusSubscriptionsManager,
+            ILifetimeScope autofac)
+        {
+            _autofac = autofac;
+            _subsManager = eventBusSubscriptionsManager;
+            _logger = logger;
+            _options = options.Value;
+            _consumer = new ConsumerBuilder<string, byte[]>(new ConsumerConfig
+            {
+                BootstrapServers = _options.Servers,
+                GroupId = _options.GroupId,
+                AutoOffsetReset = AutoOffsetReset.Earliest,
+                AllowAutoCreateTopics = true,
+                EnableAutoCommit = false,
+                LogConnectionClose = false
+            }).SetErrorHandler(ConsumerClient_OnConsumeError)
+            .Build();
+        }
+        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+        {
+            var result =  await FetchTopicAsync();
+            if (result)
+            {
+                _consumer.Subscribe(_options.Topic);
+                while (!cts.Token.IsCancellationRequested)
+                {
+                    var consumerResult = _consumer.Consume(cts.Token);
+                    try
+                    {
+                        if (consumerResult.IsPartitionEOF || consumerResult.Message.Value == null) continue;
+
+                        var @event = Protobuf.Deserialize<string>(consumerResult.Message.Value);
+                        await ProcessEvent(consumerResult.Message.Key, @event);
+                    }
+                    catch (ConsumeException e)
+                    {
+                        _logger.LogError($"Error occured: {e.Error.Reason}");
+                    }
+                    finally
+                    {
+                        _consumer.Commit(consumerResult);
+                    }
+                }
+            }
+        }
+        public override Task StopAsync(CancellationToken cancellationToken)
+        {
+            cts.Cancel();
+            _logger.LogInformation("kafka consumer stop and disposable");
+            _consumer.Dispose();
+            return base.StopAsync(cancellationToken);
+        }
+        /// <summary>
+        /// 检测当前Topic是否存在
+        /// </summary>
+        /// <returns></returns>
+        private async Task<bool> FetchTopicAsync()
+        {
+            if (string.IsNullOrEmpty(_options.Topic))
+                throw new ArgumentNullException(nameof(_options.Topic));
+            
+            try
+            {
+                var config = new AdminClientConfig { BootstrapServers = _options.Servers };
+                using var adminClient = new AdminClientBuilder(config).Build();
+                await adminClient.CreateTopicsAsync(Enumerable.Range(0,1).Select(u=> new TopicSpecification
+                {
+                    Name = _options.Topic,
+                    NumPartitions = _options.NumPartitions
+                }));
+            }
+            catch (CreateTopicsException ex) when (ex.Message.Contains("already exists"))
+            {
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError("An error was encountered when automatically creating topic! -->" + ex.Message);
+                return false;
+            }
+            return true;
+        }
+        /// <summary>
+        /// 接收到消息进行处理
+        /// </summary>
+        /// <param name="eventName">事件名称</param>
+        /// <param name="message">消息内容</param>
+        /// <returns></returns>
+        private async Task ProcessEvent(string eventName, string message)
+        {
+            _logger.LogTrace("Processing Kafka event: {EventName}", eventName);
+
+            if (_subsManager.HasSubscriptionsForEvent(eventName))
+            {
+                using (var scope = _autofac.BeginLifetimeScope(AUTOFAC_SCOPE_NAME))
+                {
+                    var subscriptions = _subsManager.GetHandlersForEvent(eventName);
+                    foreach (var subscription in subscriptions)
+                    {
+                        if (subscription.IsDynamic)
+                        {
+                            var handler = scope.ResolveOptional(subscription.HandlerType) as IDynamicIntegrationEventHandler;
+                            if (handler == null) continue;
+                            dynamic eventData = JObject.Parse(message);
+
+                            await Task.Yield();
+                            await handler.Handle(eventData);
+                        }
+                        else
+                        {
+                            var handler = scope.ResolveOptional(subscription.HandlerType);
+                            if (handler == null) continue;
+                            var eventType = _subsManager.GetEventTypeByName(eventName);
+                            var integrationEvent = JsonConvert.DeserializeObject(message, eventType);
+                            var concreteType = typeof(IIntegrationEventHandler<>).MakeGenericType(eventType);
+
+                            await Task.Yield();
+                            await (Task)concreteType.GetMethod("Handle").Invoke(handler, new object[] { integrationEvent });
+                        }
+                    }
+                }
+            }
+            else
+            {
+                _logger.LogWarning("No subscription for Kafka event: {EventName}", eventName);
+            }
+        }
+
+        private void ConsumerClient_OnConsumeError(IConsumer<string, byte[]> consumer, Error e)
+        {
+            _logger.LogError("An error occurred during connect kafka:" + e.Reason);
+        }
+    }
+}
--- a/DouyinApi.EventBus/EventBusKafka/KafkaOptions.cs
+++ b/DouyinApi.EventBus/EventBusKafka/KafkaOptions.cs
@@ -0,0 +1,28 @@
+
+
+namespace DouyinApi.EventBus
+{
+    /// <summary>
+    /// Kafka 配置项
+    /// </summary>
+   public class KafkaOptions
+    {
+        public int ConnectionPoolSize { get; set; } = 10;
+        /// <summary>
+        /// 地址
+        /// </summary>
+        public string Servers { get; set; }
+        /// <summary>
+        /// 主题
+        /// </summary>
+        public string Topic { get; set; }
+        /// <summary>
+        /// 消费者组Id
+        /// </summary>
+        public string GroupId { get; set; }
+        /// <summary>
+        /// 主题分区
+        /// </summary>
+        public int NumPartitions { get; set; }
+    }
+}
--- a/DouyinApi.EventBus/EventBusKafka/ProtobufTransfer.cs
+++ b/DouyinApi.EventBus/EventBusKafka/ProtobufTransfer.cs
@@ -0,0 +1,32 @@
+using System;
+using System.IO;
+namespace DouyinApi.EventBus
+{
+	public class Protobuf
+	{
+		/// <summary>
+		/// Protobuf 反序列化
+		/// </summary>
+		public static T Deserialize<T>(ReadOnlySpan<byte> data)
+		{
+			Stream stream = new MemoryStream(data.ToArray());
+			var info = ProtoBuf.Serializer.Deserialize<T>(stream);
+			return info;
+		}
+		/// <summary>
+		/// 通过Protobuf 转字节
+		/// </summary>
+		public static byte[] Serialize<T>(T data)
+		{
+			byte[] datas;
+			using (var stream = new MemoryStream())
+			{
+				ProtoBuf.Serializer.Serialize(stream, data);
+				datas = stream.ToArray();
+			}
+			return datas;
+
+
+		}
+	}
+}