Merge pull request 'v2' (#418) from eithday/Admin.NET.Pro:v2 into v2

Reviewed-on: https://code.adminnet.top/Admin.NET/Admin.NET.Pro/pulls/418

Admin.NET/Admin.NET.Application/Configuration/EventBus.json

@@ -1,9 +1,6 @@
 {
   "$schema": "https://gitee.com/dotnetchina/Furion/raw/v4/schemas/v4/furion-schema.json",
 
-  "EventBus": {
-    // 事件源存储器类型，默认内存存储（Redis则需要配合缓存相关配置）
-    "EventSourceType": "Memory", // Memory、Redis、RabbitMQ、Kafka
   "RabbitMQ": {
     "UserName": "adminnet",
     "Password": "adminnet++123456",
@@ -11,6 +8,11 @@
     "Port": 5672,
     "VirtualHost": "/"
   },
+
+  "EventBus": {
+    // 事件源存储器类型，默认内存存储（Redis则需要配合缓存相关配置）
+    "EventSourceType": "RabbitMQ", // Memory、Redis、RabbitMQ、Kafka
+
     "Kafka": {
     }
   }

Admin.NET/Admin.NET.Application/Service/Test/RabbitMqHandler.cs

@@ -0,0 +1,20 @@
+// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
+//
+// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
+//
+// 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！
+
+namespace Admin.NET.Application;
+public class RabbitMqHandler : IMessageHandler
+{
+    public string QueueName => "admin.net.rabbitmq";//队列名称
+
+    //消息处理函数
+    public async Task HandleMessageAsync(string message)
+    {
+        //Todo 
+        Console.WriteLine(message);
+
+        await Task.CompletedTask;
+    }
+}

Admin.NET/Admin.NET.Application/Service/Test/TestService.cs

@@ -5,6 +5,7 @@
 // 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！
 
 using Admin.NET.Plugin.WorkWeixin;
+using AngleSharp.Dom;
 using Furion.Localization;
 using Furion.Logging;
 using Microsoft.Extensions.Logging;
@@ -21,13 +22,16 @@ public class TestService : IDynamicApiController
     private readonly IEventPublisher _eventPublisher;
     private readonly ILogger _logger;
     private readonly WorkWxUserService _workWxUserService;
+    private readonly RabbitMqProducer _producer;
 
-    public TestService(UserManager userManager, IEventPublisher eventPublisher, WorkWxUserService workWxUserService, ILoggerFactory loggerFactory)
+    public TestService(UserManager userManager, IEventPublisher eventPublisher, WorkWxUserService workWxUserService, ILoggerFactory loggerFactory, RabbitMqProducer producer)
     {
         _userManager = userManager;
         _eventPublisher = eventPublisher;
         _workWxUserService = workWxUserService;
         _logger = loggerFactory.CreateLogger(CommonConst.SysLogCategoryName); // 日志过滤标识（会写入数据库）
+        _producer = producer;
+
     }
 
     [HttpGet("helloWord")]
@@ -90,4 +94,12 @@ public class TestService : IDynamicApiController
     {
         return await App.GetRequiredService<SysFileService>().UploadFile(input);
     }
+
+    [AllowAnonymous]
+    [ApiDescriptionSettings(Name = "SendMessageToRabbitMQ", Description = "测试")]
+    public async Task SendMessageToRabbitMQ([FromQuery] string input)
+    {
+        await _producer.SendMessage("admin.net.rabbitmq",$"Hello {input}!");
+    }
+
 }

Admin.NET/Admin.NET.Application/Startup.cs

@@ -6,6 +6,7 @@
 
 using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Hosting;
+using Microsoft.Extensions.Hosting;
 
 namespace Admin.NET.Application;
 
@@ -14,6 +15,14 @@ public class Startup : AppStartup
 {
     public void ConfigureServices(IServiceCollection services)
     {
+        // 注册RabbitMQ服务
+        services.AddRabbitMQ<RabbitMqHandler>();
+
+        // 后台服务异常，不影响主程序（配置的服务器连接异常）
+        services.Configure<HostOptions>(options =>
+        {
+            options.BackgroundServiceExceptionBehavior = BackgroundServiceExceptionBehavior.Ignore;
+        });
     }
 
     public void Configure(IApplicationBuilder app, IWebHostEnvironment env)

Admin.NET/Admin.NET.Core/RabbitMQ/IMessageHandler.cs

@@ -0,0 +1,11 @@
+// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
+//
+// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
+//
+// 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！
+namespace Admin.NET.Core;
+public interface IMessageHandler
+{
+    string QueueName { get; }
+    Task HandleMessageAsync(string message);
+}

Admin.NET/Admin.NET.Core/RabbitMQ/RabbitMQExtention.cs

@@ -0,0 +1,23 @@
+// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
+//
+// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
+//
+// 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！
+namespace Admin.NET.Core;
+public static class RabbitMQExtention
+{
+    /// <summary>
+    /// 注册RabbitMQ
+    /// </summary>
+    /// <param name="services"></param>
+    /// <returns></returns>
+    public static IServiceCollection AddRabbitMQ<THandler>(this IServiceCollection services)
+        where THandler : class, IMessageHandler
+    {
+        services.AddSingleton<IMessageHandler, THandler>();
+        services.AddSingleton<RabbitMqConnection>();
+        services.AddSingleton<RabbitMqProducer>();
+        services.AddHostedService<RabbitMqConsumer>();
+        return services;
+    }
+}

Admin.NET/Admin.NET.Core/RabbitMQ/RabbitMqConnection.cs

@@ -0,0 +1,47 @@
+// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
+//
+// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
+//
+// 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！
+
+using RabbitMQ.Client;
+
+namespace Admin.NET.Core;
+public class RabbitMqConnection
+{
+    private IConnection _connection;
+    private IChannel _channel;
+    public IChannel Channel => _channel;
+
+    public async Task<bool> TryConnectAsync()
+    {
+        try
+        {
+            var options = App.GetConfig<RabbitMqOptions>("RabbitMQ");
+            var factory = new ConnectionFactory
+            {
+                HostName = options.HostName,
+                UserName = options.UserName,
+                Password = options.Password,
+                Port = options.Port,
+                VirtualHost = options.VirtualHost
+            };
+
+            _connection = await factory.CreateConnectionAsync();
+            _channel = await _connection.CreateChannelAsync();
+            return true;
+        }
+        catch(Exception ex) 
+        {
+            Console.WriteLine($"RabbitMQ 连接失败: {ex.Message}");
+            return false;
+        }
+        
+    }
+
+    public void Dispose()
+    {
+        _channel?.Dispose();
+        _connection?.Dispose();
+    }
+}

Admin.NET/Admin.NET.Core/RabbitMQ/RabbitMqConsumer.cs

@@ -0,0 +1,50 @@
+// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
+//
+// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
+//
+// 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
+
+namespace Admin.NET.Core;
+public class RabbitMqConsumer : BackgroundService
+{
+    private readonly RabbitMqConnection _connection;
+    private readonly IMessageHandler _handler;
+    private readonly ILogger _logger;
+
+    public RabbitMqConsumer(RabbitMqConnection connection, IMessageHandler handler, ILoggerFactory loggerFactory)
+    {
+        _connection = connection ?? throw new ArgumentNullException(nameof(connection));
+        _handler = handler ?? throw new ArgumentNullException(nameof(handler));
+        _logger = loggerFactory.CreateLogger(CommonConst.SysLogCategoryName); // 日志过滤标识（会写入数据库）
+    }
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        var channel = _connection.Channel;
+        if (channel == null)
+            throw new InvalidOperationException("RabbitMQ channel is null");
+
+        if (!await _connection.TryConnectAsync())
+        {
+            _logger.LogError($"RabbitMQ连接失败，请检查配置文件。");
+            return;
+        }
+        await channel.QueueDeclareAsync(queue: _handler.QueueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
+        await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false);
+        var consumer = new AsyncEventingBasicConsumer(channel);
+
+        consumer.ReceivedAsync += async (sender, ea) =>
+        {
+            byte[] body = ea.Body.ToArray();
+            var message = Encoding.UTF8.GetString(body);
+
+            await _handler.HandleMessageAsync(message);
+
+            Console.WriteLine($"[x] Received from {_handler.QueueName}: {message}");
+            await channel.BasicAckAsync(deliveryTag: ea.DeliveryTag, multiple: false);
+        };
+
+        await channel.BasicConsumeAsync(queue: _handler.QueueName, autoAck: false, consumer: consumer);
+    }
+}

Admin.NET/Admin.NET.Core/RabbitMQ/RabbitMqOptions.cs

@@ -0,0 +1,34 @@
+// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
+//
+// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
+//
+// 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！
+
+namespace Admin.NET.Core;
+public class RabbitMqOptions :  IConfigurableOptions
+{
+    /// <summary>
+    /// 账号
+    /// </summary>
+    public string UserName { get; set; }
+
+    /// <summary>
+    /// 密码
+    /// </summary>
+    public string Password { get; set; }
+
+    /// <summary>
+    /// 主机
+    /// </summary>
+    public string HostName { get; set; }
+
+    /// <summary>
+    /// 端口
+    /// </summary>
+    public int Port { get; set; }
+
+    /// <summary>
+    /// 虚拟主机
+    /// </summary>
+    public string VirtualHost { get; set; } = "/";
+}

Admin.NET/Admin.NET.Core/RabbitMQ/RabbitMqProducer.cs

@@ -0,0 +1,30 @@
+// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
+//
+// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
+//
+// 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！
+
+using RabbitMQ.Client;
+
+namespace Admin.NET.Core;
+public class RabbitMqProducer
+{
+    private readonly RabbitMqConnection _connection;
+
+    public RabbitMqProducer(RabbitMqConnection connection)
+    {
+        _connection = connection;
+    }
+
+    public async Task SendMessage(string queueName, string message)
+    {
+        var channel = _connection.Channel;
+        await channel.QueueDeclareAsync(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
+        var properties = new BasicProperties
+        {
+            Persistent = true
+        };
+        var body = Encoding.UTF8.GetBytes(message);
+        await channel.BasicPublishAsync(exchange: string.Empty, routingKey: queueName, mandatory: true, basicProperties: properties, body: body);
+    }
+}

Admin.NET/Admin.NET.Web.Core/Startup.cs

@@ -13,6 +13,7 @@ using Furion.SpecificationDocument;
 using Furion.VirtualFileServer;
 using IGeekFan.AspNetCore.Knife4jUI;
 using IPTools.Core;
+using Mapster;
 using Microsoft.AspNetCore.Authentication.JwtBearer;
 using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Hosting;
@@ -231,6 +232,9 @@ public class Startup : AppStartup
             }
             else if (eventBusOpt.EventSourceType == "RabbitMQ")
             {
+                var rmq = App.GetConfig<RabbitMqOptions>("RabbitMQ");
+                eventBusOpt.RabbitMQ = rmq.Adapt<RabbitMQSettings>();
+
                 // RabbitMQ消息队列
                 var rbmqEventSourceStorer = new RabbitMQEventSourceStore(new ConnectionFactory
                 {