使用EasyNetQ组件操作RabbitMQ消息队列服务

发送请求获取响应的代码如下所示。

private void btnRequest_Click(object sender, EventArgs e)
{
//定义请求消息的对象
var request = new MyRequest()
{
Text = string.Format("请求消息，{0}", DateTime.Now)
};
//异步获取请求消息的结果并进行处理，展示应答消息在窗体中的
var task = bus.RequestAsync(request);
task.ContinueWith(respOnse=>
{
StringBuilder sb = new StringBuilder();
sb.AppendLine(response.Result.Text);
sb.AppendLine(this.txtContent.Text);
this.txtContent.Invoke(new MethodInvoker(delegate()
{
this.txtContent.Text = sb.ToString();
}));
});
}

private void btnRequest_Click(object sender, EventArgs e)
{
//定义请求消息的对象
var request = new MyRequest()
{
Text = string.Format(“请求消息，{0}”, DateTime.Now)
};
//异步获取请求消息的结果并进行处理，展示应答消息在窗体中的
var task = bus.RequestAsync(request);
task.ContinueWith(respOnse=>
{
StringBuilder sb = new StringBuilder();
sb.AppendLine(response.Result.Text);
sb.AppendLine(this.txtContent.Text);
this.txtContent.Invoke(new MethodInvoker(delegate()
{
this.txtContent.Text = sb.ToString();
}));
});
}

两个项目联合进行测试如下界面所示。

发布者多次发送消息的情况下，订阅者中，会进行消息的轮训处理，也就是进行均匀分配。

  5）消息发送（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));

//发送端代码
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));

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端。

   6）远程过程调用：

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

var request = new TestRequestMessage {Text = “Hello from the client! “};
bus.Request(request, respOnse=>
Console.WriteLine(“Got response: “{0}””, response.Text));

   7）RPC服务器：

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

bus.Respond(request =>
new TestResponseMessage{ Text = request.Text + ” all done!” });

   8）记录器：

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

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

   9）路由：

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

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

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”, 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"));

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

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

 RabbitMQ的应用场景，一般在快速处理订单，以及异步的多任务处理中可以得到很好的体现，下面是几个应用场景。

邮件和短消息的处理

订单的解耦处理

RabbitMQ的服务器架构

3、RabbitMQ查询状态出现错误的处理

安装成功之后使用rabbitmqctl status命令之后出现如下错误。

Status of node rabbit@WUHUACONG ...
Error: unable to perform an operation on node "rabbit@WUHUACONG". Please see diagnostics information and suggestions below.
Most common reasons for this are:
* Target node is unreachable (e.g. due to hostname resolution, TCP connection or firewall issues)
* CLI tool fails to authenticate with the server (e.g. due to CLI tool"s Erlang COOKIE not matching that of the server)
* Target node is not running
In addition to the diagnostics info below:
* See the CLI, clustering and networking guides on http://rabbitmq.com/documentation.html to learn more
* Consult server logs on node rabbit@WUHUACONG
DIAGNOSTICS
===========
attempted to contact: [rabbit@WUHUACONG]
rabbit@WUHUACONG:
* connected to epmd (port 4369) on WUHUACONG
* epmd reports node "rabbit" uses port 25672 for inter-node and CLI tool traffic
* TCP connection succeeded but Erlang distribution failed
* Authentication failed (rejected by the remote node), please check the Erlang COOKIE
Current node details:
* node name: rabbitmqcli100@WUHUACONG
* effective user"s home directory: C:UsersAdministrator
* Erlang COOKIE hash: RgaUM2coc+rxIhJrfLS7Jw==

Status of node rabbit@WUHUACONG …
Error: unable to perform an operation on node “rabbit@WUHUACONG”. Please see diagnostics information and suggestions below.
Most common reasons for this are:
* Target node is unreachable (e.g. due to hostname resolution, TCP connection or firewall issues)
* CLI tool fails to authenticate with the server (e.g. due to CLI tool”s Erlang COOKIE not matching that of the server)
* Target node is not running
In addition to the diagnostics info below:
* See the CLI, clustering and networking guides on http://rabbitmq.com/documentation.html to learn more
* Consult server logs on node rabbit@WUHUACONG
DIAGNOSTICS
===========
attempted to contact: [rabbit@WUHUACONG]
rabbit@WUHUACONG:
* connected to epmd (port 4369) on WUHUACONG
* epmd reports node “rabbit” uses port 25672 for inter-node and CLI tool traffic
* TCP connection succeeded but Erlang distribution failed
* Authentication failed (rejected by the remote node), please check the Erlang COOKIE
Current node details:
* node name: rabbitmqcli100@WUHUACONG
* effective user”s home directory: C:UsersAdministrator
* Erlang COOKIE hash: RgaUM2coc+rxIhJrfLS7Jw==

这个问题出现比较常见，主要原因是两个目录的.erlang.COOKIE文件内容不一样。

要确保.erlang.COOKIE文件的一致性，不知道什么原因导致了C:Users{UserName}.erlang.COOKIE和默认情况下C:WINDOWSSystem32configsystemprofile.erlang.COOKIE不一致了，将Windows目录下的拷贝到用户目录下就可以了。

反正无论如何，两个地址的COOKIE内容一致就可以了，然后重启下RabbitMQ服务器即可正常运行，并可以正常获取它的状态信息。