189 8069 5689

.NET操作RabbitMQ组件EasyNetQ使用中文简版文档。

EasyNetQ简介

EasyNetQ是基于官方.NET组件RabbitMQ.Client 的又一层封装，使用起来更加方便，开发者不用关心具体队列声明，路由声明等细节，几句简单代码即可发送消息到队列，接收消息也很简单，下面将简单介绍EasyNetQ的使用方法。不知道什么是RabbitMQ？您可以关闭网页了。

成都创新互联公司"三网合一"的企业建站思路。企业可建设拥有电脑版、微信版、手机版的企业网站。实现跨屏营销，产品发布一步更新，电脑网络+移动网络一网打尽，满足企业的营销需求！成都创新互联公司具备承接各种类型的成都网站制作、网站设计、外贸网站建设项目的能力。经过十载的努力的开拓，为不同行业的企事业单位提供了优质的服务，并获得了客户的一致好评。

安装EasyNetQ

从NuGet上安装即可，由于EasyNetQ是依赖RabbitMQ.Client所以，会同时安装两个dll。

PM> Install-Package EasyNetQ

连接RabbitMQ

　　使用EasyNetQ连接RabbitMQ，是在应用程序启动时创建一个IBus对象，并且，在应用程序关闭时释放该对象。RabbitMQ连接是基于IBus接口的，当IBus中的方法被调用，连接才会开启。创建一个IBus对象的方法如下：

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

根据上述代码可以看出，连接字符串中是基于Key/Value形式的，每个Key中间用分号（;）断开。其中host是必须要写的，其他的值都是可以不用写，会采用默认的配置。连接中可能用到的Key如下：

  • host，host=localhost 或者host =192.168.1.102或者host=my.rabbitmq.com,如果用到集群配置的话，那么可以用逗号将服务地址隔开，例如host=a.com,b.com,c.com

  • virtualHost,虚拟主机，默认为'/'

  • username,用户登录名

  • password，用户登录密码

  • requestedHeartbeat,心跳设置，默认是10秒

  • prefetchcount，默认是50

  • pubisherConfirms，默认为false

  • persistentMessages，消息持久化，默认为true

  • product,产品名

  • platform,平台

  • timeout，默认为10秒

关闭连接，可以使用 bus.Dispose();

EasyNetQ日志（Logging）

EasyNetQ提供了一个日志接口 IEasyNetQLogger

.NET操作RabbitMQ组件EasyNetQ使用中文简版文档。

public interface IEasyNetQLogger
{   void DebugWrite(string format, params object[] args);   void InfoWrite(string format, params object[] args);   void ErrorWrite(string format, params object[] args);   void ErrorWrite(Exception exception);
}

.NET操作RabbitMQ组件EasyNetQ使用中文简版文档。

内部默认用的是NullLogger，即什么也不做，不记录日志。在测试的时候也可以用ConsoleLogger来显示EasyNetQ运行中的各种信息。不过一般在正式使用环境中，可以自定义日志并实现IEasyNetQLogger接口。然后在RabbitHutch.CreateBus的重载方法中注册想用的日志类型。（日志中会记录连接RabbitMQ的过程和队列创建细节等信息，对于不懂RabbitMQ的同学，可能这些日志没有什么意义）。代码如下：

var logger = new MyLogger() // 继承自 IEasyNetQLoggervar bus = RabbitHutch.CreateBus(“my connection”, x => x.Register(_ => logger));

消息发布（Publish）

EasyNetQ支持最简单的消息模式是发布和订阅。发布消息后，任意消费者可以订阅该消息，也可以多个消费者订阅。并且不需要额外配置。首先，如上文中需要先创建一个IBus对象，然后，在创建一个可序列化的.NET对象。调用Publish方法即可。

 message =  MyMessage { Text =

警告，Publish只顾发送消息到队列，但是不管有没有消费端订阅，所以，发布之后，如果没有消费者，该消息将不会被消费甚至丢失。

消息订阅（Subscribe）

EasyNetQ提供了消息订阅，当调用Subscribe方法时候，EasyNetQ会创建一个用于接收消息的队列，不过与消息发布不同的是，消息订阅增加了一个参数，subscribe_id.代码如下：

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

第一个参数是订阅id，另外一个是delegate参数，用于处理接收到的消息。这里要注意的是，subscribe_id参数很重要，假如开发者用同一个subscribeid订阅了同一种消息类型两次或者多次，RabbitMQ会以轮训的方式给每个订阅的队列发送消息。接收到之后，其他队列就接收不到该消息。如果用不同的subscribeid订阅同一种消息类型，那么生成的每一个队列都会收到该消息。

举个例子：出库发货，我们有五个商品仓库，每个仓库的商品都是一样的，假如来了一堆订单，那么我们需要五个仓库共同工作，分别处理订单。而同样，总仓库需要知道总出货量，正常情况下，可以用每个仓库的出货量相加即可。不过如果我们在总仓库也监听商品订单消息，那么，每次来订单，总仓库也都会收到一份，那么可以作相应的统计了。

需要注意的是，在收到消息处理消息时候，不要占用太多的时间，会影响消息的处理效率，所以，遇到占用长时间的处理方法，最好用异步处理。代码如下：

.NET操作RabbitMQ组件EasyNetQ使用中文简版文档。

bus.SubscribeAsync("subscribe_async_test", message => 
    new WebClient().DownloadStringTask(new Uri("http://localhost:1338/?timeout=500"))
        .ContinueWith(task => 
            Console.WriteLine("Received: '{0}', Downloaded: '{1}'", 
                message.Text, 
                task.Result)));

.NET操作RabbitMQ组件EasyNetQ使用中文简版文档。

取消订阅，可以用如下方法：

var subscriptionResult = bus.Subscribe("sub_id", MyHandler);

...

subscriptionResult.Dispose();

或者直接IBus.Dispose();

消息发送（Send）和接收（Receive）

与Publish/Subscribe略有不同的是，Send/Receive 可以自己定义队列名称。

bus.Send("my.queue", new MyMessage{ Text = "Hello Widgets!" });
bus.Receive("my.queue", message => Console.WriteLine("MyMessage: {0}", message.Text));

并且，也可以在同一个队列上发送不同的消息类型，Receive方法可以这么写：

bus.Receive("my.queue", x => x
    .Add(message => deliveredMyMessage = message)
    .Add(message => deliveredMyOtherMessage = message));

如果消息到达队列，但是没有发现相应消息类型的处理时，EasyNetQ会发送一条消息到error队列，并且，带上一个异常信息：No handler found for message type 。与Subscribe类型，如果在同一个队列，同一个消息类型，多次调用Receive方法时，消息会通过轮询的形式发送给每个Receive端。

消息路由（Topic Based Routing）

Publish方法，可以加一个topic参数。

bus.Publish(message, "X.A");

消息订阅方可以通过路由来过滤相应的消息。

* 匹配一个字符

#匹配0个或者多个字符

所以 X.A.2 会匹配到 "#", "X.#", "*.A.*" 但不会匹配 "X.B.*" 或者 "A". 当消息订阅需要用到topic时候，需要调用Subscribe的重载方法

bus.Subscribe("my_id", handlerOfXDotStar, x => x.WithTopic("X.*"));
bus.Subscribe("my_id", handlerOfStarDotB, x => x.WithTopic("*.B"));

上述这种方式，会将消息轮询发送给两个订阅者，如果只需要一个订阅者的话，可以这么调用：

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

总结

以上就是EasyNetQ的一些基本用法了，是不是很简单呢，就这么轻松实现了消息队列的使用。当然，要深入内部还是有很多东西的。比如依赖注入，自定义EasyNetQ组件，RPC实现等。而且，他的源码也是比较有参考价值的，相对于之前自己写的基于RabbitMQ的封装，自己的简直是不能看呀。希望本文能给读完的你带来帮助。


当前名称：.NET操作RabbitMQ组件EasyNetQ使用中文简版文档。
链接分享：http://cdxtjz.cn/article/gpghcj.html

其他资讯