侧边栏壁纸
博主头像
进击的码农博主等级

新年新气象,开搞开搞

  • 累计撰写 31 篇文章
  • 累计创建 11 个标签
  • 累计收到 1 条评论

目 录CONTENT

文章目录

RabbitMQ 封装事件总线基本功能

wosperry
2021-12-05 / 0 评论 / 0 点赞 / 6 阅读 / 16055 字

RabbitMQ 封装

代码

https://gitee.com/wosperry/wosperry-rabbit-mqtest/tree/master

参考Abp事件总线的用法,对拷贝的Demo进行简单封装

定义 RabbitMQOptions 用于配置

我希望可以通过修改appsettings.json配置RabbitMQ的地址、密码等信息

{ 
  "MyRabbitMQOptions": {
    "UserName": "admin",
    "Password": "admin",
    "Host": "192.168.124.220",
    "Port": 5672,
    "ExchangeName": "PerryExchange"
  }
}

    public class MyRabbitMQOptions
    {
        public string UserName { get; set; }
        public string Password { get; set; }
        public string Host { get; set; }
        public int Port { get; set; }
        public string ExchangeName { get; set; } = "";
    }

定义 QueueNameAttribute 控制队列名字

由于一开始写的时候,直接使用类的FullName为队列名称,不太友好,所以单独加了一个特性,用于更改队列名

    /// <summary>
    /// 定义队列名字,优先级高于类完整名
    /// </summary>
    [AttributeUsage(AttributeTargets.Class, AllowMultiple = false, Inherited = false)]
    public class QueueNameAttribute : Attribute
    {
        public string QueueName { get; }
        public QueueNameAttribute(string queueName)
        {
            QueueName = queueName;
        }
    }

定义 IMyPublisher<T>,通过注入某个类型的 IMyPublisher<T>,自动序列化对象并发布到配置好的MQ里

我希望可以一次配置之后,在使用到的地方直接注入一个服务,调用某一个方法,即可发布消息,不需要在意具体怎么实现的。于是定义一个接口,只添加了一个方法。
然后在接口实现类里完成实体转换成消息内容并发送到RabbitMQ中去。
具体实现可以看代码注释,关键地方都加了描述

    /// <summary>
    /// 用于注入使用
    /// </summary>
    public interface IMyEventHandler<T> : IMyEventHandler where T : class
    {
        Task PublishAsync(T data, Encoding encoding = null);
    } 
    /// <summary>
    /// 方便程序寻找IMyEventHandler的实现
    /// </summary>
    public interface IMyEventHandler
    {
        void Begin(IConnection connection);
    }
    public class MyPublisher<T> : IMyPublisher<T>, IDisposable where T : class
    {
        private readonly MyRabbitMQOptions _myOptions;
        private readonly IConnection _connection;
        private readonly IModel _channel;
        private readonly string _queueName;


        /// <summary>
        /// 非注入时使用此构造方法
        /// </summary>
        public MyPublisher(IConnection connection)
        {
            _connection = connection;
        }

        /// <summary>
        /// 依赖注入自动走这个构造方法
        /// </summary>
        /// <param name="optionsMonitor"></param>
        /// <param name="factory"></param>
        public MyPublisher(IOptionsMonitor<MyRabbitMQOptions> optionsMonitor, ConnectionFactory factory)
        {
            _myOptions = optionsMonitor.CurrentValue;
            _connection = factory.CreateConnection();

            // 创建通道
            _channel = _connection.CreateModel();

            // 声明一个Exchange
            _channel.ExchangeDeclare(_myOptions.ExchangeName, ExchangeType.Direct, false, false, null);


            var type = typeof(T);
            // 获取类上的QueueNameAttribute特性,如果不存在则使用类的完整名
            var attr = type.GetCustomAttribute<QueueNameAttribute>();
            _queueName = string.IsNullOrWhiteSpace(attr?.QueueName) ? type.FullName : attr.QueueName;

            // 声明一个队列 
            _channel.QueueDeclare(_queueName, false, false, false, null);

            //将队列绑定到交换机
            _channel.QueueBind(_queueName, _myOptions.ExchangeName, _queueName, null);
        }

        /// <summary>
        /// 发布消息
        /// </summary>
        public Task PublishAsync(T data, Encoding encoding = null)
        {
            // 对象转 object[] 发送
            var msg = JsonConvert.SerializeObject(data);
            byte[] bytes = (encoding ?? Encoding.UTF8).GetBytes(msg);
            _channel.BasicPublish(_myOptions.ExchangeName, _queueName, null, bytes);

            return Task.CompletedTask;
        }

        public void Dispose()
        {
            // 结束
            _channel.Close();
            _connection.Close();
        }
    }

定义 IMyEventHandler<T> ,供 NetCore 项目注入使用,配置后,可以在程序启动的时候,找到该接口所有的实现类,并开启消费者

我希望可以像Abp那样,通过一个抽象类基类,自动获取所有实现了这个抽象类的派生类,自动开启对应的消息监听,监听到消息后,自动转回对象类型,并执行派生类里的某个重写方法。
具体实现可以查看代码注释

    /// <summary>
    /// Handler的配置
    /// </summary>
    public class MyEventHandlerOptions
    {
        /// <summary>
        /// 禁用 byte[] 解析
        /// </summary>
        public bool DisableDeserializeObject { get; set; } = false;
        /// <summary>
        /// 配置Encoding
        /// </summary>
        public Encoding Encoding { get; set; } = Encoding.UTF8;
    }
    public abstract class MyEventHandler<T> : IMyEventHandler<T> where T : class
    {
        private IModel _channel;
        private string _queueName;
        private EventingBasicConsumer _consumer;
        public MyEventHandlerOptions Options = new()
        {
            DisableDeserializeObject = false
        };

        public void Begin(IConnection connection)
        {
            var type = typeof(T);
            // 获取类上的QueueNameAttribute特性,如果不存在则使用类的完整名
            var attr = type.GetCustomAttribute<QueueNameAttribute>();
            _queueName = string.IsNullOrWhiteSpace(attr?.QueueName) ? type.FullName : attr.QueueName;

            //创建通道
            _channel = connection.CreateModel();

            _consumer = new EventingBasicConsumer(_channel);
            _consumer.Received += MyReceivedHandler;
            //消费者
            _channel.BasicConsume(_queueName, false, _consumer);
        }

        // 收到消息后
        private void MyReceivedHandler(object sender, BasicDeliverEventArgs e)
        {
            try
            {
                // 反序列化为对象
                var message = Options.Encoding.GetString(e.Body);
                T data = null;
                // 如果未配置禁用则不解析,后面抽象方法的data参数会始终为空
                if (!Options.DisableDeserializeObject)
                {
                    data = JsonConvert.DeserializeObject<T>(message);
                }

                OnReceivedAsync(data, message).Wait();

                // 确认该消息已被消费
                _channel?.BasicAck(e.DeliveryTag, false);
            }
            catch (Exception ex)
            {
                OnConsumerException(ex);
            }
        }

        /// <summary>
        /// 收到消息 
        /// </summary>
        /// <param name="data">解析后的对象</param>
        /// <param name="message">消息原文</param> 
        /// <remarks>Options.DisableDeserializeObject为true时,data始终为null</remarks>
        public abstract Task OnReceivedAsync(T data, string message);

        /// <summary>
        /// 异常
        /// </summary>
        /// <param name="ex">派生类不重写的话,异常被隐藏</param>
        public virtual void OnConsumerException(Exception ex)
        {

        }
    }


给依赖注入写一些拓展方法

这里是因为如果要宿主机程序配置Ioc容器注册的话,需要在宿主程序写好些个代码,为了减少使用时候的配置,把所有相关的Ioc容器注册步骤封装到拓展方法,减少使用时的代码量

另外,事件消息消费者的启动在有两个方法,实现逻辑是先在app尚未被创建之前,把需要启动的EventHandler注册到容器里(可以直接使用程序集的所有类)
在app创建之后,调用UseMyEventHandler时,会从Ioc容器里找到所有的实现,并调用他们的Begin方法。

    public static class MyRabbiteMQExtensions
    {
        /// <summary>
        /// 初始化消息队列,并添加Publisher到IoC容器
        /// </summary>
        /// <remarks>从Configuration读取"MyRabbbitMQOptions配置项"</remarks>
        public static IServiceCollection AddMyRabbitMQ(this IServiceCollection services, IConfiguration configuration)
        {
            #region 配置项
            // 从Configuration读取"MyRabbbitMQOptions配置项
            var optionSection = configuration.GetSection("MyRabbitMQOptions");

            // 这个myOptions是当前方法使用
            MyRabbitMQOptions myOptions = new();
            optionSection.Bind(myOptions);

            // 加了这行,才可以注入IOptions<MyRabbitMQOptions>或者IOptionsMonitor<MyRabbitMQOptions>
            services.Configure<MyRabbitMQOptions>(optionSection);
            #endregion

            // 加了这行,才可以注入任意类型参数的 IMyPublisher<> 使用
            services.AddTransient(typeof(IMyPublisher<>), typeof(MyPublisher<>));

            // 创建一个工厂对象,并配置单例注入
            services.AddSingleton(new ConnectionFactory
            {
                UserName = myOptions.UserName,
                Password = myOptions.Password,
                HostName = myOptions.Host,
                Port = myOptions.Port
            });

            return services;
        }

        /// <summary>
        /// IServiceCollection的拓展方法,用于发现自定义的EventHandler并添加到服务容器
        /// </summary> 
        /// <param name="types">包含了自定义Handler的类集合,可以使用assembly.GetTypes()</param> 
        /// <remarks>遍历所有types,将继承自IMyEventHandler的类注册到容器</remarks>
        public static IServiceCollection AddMyRabbitMQEventHandlers(this IServiceCollection services, Type[] types)
        {
            var baseType = typeof(IMyEventHandler);

            foreach (var type in types)
            {
                // baseType可以放type,并且type不是baseType
                if (baseType.IsAssignableFrom(type) && baseType != type)
                {
                    // 瞬态注入配置
                    services.AddTransient(typeof(IMyEventHandler), type);
                }
            }

            return services;
        }

        /// <summary>
        /// 给app拓展方法
        /// </summary>
        /// <remarks>
        /// 在IoC容器里获取到所有继承自IMyEvetnHandler的实现类,并开启消费者
        /// </remarks>
        public static IApplicationBuilder UseMyEventHandler(this IApplicationBuilder app)
        {
            var handlers = app.ApplicationServices.GetServices(typeof(IMyEventHandler));
            var factory = app.ApplicationServices.GetService<ConnectionFactory>();

            // 遍历调用自定义的Begin方法
            foreach (var h in handlers)
            {
                var handler = h as IMyEventHandler;
                handler?.Begin(factory.CreateConnection());
            }

            return app;
        }
    }

在Net6 WebApi中使用

这个项目是引用了上方封装的一个独立的宿主项目,
如果仅使用发布者,则只需要提供appsettings.json里的配置,和 services.AddMyRabbitMQ(builder.Configuration); 方法即可。

如果需要使用事件队列消费者,则需要添加两个方法调用,一是把EventHandler添加到容器中,二是需要在app创建后调用UseMyEventHandler();
代码执行如上文所说,会从Ioc容器里找到所有的实现,并调用他们的Begin方法开启监听。

program.cs

    var builder = WebApplication.CreateBuilder(args);
    builder.Services.AddControllers();


    // 添加MyRabbitMQ到services
    builder.Services.AddMyRabbitMQ(builder.Configuration);

    // 当前程序集所有类型
    var currentAssemblyTypes = Assembly.GetExecutingAssembly().GetTypes();
    builder.Services.AddMyRabbitMQEventHandlers(currentAssemblyTypes ); 

    // 上面参数可以像下面这样配置
    // builder.Services.AddMyRabbitMQEventHandlers(new Type[]{typeof(MyHandler1), typeof(MyHandler2), typeof(MyHandler3)}); 
    // builder.Services.AddMyRabbitMQEventHandlers(typeof(MyHandler1), typeof(MyHandler2), typeof(MyHandler3)); 

    var app = builder.Build();

    // 使用 MyEventHandler
    app.UseMyEventHandler();


    app.MapControllers();
    app.Run();

定义ETO
其实是一个普通的类
这个类有两个用处,

一是要给是在使用时注入IPublisher的时候,指定消息的实体是什么。
二是写消息消费者时,需要给BaseEventHandler指定消息的实体是什么。这样在接收到消息之后,程序就知道该反序列化成什么类型的对象。

    [QueueName("perry.test")]
    public class PerryTest
    {
        public Guid Id { get; set; }
        public string? Name { get; set; }
        public int Count { get; set; }
        public string? Remark { get; set; }
    }

定义EventHandler

只需要继承了MyEventHandler并提供一个普通类型,则可以监听这个类型的消息,队列名字可以通过[QueueName("队列名")]修改

    public class PerryTestEventHandler : MyEventHandler<PerryTest>
    {
        public override Task OnReceivedAsync(PerryTest data, string message)
        {
            Console.WriteLine(message);
            return Task.CompletedTask;
        }

        public override void OnConsumerException(Exception ex)
        {
            Console.WriteLine(ex.Message);
        }
    }

控制器中检查是否可以正常使用

指定类型注入IMyPublisher<>,即可使用PublishAsync(data)方法发送消息,队列名字可以通过[QueueName("队列名")]修改

    [Route("api")]
    [ApiController]
    public class TestController : ControllerBase
    {
        public IMyPublisher<PerryTest> TestPublisher { get; }

        public TestController(IMyPublisher<PerryTest> testPublisher)
        {
            TestPublisher = testPublisher;
        }
        [HttpGet("test")]
        public async Task<string> TestAsync()
        {
            var data = new PerryTest()
            {
                Id = Guid.NewGuid(),
                Name = "AAA",
                Count = 123,
                Remark = "哈哈哈"
            };

            await TestPublisher.PublishAsync(data);

            return "发送了一个消息";
        }
    }

运行截图

参考 .NET Core 使用RabbitMQ,拷贝了一些Demo

文章里的生产者Demo

    //创建连接工厂
    ConnectionFactory factory = new ConnectionFactory
    {
        UserName = "admin",//用户名
        Password = "admin",//密码
        HostName = "192.168.157.130"//rabbitmq ip
    };

    //创建连接
    var connection = factory.CreateConnection();
    //创建通道
    var channel = connection.CreateModel();
    //声明一个队列
    channel.QueueDeclare("hello", false, false, false, null);

    Console.WriteLine("\nRabbitMQ连接成功,请输入消息,输入exit退出!");

    string input;
    do
    {
        input = Console.ReadLine();
        var sendBytes = Encoding.UTF8.GetBytes(input);
        //发布消息
        channel.BasicPublish("", "hello", null, sendBytes);
    } while (input.Trim().ToLower()!="exit");
    channel.Close();
    connection.Close();

文章里的消费者Demo

    //创建连接工厂
    ConnectionFactory factory = new ConnectionFactory
    {
        UserName = "admin",//用户名
        Password = "admin",//密码
        HostName = "192.168.157.130"//rabbitmq ip
    };

    //创建连接
    var connection = factory.CreateConnection();
    //创建通道
    var channel = connection.CreateModel();

    //事件基本消费者
    EventingBasicConsumer consumer = new EventingBasicConsumer(channel);

    //接收到消息事件
    consumer.Received += (ch, ea) =>
    {
        var message = Encoding.UTF8.GetString(ea.Body);
        Console.WriteLine($"收到消息: {message}");
        //确认该消息已被消费
        channel.BasicAck(ea.DeliveryTag, false);
    };
    //启动消费者 设置为手动应答消息
    channel.BasicConsume("hello", false, consumer);
    Console.WriteLine("消费者已启动");
    Console.ReadKey();
    channel.Dispose();
    connection.Close();

这次封装的总结

  1. 上网找个Demo
  2. 先按最简单的写法,写完能正常使用的,功能独立的代码。其实是提醒自己,不要陷入到杂七杂八的各项优化中去,先写完实现再考虑怎么去改成更加好的。
  3. 脑袋里想着,我要怎么样使用这个功能,怎么样才能让用的时候写的代码少一些,配置简单一些
  4. 检查哪些内容应该分离到配置项,然后抽离出去
  5. 考虑要支持哪些类型的项目,如果想要支持低版本,可能需要降级一些依赖包
  6. 支持构造函数注入的话,要注意最多参数的构造函数给依赖注入使用,依赖注入用的构造函数不好被非注入时使用的话,考虑多提供一个方法给非注入时使用。
  7. 最后,希望这篇文章能帮助到正在学习MQ或者ABP的事件总线的朋友们
0

评论区