添加项目文件。

2026-10-12 16:22:04 +08:00
parent 7c68a0ca0a
commit aad11be43d
1000 changed files with 120807 additions and 0 deletions
@@ -0,0 +1,14 @@
+using System;
+
+namespace RabbitMQHelper.Attributes
+{
+    public class RabbitMQAttribute : Attribute
+    {
+        public string Topic { get; set; }
+
+        public RabbitMQAttribute(string topic)
+        {
+            this.Topic = topic;
+        }
+    }
+}
@@ -0,0 +1,15 @@
+using RabbitMQHelper.Attributes;
+using System.Reflection;
+
+namespace RabbitMQHelper
+{
+    public class ConsumerExecutorDescriptor
+    {
+        public MethodInfo MethodInfo { get; set; }
+
+        public TypeInfo ImplTypeInfo { get; set; }
+
+        public RabbitMQAttribute Attribute { get; set; }
+        public string ClassName { get; set; }
+    }
+}
@@ -0,0 +1,6 @@
+namespace RabbitMQHelper.Attributes
+{
+    public interface IRabbitSubscribe
+    {
+    }
+}
@@ -0,0 +1,9 @@
+using System.Threading.Tasks;
+
+namespace RabbitMQHelper
+{
+    public interface ISendMQMessage
+    {
+        Task<bool> Send(string topic, object value);
+    }
+}
@@ -0,0 +1,85 @@
+using System;
+using EasyNetQ;
+using EasyNetQ.Topology;
+using Microsoft.AspNetCore.Http;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection.Extensions;
+using Microsoft.Extensions.Logging;
+
+namespace RabbitMQHelper
+{
+    public static class InitRabbitMQSetup
+    {
+        public static void AddInitMQ(this IServiceCollection services, Action<RabbitConfig> setupAction, ILogger logger = null)
+        {
+            //#if NET5_0
+            services.AddHttpContextAccessor();//<IHttpContextAccessor, HttpContextAccessor>();
+//#endif
+
+            var options = new RabbitConfig();
+            setupAction(options);
+            try
+            {
+                var bus = RabbitHutch.CreateBus($"host={options.Host};virtualHost={options.VirtualHost};username={options.UserName};password={options.Password}");
+                bus.Advanced.Connected += delegate (object sender1, ConnectedEventArgs e1) { c_Conneted(sender1, e1, logger); };
+                bus.Advanced.Disconnected += delegate (object sender1, DisconnectedEventArgs e1) { c_DisConnect(sender1, e1, logger); };
+                services.AddSingleton<IBus>(bus);
+            }
+            catch (Exception ex)
+            {
+                throw new Exception("MQ初始化异常:" + ex.Message);
+            }
+            services.AddSingleton<IRabbitConfig>(sp => options); 
+            services.AddScoped<ISendMQMessage, SendMQMessage>();
+            services.AddHostedService<MQHostedService>();
+
+            if (options.ListSubscribe != null)
+            {
+                foreach (var item in options.ListSubscribe)
+                {
+                    if (options.DefaultRegisterWay.ToLower() == "singleton")
+                    {
+                        services.TryAddSingleton(item);
+                    }
+                    else if (options.DefaultRegisterWay.ToLower() == "scoped")
+                    {
+                        services.TryAddScoped(item);
+                    }
+                    else if (options.DefaultRegisterWay.ToLower() == "transient")
+                    {
+                        services.TryAddTransient(item);
+                    }                      
+
+                    else
+                    {
+                        services.TryAddSingleton(item);
+                    }
+                }
+            }
+        }
+        static void c_Conneted(object sender, EventArgs e, ILogger logger)
+        {
+            if (logger == null)
+            {
+                Console.WriteLine("RabbitMQ已链接");
+            }
+            else
+            {
+                
+            }
+        }
+        static void c_DisConnect(object sender, EventArgs e, ILogger logger)
+        {
+            if (logger == null)
+            {
+                Console.WriteLine("RabbitMQ已断开");
+            }
+            else
+            {
+                
+            }
+        }
+
+    }
+
+}
@@ -0,0 +1,123 @@
+using EasyNetQ;
+using EasyNetQ.Topology;
+using RabbitMQHelper.Attributes;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Reflection;
+using System.Threading;
+using System.Threading.Tasks;
+using RabbitMQHelper.Models;
+using MongoDB.Driver;
+
+namespace RabbitMQHelper
+{
+    public class MQHostedService : BackgroundService
+    {
+        private readonly IServiceProvider _serviceProvider;
+        private readonly IBus _bus;
+        private readonly IRabbitConfig _mongoDBConfig;
+        private readonly IMongoCollection<ReceiveRecord> _collection;
+        private static readonly object lockHelper = new object();
+        public MQHostedService(IServiceProvider serviceProvider, IBus bus, IRabbitConfig mongoDBConfig)
+        {
+            _serviceProvider = serviceProvider;
+            _bus = bus;
+            _mongoDBConfig = mongoDBConfig;
+            var database = new MongoClient(_mongoDBConfig.MongoConnectionString).GetDatabase(_mongoDBConfig.MongoDatabaseName);
+            _collection = database.GetCollection<ReceiveRecord>("mq_receive_record");
+        }
+        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+        {
+            var executorDescriptorList = new List<ConsumerExecutorDescriptor>();
+            foreach (Type ty in _mongoDBConfig.ListSubscribe)
+            {
+                executorDescriptorList.AddRange(GetTopicAttributesDescription(ty.GetTypeInfo()));
+            }
+
+
+            var exchange = await _bus.Advanced.ExchangeDeclareAsync(_mongoDBConfig.DefaultExchange, ExchangeType.Topic);
+                        
+            foreach (var item in executorDescriptorList)
+            {
+                using (var scope = _serviceProvider.GetRequiredService<IServiceScopeFactory>().CreateScope())
+                {
+                    var topic = item.Attribute.Topic;
+                    var provider = scope.ServiceProvider;
+                    var obj = ActivatorUtilities.GetServiceOrCreateInstance(provider, item.ImplTypeInfo.GetTypeInfo());
+
+                    var subscriptId = $"{ item.ClassName.ToLower() }.{item.MethodInfo.Name.ToLower()}";
+
+                    var queue = await _bus.Advanced.QueueDeclareAsync(subscriptId);
+
+                    await _bus.Advanced.BindAsync(exchange, queue, topic, null);
+
+                    _bus.Advanced.Consume<RabbitMessage>(queue, (msg, info) => Task.Factory.StartNew(async () =>
+                    {
+                        await ReceiveMessage(msg.Body, info.Queue, info.RoutingKey);
+                        lock (lockHelper)
+                        {
+                            object[] parameters = new object[] { msg.Body.Data, msg.Body.Headers };
+                            item.MethodInfo.Invoke(obj, parameters);
+                        }
+                    }));
+
+                    //await _bus.PubSub.SubscribeAsync<RabbitMessage>(subscriptId, async msg =>
+                    //{
+                    //    await ReceiveMessage(msg, subscriptId, topic);
+                    //    object[] parameters = new object[] { msg.Data, msg.Headers };
+                    //    item.MethodInfo.Invoke(obj, parameters);
+                        
+                    //}, x => x.WithTopic(topic).w);
+                }
+            }
+            await Task.Delay(Timeout.Infinite, stoppingToken);
+        }
+        public async Task ReceiveMessage(RabbitMessage msg,string subscriptId,string topic)
+        {
+            try {
+                //var database = new MongoClient(_mongoDBConfig.MongoConnectionString).GetDatabase(_mongoDBConfig.MongoDatabaseName);
+                //var collection = database.GetCollection<ReceiveRecord>("mq_receive_record");
+                await _collection.InsertOneAsync(new ReceiveRecord(msg.RecordId, subscriptId, topic));
+            }
+            catch (Exception ex)
+            {
+                
+            }
+        }
+        private IEnumerable<ConsumerExecutorDescriptor> GetTopicAttributesDescription(TypeInfo typeInfo)
+        {
+            foreach (var method in typeInfo.DeclaredMethods)
+            {
+                var topicAttr = method.GetCustomAttributes<RabbitMQAttribute>(true);
+                var topicAttributes = topicAttr as IList<RabbitMQAttribute> ?? topicAttr.ToList();
+
+                if (!topicAttributes.Any())
+                {
+                    continue;
+                }
+
+                foreach (var attr in topicAttributes)
+                {
+                    yield return InitDescriptor(attr, method, typeInfo, typeInfo.Name);
+                }
+            }
+        }
+        private ConsumerExecutorDescriptor InitDescriptor(RabbitMQAttribute attr, MethodInfo methodInfo, TypeInfo implType, string name)
+        {
+            var descriptor = new ConsumerExecutorDescriptor
+            {
+                Attribute = attr,
+                MethodInfo = methodInfo,
+                ImplTypeInfo = implType,
+                ClassName = name
+            };
+            return descriptor;
+        }
+    }
+
+}
@@ -0,0 +1,40 @@
+using System;
+
+
+namespace RabbitMQHelper.Models
+{
+    public class PublishRecord
+    {
+        public PublishRecord(string publishId, string topic,object message,bool success)
+        {
+            this.PublishId = publishId;
+            this.Topic = topic;
+            this.MessageContent = message;
+            this.Success = success;
+        }
+        public string id { get; set; } = Guid.NewGuid().ToString("N");
+        public string PublishId { get; set; }
+        public string Topic { get; set; }
+        public object MessageContent { get; set; }
+        public bool Success { get; set; }
+        public DateTime PublishTime { get; set; } = DateTime.Now;
+    }
+
+    public class ReceiveRecord
+    {
+        public ReceiveRecord(string publishId, string subscriptId, string topic)
+        {
+            this.PublishId = publishId;
+            this.SubscriptId = subscriptId;
+            this.Topic = topic;
+        }
+
+        public string id { get; set; } = Guid.NewGuid().ToString("N");
+        public string PublishId { get; set; }
+        public string SubscriptId { get; set; }
+        public string Topic { get; set; }
+        public DateTime ReceiveTime { get; set; } = DateTime.Now;
+
+    }
+}
+
@@ -0,0 +1,27 @@
+{
+  "iisSettings": {
+    "windowsAuthentication": false,
+    "anonymousAuthentication": true,
+    "iisExpress": {
+      "applicationUrl": "http://localhost:60480/",
+      "sslPort": 44373
+    }
+  },
+  "profiles": {
+    "IIS Express": {
+      "commandName": "IISExpress",
+      "launchBrowser": true,
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      }
+    },
+    "RabbitMQHelper": {
+      "commandName": "Project",
+      "launchBrowser": true,
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      },
+      "applicationUrl": "https://localhost:5001;http://localhost:5000"
+    }
+  }
+}
@@ -0,0 +1,33 @@
+using System;
+using System.Collections.Generic;
+
+namespace RabbitMQHelper
+{
+    public interface IRabbitConfig
+    {
+        string Host { get; set; }
+        string VirtualHost { get; set; }
+        string UserName { get; set; }
+        string Password { get; set; }
+        List<Type> ListSubscribe { get; set; }
+        string MongoDatabaseName { get; set; }
+        string MongoConnectionString { get; set; }
+        string DefaultExchange { get; set; }
+
+        string DefaultRegisterWay { get; set; }
+    }
+    public class RabbitConfig: IRabbitConfig
+    {
+        public string Host { get; set; } = "49.233.122.91";
+        public string VirtualHost { get; set; } = "/";
+        public string UserName { get; set; } = "guest";
+        public string Password { get; set; } = "guest";
+        public List<Type> ListSubscribe { get; set; }
+        public string MongoDatabaseName { get; set; }
+        public string MongoConnectionString { get; set; }
+        public string DefaultExchange { get; set; } = "hscw.default.topic";
+
+
+        public string DefaultRegisterWay { get; set; } = "singleton";
+    }
+}
@@ -0,0 +1,18 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>net8.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="DotNetCore.CAP" Version="8.3.4" />
+    <PackageReference Include="EasyNetQ" Version="7.8.0" />
+    <PackageReference Include="Microsoft.AspNetCore.Http" Version="2.3.0" />
+    <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="9.0.7" />
+    <PackageReference Include="Microsoft.Extensions.Logging" Version="9.0.4" />
+    <PackageReference Include="MongoDB.Driver" Version="3.6.0" />
+  </ItemGroup>
+
+</Project>
@@ -0,0 +1,17 @@
+using System;
+using System.Collections.Generic;
+
+namespace RabbitMQHelper
+{
+    public class RabbitMessage
+    {
+        public RabbitMessage(object data, Dictionary<string, string> headers = null)
+        {
+            this.Data = data;
+            this.Headers = headers;
+        }
+        public string RecordId { get; set; } = Guid.NewGuid().ToString("N");
+        public object Data { get; set; }
+        public Dictionary<string,string> Headers { get; set; }
+    }
+}
@@ -0,0 +1,78 @@
+using System;
+using System.Collections.Generic; 
+using System.Threading.Tasks;
+using EasyNetQ;
+using EasyNetQ.Topology;
+using RabbitMQHelper.Models;
+using Microsoft.AspNetCore.Http;
+using Microsoft.Extensions.Logging;
+using MongoDB.Driver;
+
+namespace RabbitMQHelper
+{
+    public class SendMQMessage:ISendMQMessage
+    {
+        private readonly IBus _bus;
+        private readonly ILogger<SendMQMessage> _logger;
+        private readonly IRabbitConfig _mongoDBConfig;
+        private readonly IMongoCollection<PublishRecord> _collection;
+        private readonly IHttpContextAccessor _httpContextAccessor;
+        
+        public SendMQMessage(IBus bus, ILogger<SendMQMessage> logger, IRabbitConfig mongoDBConfig, IHttpContextAccessor httpContextAccessor)
+        {
+            _bus = bus;
+            _logger = logger;
+            _mongoDBConfig = mongoDBConfig;
+            _httpContextAccessor = httpContextAccessor;
+            var database = new MongoClient(_mongoDBConfig.MongoConnectionString).GetDatabase(_mongoDBConfig.MongoDatabaseName);
+            _collection = database.GetCollection<PublishRecord>("mq_publish_record");
+        }
+        public async Task<bool> Send(string topic, object value)
+        {
+            if (!_bus.Advanced.IsConnected)
+            {
+                
+                return false;
+            }
+            Dictionary<string, string> headers = new Dictionary<string, string>();
+
+            if (_httpContextAccessor.HttpContext != null && _httpContextAccessor.HttpContext.Request != null && _httpContextAccessor.HttpContext.Request.Headers != null)
+            {
+                if (_httpContextAccessor.HttpContext.Request.Headers.ContainsKey("token"))
+                {
+                    headers.Add("token", _httpContextAccessor.HttpContext.Request.Headers["token"].ToString() ?? "");
+                }
+
+               ;
+            }
+            bool success = false;
+
+            var model = new RabbitMessage(value, headers);
+
+            var exchange = await _bus.Advanced.ExchangeDeclareAsync(_mongoDBConfig.DefaultExchange, ExchangeType.Topic);
+
+            var message = new Message<RabbitMessage>(model);
+            await _bus.Advanced.PublishAsync<RabbitMessage>(exchange, topic, false, message).ContinueWith(m =>
+              {
+                  if (m.IsCompleted && m.IsCompletedSuccessfully)
+                  {
+                      success = true;
+                  }
+                  else
+                  {
+                      _logger.LogError(m.Exception, "RabbitMQ消息推送失败 topic={Topic}", topic);
+                  }
+              });
+            try
+            {
+                await _collection.InsertOneAsync(new PublishRecord(model.RecordId, topic, value, success));
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "RabbitMQ推送记录写入异常 topic={Topic}", topic);
+            }
+            
+            return success;
+        }
+    }
+}