开源RabbitMQ操作组件

时间：2022-05-06 20:45:11

开源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的实现并将其注册到容器。

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

开源RabbitMQ操作组件

三.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()：

开源RabbitMQ操作组件
 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>();

        }
开源RabbitMQ操作组件

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

 2.IBus.Publish()：

开源RabbitMQ操作组件
  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);

        }
开源RabbitMQ操作组件

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

五.总结

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

相关文章