EasyNetQ使用(四)【Request与Response,Send与Receive】

时间:2019-11-18
本文章向大家介绍EasyNetQ使用(四)【Request与Response,Send与Receive】,主要包括EasyNetQ使用(四)【Request与Response,Send与Receive】使用实例、应用技巧、基本知识点总结和需要注意事项,具有一定的参考价值,需要的朋友可以参考一下。

EasyNetQ也支持Request/Response这种方式的消息模式。这种方式很容易在client/Server应用中执行,客户端发送一个请求到服务器,服务器然后处理请求后返回一个响应。和传统的RPC机制不同,EasyNetQ request/response操作不需要有一个名字,仅仅需要简单的定义一对request/response消息类型。

另外,不同于传统的RPC机制包括了众多的web service 工具集,EasyNetQrequest/response模式是基于消息传递,因此它是开箱即用型的异步模式。

发送请求,处理响应

EasyNet发送请求,调用IBus上的Request方法。

var myRequest = new MyRequest{Text ="hello Server"};
var response = bus.Request<MyRequest, MyResponse>(myRequest);
Console.WriteLine(response.Text);

这里我们创建了一个新的类型为MyRequest的请求,然后调用Request方法,并用这个消息作为它的参数。当这个response响应返回时,消息的Text属性值输出到控制台。

异步请求

消息天然是异步的。你发送一个消息,然后允许你的程序继续执行其他任务。在之后的某一个时间点上,你收到响应。用异步Request方法实现上面的方法,你的线程将阻塞直到响应返回。通常来说,使用RequestAsync方法返回一个Task是比较好的选择。

var task = bus.RequestAsync<TestRequestMessage,TestResponseMessage>(request);
task.ContinueWith(response =>{
    Console.WriteLine("Got Response:{0}",
    response.Result.Text);
    });
  • 1
  • 2
  • 3
  • 4
  • 5

响应请求

去写一个响应请求的服务,简单使用IBus.Respond方法。如下:

bus.Respond<MyRequest,MyResponse>(request =>
new MyResponse{ Text ="Responding to " + request.Text});

Respond带有单个参数,一个Func,接收一个请求并返回响应。同样的建议适用于订阅回调也适用于responder。不要被持续很长时间的IO操作所阻塞。假如你有持续很长时间的IO,应该使用ResponseAsync

异步响应

EasynetQ也提供了一个RespondAsync 方法,它带有一个Func>委托参数。它允许你执行持续很长时间IO密集型的操作,而不会阻塞EasyNetQ订阅处理循环。

static void Main(string[] args)
{
    //创建一组工作线程对象。
    var workers = new BlockingCollection<MyWorkder>();
    for(int i = 0; i < 10; i++)
    {
        workers.Add(new MyWorker());
    }
    //创建一个Bus
    var bus = RabbitHutch.CreateBus("host=localhost");
    //响应请求。
    bus.RespondAsync<RequestServerTime, ResponseServerTime>( 
        request =>
            Task.Factory.StartNew(() =>
            {
                var worker = worker.Take();
                try
                {
                    return worker.Execute(request);
                }
                finally
                {
                    workers.Add(worker);
                }
            }));
    Console.ReadLine();
    bus.Dispose();
}

示例应用

EasynetQ展示Request ResponseAutosubcriber,连上使用Windsor IOC的示例代码地址如下:
https://bitbucket.org/philipogorman/createrequestservice/src


鉴于Publish/SubsrcibeRequest/Response模式是位置透明的,在这个方面,你不需要去指定消息的消费者具体所处的位置,Send/Receive模式是特别针对通过命名队列来设计的。它也使得不用假定关于什么样的消息类型能发送到这个队列。这意味着,你能够发送不同类型的消息到相同的队列中。

Send/Receive模式对于创建“命令管道”,在那种你想为单个命令处理器建立一个缓冲通道情况下,是最理想的。

去发送一个消息,使用IBusSend方法,需要指定你想发送消息到指定队列的名称和消息本身。

bus.Send("my.queue", new MyMessage{ Text = "hello widget!"});
  • 1

为特定消息类型创建一个消息接收器,使用IBus上的Receive方法:

bus.Receive<MyMessage>("my.queue", 
    message => Console.WriteLine("Message:{0}",message.Text));

你能够在相同队列上为不同的消息类型创建多个接收者,通过使用Receive 带有一个Action的重载方法,代码如下:

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

如果消息送达了一个接收队列,但是没有匹配到任何接收者,EasyNetQ将会把消息带着一个异常“No handler for message type写进EasyNetQ的错误队列”。

注意:你可能不想对相同队列多次调用bus.Receive方法。这就将在队列和RatbbitMQ上创建一个新的消费者去轮询这些消息。假如你正在消费不同类型的消息,用不同的Receive调用(因此也是不同的消费者),你的有些消息将会最终发送到错误队列,因为EasyNetQ不能找到处理器去处理与你消息类型有关系的消费者,消息将会被消耗掉

原文地址:https://www.cnblogs.com/lhxsoft/p/11881519.html