开源RabbitMQ操作组件

Posted 2020-09-09 Leo_wlCnBlogs

开源RabbitMQ操作组件

对于目前大多的.NET项目，其实使用的技术栈都是差不多，估计现在很少用控件开发项目的了，毕竟一大堆问题。对.NET的项目，目前比较适合的架构ASP.NET MVC，ASP.NET WebAPI，ORM（较多Dapper.NET或者其扩展，稍大一些的项目用EF等等），为了提高速度也会采用缓存（.NET自带的Memcache，或者Redis），请求较多的项目，使用nginx做负载均衡和使用队列等等。

   上面简单的介绍一下.NET的项目的技术架构，具体的技术根据具体的需求做出选择。介绍到队列，很多人都会很熟悉，例如MSMQ，RabbitMQ等等队列。既然需要使用队列，那就要考虑如何使用C#更好的操作队列。

一.RabbitMQ概述

   在现在的项目中，消息队列的使用比较的频繁，消息队列的种类也较多，如：ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ等。消息队列中间件是分布式系统中重要的组件，主要解决应用耦合，异步消息，流量削锋等问题。实现高性能，高可用，可伸缩和最终一致性架构。是大型分布式系统不可缺少的中间件。

   在这里主要介绍RabbitMQ消息队列，支持开放的高级消息队列协议 (AMQP)。RabbitMQ的特点：强大的应用程序消息传递；使用方便；运行在所有主要操作系统上；支持大量开发人员平台；开源和商业支持。消息队列的模式有两种模式：P2P（Point to Point），P2P模式包含三个角色：消息队列（Queue），发送者(Sender)，接收者(Receiver)。每个消息都被发送到一个特定的队列，接收者从队列中获取消息。队列保留着消息，直到他们被消费或超时。Publish/Subscribe(Pub/Sub)，包含三个角色主题（Topic），发布者（Publisher），订阅者（Subscriber） 。多个发布者将消息发送到Topic,系统将这些消息传递给多个订阅者。

   上面介绍了RabbitMQ的相关特点和模式，更多的知识就不再介绍，需要了解安装和配置，可以进入官网进行细致的了解。

二.EasyNetQ组件概述

    上面介绍了RabbitMQ的应用场景和使用的模式，在.NET的项目开发中，较多的使用MSMQ作为消息队列，很多人对于MSMQ的操作比较熟悉，也属于轻量级的消息队列。对于RabbitMQ是较为重量级的消息队列，有多个语言的版本，作为.NET开发者对于RabbitMQ的操作可能就比较少。在.NET项目中如何更方便的使用RabbitMQ，在这里就介绍一个.NET操作RabbitMQ的组件EasyNetQ。

    EasyNetQ的目标是提供一个使.NET中的RabbitMQ尽可能简单的库。在EasyNetQ中消息应由.NET类型表示，消息应通过其.NET类型进行路由。EasyNetQ按消息类型进行路由。发布消息时，EasyNetQ会检查其类型，并根据类型名称，命名空间和装配体给出一个路由密钥。在消费方面，用户订阅类型。订阅类型后，该类型的消息将路由到订户。默认情况下，EasyNetQ使用Newtonsoft.Json库将.NET类型序列化为JSON。这具有消息是人类可读的优点，因此您可以使用RabbitMQ管理应用程序等工具来调试消息问题。

   EasyNetQ是在RabbitMQ.Client库之上提供服务的组件集合。这些操作可以像序列化，错误处理，线程编组，连接管理等。它们由mini-IoC容器组成。您可以轻松地用自己的实现替换任何组件。因此，如果您希望XML序列化而不是内置的JSON，只需编写一个ISerializer的实现并将其注册到容器。

   以下是官方提供的一个结构图，这个结构图可以很好的解析该组件的结构：

 

三.EasyNetQ组件使用方式

   介绍完毕EasyNetQ组件的相关背景，现在就要介绍一下该组件的使用方式。EasyNetQ组件的使用方式比较简单，跟很多组件都类似，例如：建立连接，进行操作做等等，对于EasyNetQ组件也是如此。

   1.创建连接：

var bus = RabbitHutch.CreateBus(“host=myServer;virtualHost=myVirtualHost;username=mike;password=topsecret”);

    与RabbitMQ服务器的延迟连接由IBus接口表示，创建连接的方式连接字符串由格式为key = value的键/值对组成，每一个用分号（;）分隔。host：主机地址；virtualHost：默认是默认的虚拟主机\'/\'；username：用户名，默认为\'guest\'；password：密码，默认是\'guest\'；

   2.关闭连接：

bus.Dispose();

   要关闭连接，只需简单地处理总线，这将关闭EasyNetQ使用的连接，渠道，消费者和所有其他资源。

   3.发布消息：

var message = new MyMessage { Text = "Hello Rabbit" };
bus.Publish(message);

   4.订阅邮件：

bus.Subscribe<MyMessage>("my_subscription_id", msg => Console.WriteLine(msg.Text));

   5.远程过程调用：

var request = new TestRequestMessage {Text = "Hello from the client! "};
bus.Request<TestRequestMessage, TestResponseMessage>(request, response => 
    Console.WriteLine("Got response: \'{0}\'", response.Text));

   6.RPC服务器：

bus.Respond<TestRequestMessage, TestResponseMessage>(request => 
    new TestResponseMessage{ Text = request.Text + " all done!" });

   7.记录器：

var logger = new MyLogger() ;
var bus = RabbitHutch.CreateBus(“my connection string”, x => x.Register<IEasyNetQLogger>(_ => logger));

   8.路由：

bus.Subscribe("my_id", handler, x => x.WithTopic("X.*"));

    RabbitMQ具有非常好的功能，基于主题的路由，允许订阅者基于多个标准过滤消息。*（星号）匹配一个字。＃（哈希）匹配为零个或多个单词。

四.EasyNetQ组件核心对象解析

    上面简单的介绍了一下该组件的应用方式，还有比较多的方式没有做介绍，又需要的可以做深入的了解。在这里介绍一下该组件的一些核心的对象。

  1.RabbitHutch.CreateBus()：

 public static IBus CreateBus(ConnectionConfiguration connectionConfiguration, AdvancedBusEventHandlers advancedBusEventHandlers, 
             Action<IServiceRegister> registerServices)
        {
            Preconditions.CheckNotNull(connectionConfiguration, "connectionConfiguration");
            Preconditions.CheckNotNull(advancedBusEventHandlers, "advancedBusEventHandlers");
            Preconditions.CheckNotNull(registerServices, "registerServices");
            var container = createContainerInternal();
            if (container == null)
            {
                throw new EasyNetQException("Could not create container. " +
                    "Have you called SetContainerFactory(...) with a function that returns null?");
            }
            connectionConfiguration.Validate();
            container.Register(_ => connectionConfiguration);
            container.Register(_ => advancedBusEventHandlers);
            registerServices(container);
            ComponentRegistration.RegisterServices(container);
            return container.Resolve<IBus>();
        }

   在RabbitHutch类中主要包含的方法是CreateBus()方法，具有12个重载。该方法主要根据用户的连接配置信息，连接服务端。该方法接收三个参数，connectionConfiguration表示连接实例，advancedBusEventHandlers用于添加处理程序的AdvancedBusEventHandlers实例到新创建的IBus.Advanced”的事件。registerServices覆盖默认服务。 ComponentRegistration.RegisterServices(container);在我们内部的超简单IoC容器中注册默认的EasyNetQ组件。container.Resolve<IBus>()获取所请求的服务的实例。 注意所有服务都是单例的，多次通话Resolve将返回相同的实例。

   2.IBus.Publish()：

  public virtual void Publish<T>(T message, Action<IPublishConfiguration> configure) where T : class
        {
            Preconditions.CheckNotNull(message, "message");
            Preconditions.CheckNotNull(configure, "configure");
            var configuration = new PublishConfiguration(conventions.TopicNamingConvention(typeof(T)));
            configure(configuration);
            var messageType = typeof(T);
            var easyNetQMessage = new Message<T>(message)
            {
                Properties =
                {
                    DeliveryMode = messageDeliveryModeStrategy.GetDeliveryMode(messageType)
                }
            };
            if (configuration.Priority != null)
                easyNetQMessage.Properties.Priority = configuration.Priority.Value;
            if (configuration.Expires != null)
                easyNetQMessage.Properties.Expiration = configuration.Expires.ToString();
            var exchange = publishExchangeDeclareStrategy.DeclareExchange(advancedBus, messageType, ExchangeType.Topic);
            advancedBus.Publish(exchange, configuration.Topic, false, easyNetQMessage);
        }

     该方法用于发布消息，该方法是一个虚方法，在子类中可以被重写。 var configuration = new PublishConfiguration(conventions.TopicNamingConvention(typeof(T)))用于定义发布信息的配置，Message定义邮件正文内容。

五.总结

   以上是对该组件的简单的介绍，如果需要了解更多的内容可以自己去深入的学习和研究。知识在于自己的勤奋，他人只是一个简单的引导。