本篇內(nèi)容主要講解“怎么優(yōu)雅的使用RabbitMQ”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學(xué)習(xí)“怎么優(yōu)雅的使用RabbitMQ”吧!

在江漢等地區(qū)，都構(gòu)建了全面的區(qū)域性戰(zhàn)略布局，加強發(fā)展的系統(tǒng)性、市場前瞻性、產(chǎn)品創(chuàng)新能力，以專注、極致的服務(wù)理念，為客戶提供網(wǎng)站設(shè)計制作、做網(wǎng)站 網(wǎng)站設(shè)計制作按需網(wǎng)站設(shè)計,公司網(wǎng)站建設(shè),企業(yè)網(wǎng)站建設(shè),品牌網(wǎng)站制作,網(wǎng)絡(luò)營銷推廣,成都外貿(mào)網(wǎng)站建設(shè)公司,江漢網(wǎng)站建設(shè)費用合理。

RabbitMQ無疑是目前最流行的消息隊列之一，對各種語言環(huán)境的支持也很豐富，作為一個.NET developer有必要學(xué)習(xí)和了解這一工具。消息隊列的使用場景大概有3種：

1、系統(tǒng)集成，分布式系統(tǒng)的設(shè)計。各種子系統(tǒng)通過消息來對接，這種解決方案也逐步發(fā)展成一種架構(gòu)風(fēng)格，即“通過消息傳遞的架構(gòu)”。

2、當(dāng)系統(tǒng)中的同步處理方式嚴(yán)重影響了吞吐量，比如日志記錄。假如需要記錄系統(tǒng)中所有的用戶行為日志，如果通過同步的方式記錄日志勢必會影響系統(tǒng)的響應(yīng)速度，當(dāng)我們將日志消息發(fā)送到消息隊列，記錄日志的子系統(tǒng)就會通過異步的方式去消費日志消息。

3、系統(tǒng)的高可用性，比如電商的秒殺場景。當(dāng)某一時刻應(yīng)用服務(wù)器或數(shù)據(jù)庫服務(wù)器收到大量請求，將會出現(xiàn)系統(tǒng)宕機。如果能夠?qū)⒄埱筠D(zhuǎn)發(fā)到消息隊列，再由服務(wù)器去消費這些消息將會使得請求變得平穩(wěn)，提高系統(tǒng)的可用性。

如果想學(xué)習(xí)Java工程化、高性能及分布式、深入淺出。微服務(wù)、Spring，MyBatis，Netty源碼分析的朋友可以加我的Java高級交流：854630135，群里有阿里大牛直播講解技術(shù)，以及Java大型互聯(lián)網(wǎng)技術(shù)的視頻免費分享給大家。

一、開始使用RabbitMQ

RabbitMQ官網(wǎng)提供了詳細(xì)的安裝步驟，另外官網(wǎng)還提供了RabbitMQ在六種場景的使用教程。其中教程1、3、6將覆蓋99%的使用場景，所以正常來說只需要搞清楚這3個教程即可快速上手。

二、簡單分析

我們以官方提供的教程1做個簡單梳理：該教程展示了Producer如何向一個消息隊列(message queue)發(fā)送一個消息(message)，消息消費者(Consumer)收到該消息后消費該消息。

1、producer端：

var factory = new ConnectionFactory() { HostName = "localhost" };

 using (var connection = factory.CreateConnection())

 {

 while (Console.ReadLine() != null)

 {

 using (var channel = connection.CreateModel())

 {

 //創(chuàng)建一個名叫"hello"的消息隊列

 channel.QueueDeclare(queue: "hello",

 durable: false,

 exclusive: false,

 autoDelete: false,

 arguments: null);

 var message = "Hello World!";

 var body = Encoding.UTF8.GetBytes(message);

 //向該消息隊列發(fā)送消息message

 channel.BasicPublish(exchange: "",

 routingKey: "hello",

 basicProperties: null,

 body: body);

 Console.WriteLine(" [x] Sent {0}", message);

 }

 }

 }

該段代碼非常簡單，幾乎到了無法精簡的地步：創(chuàng)建了一個信道(channel)->創(chuàng)建一個隊列->向該隊列發(fā)送消息。

2、Consumer端

var factory = new ConnectionFactory() { HostName = "localhost" };

 using (var connection = factory.CreateConnection())

 {

 using (var channel = connection.CreateModel())

 {

 //創(chuàng)建一個名為"hello"的隊列，防止producer端沒有創(chuàng)建該隊列

 channel.QueueDeclare(queue: "hello",

 durable: false,

 exclusive: false,

 autoDelete: false,

 arguments: null);

 //回調(diào)，當(dāng)consumer收到消息后會執(zhí)行該函數(shù)

 var consumer = new EventingBasicConsumer(channel);

 consumer.Received += (model, ea) =>

 {

 var body = ea.Body;

 var message = Encoding.UTF8.GetString(body);

 Console.WriteLine(" [x] Received {0}", message);

 };

 //消費隊列"hello"中的消息

 channel.BasicConsume(queue: "hello",

 noAck: true,

 consumer: consumer);

 Console.WriteLine(" Press [enter] to exit.");

 Console.ReadLine();

 }

 }

該段代碼可以理解為：創(chuàng)建信道->創(chuàng)建隊列->定義回調(diào)函數(shù)->消費消息。

該實例描述了Send/Receive模式，可以簡單理解為1(producer) VS 1(consumer)的場景；

實例3則描述了Publish/Subscriber模式，即1(producer) VS 多個(consumer)；

在以上兩個示例中，producer只需要發(fā)送消息即可，并不關(guān)心consumer的返回結(jié)果。實例6則描述了一個RPC調(diào)用場景，producer發(fā)送消息后還要接收consumer的返回結(jié)果，這一場景看起來跟使用消息隊列的目的有點相悖。因為使用消息隊列的目的之一就是要異步，但是這一場景似乎又將異步變成了同步，不過這一場景也很有用，比如一個用戶操作產(chǎn)生了一個消息，應(yīng)用服務(wù)收到該消息后執(zhí)行了一些邏輯并使得數(shù)據(jù)庫發(fā)生了變化，UI會一直等待應(yīng)用服務(wù)的返回結(jié)果才刷新頁面。

如果想學(xué)習(xí)Java工程化、高性能及分布式、深入淺出。微服務(wù)、Spring，MyBatis，Netty源碼分析的朋友可以加我的Java高級交流：854630135，群里有阿里大牛直播講解技術(shù)，以及Java大型互聯(lián)網(wǎng)技術(shù)的視頻免費分享給大家。

三、 發(fā)現(xiàn)抽象

我桌子上放著一本RabbitMQ in Action，另外官網(wǎng)提供的文檔也很詳細(xì)，我感覺在一個月內(nèi)我就能精通RabbitMQ，到時候簡歷上又可以寫上“精通…”，感覺有點小得意呢... ，但是我知道這并不是使用RabbitMQ的最佳方式。

我們知道合理的抽象可以幫我們隱藏掉一些技術(shù)細(xì)節(jié)，讓我們將重心放在核心業(yè)務(wù)上，比如一個人問你：“大雁塔如何走？”你的回答可能是“小寨往東，一直走兩站，右手邊”，如果你回答：“右轉(zhuǎn)45度，向前走100米，再轉(zhuǎn)90度…”，對方就會迷失在這些細(xì)節(jié)中。

消息隊列的使用過程中實際隱藏著一種抽象——服務(wù)總線(Service Bus)。

我們在回頭看第一個例子，這個例子隱含的業(yè)務(wù)是：ClientA發(fā)送一個指令，ClientB收到該指令后做出反應(yīng)。如果是這樣，我們?yōu)槭裁匆P(guān)心如何創(chuàng)建channel，如何創(chuàng)建一個queue? 我僅僅是要發(fā)送一個消息而已。另外這個例子寫的其實不夠健壯：

沒有重試機制：如果ClientB第一次沒有執(zhí)行成功如何對該消息處理？

沒有錯誤處理機制：如果ClientB在重試了N次之后還是異常如何處理該消息？

沒有熔斷機制；

如何對ClientA做一個schedule(計劃安排），比如定時發(fā)送等；

沒有消息審計機制；

無法對消息的各個狀態(tài)做追蹤；

事物處理等。

服務(wù)總線正是這種場景的抽象，并且為我們提供了這些機制，讓我們趕快來看個究竟吧。

四、初識MassTransit

MassTransit是.NET平臺下的一款開源免費的ESB產(chǎn)品，官網(wǎng)：http://masstransit-project.com/，GitHub 700 star，500 Fork，類似的產(chǎn)品還有NServiceBus，之所以要選用MassTransit是因為他要比NServiceBus輕量級，另外在MassTransit開發(fā)之初就選用了RabbitMQ作為消息傳輸組建；同時我想拿他跟NServiceBus做個比較，看看他們到底有哪些側(cè)重點。

1、新建控制臺應(yīng)用程序：Masstransit.RabbitMQ.GreetingClient

使用MassTransit可以從Nuget中安裝：

Install-Package MassTransit.RabbitMQ

2、創(chuàng)建服務(wù)總線，發(fā)送一個命令

static void Main(string[] args)

{

 Console.WriteLine("Press 'Enter' to send a message.To exit, Ctrl + C");

 var bus = BusCreator.CreateBus();

 var sendToUri = new Uri($"{RabbitMqConstants.RabbitMqUri}{RabbitMqConstants.GreetingQueue}");

 while (Console.ReadLine()!=null)

 {

 Task.Run(() => SendCommand(bus, sendToUri)).Wait();

 }

 Console.ReadLine();

}

private static async void SendCommand(IBusControl bus,Uri sendToUri)

{

 var endPoint =await bus.GetSendEndpoint(sendToUri);

 var command = new GreetingCommand()

 {

 Id = Guid.NewGuid(),

 DateTime = DateTime.Now

 };

 await endPoint.Send(command);

 Console.WriteLine($"send command:id={command.Id},{command.DateTime}"); 

}

這一段代碼隱藏了眾多關(guān)于消息隊列的細(xì)節(jié)，將我們的注意力集中在發(fā)送消息上，同時ServiceBus提供的API也更接近業(yè)務(wù)，我們雖然發(fā)送的是一個消息，但是在這種場景下體現(xiàn)出來是一個命令，Send(command)這一API描述了我們的意圖。

如果想學(xué)習(xí)Java工程化、高性能及分布式、深入淺出。微服務(wù)、Spring，MyBatis，Netty源碼分析的朋友可以加我的Java高級交流：854630135，群里有阿里大牛直播講解技術(shù)，以及Java大型互聯(lián)網(wǎng)技術(shù)的視頻免費分享給大家。

3、服務(wù)端接收這一命令

新建一個命令臺控制程序：Masstransit.RabbitMQ.GreetingServer

var bus = BusCreator.CreateBus((cfg, host) =>

{

 cfg.ReceiveEndpoint(host, RabbitMqConstants.GreetingQueue, e =>

 {

 e.Consumer();

 });

});

這一代碼可以理解為服務(wù)端在監(jiān)聽消息，我們在服務(wù)端注冊了一個名為“GreetingConsumer”的消費者，GreetingConsumer的定義：

public class GreetingConsumer :IConsumer

{

 public async Task Consume(ConsumeContext context)

 {

 await Console.Out.WriteLineAsync($"receive greeting commmand: {context.Message.Id},{context.Message.DateTime}");

 }

}

該consumer可以消費類型為GreetingCommand的消息。這一實例幾乎隱藏了有關(guān)RabbitMQ的技術(shù)細(xì)節(jié)，將代碼中心放在了業(yè)務(wù)中，將這兩個控制臺應(yīng)用跑起來試試：

五、實現(xiàn)Publish/Subscribe模式

發(fā)布/訂閱模式使得基于消息傳遞的軟件架構(gòu)成為可能，這一能力表現(xiàn)為ClientA發(fā)送消息X，ClientB和ClientC都可以訂閱消息X。

1、我們在上面的例子中改造一下，當(dāng)GreetingConsumer收到GreetingCommand后發(fā)送一個GreetingEvent：

var greetingEvent = new GreetingEvent()

 {

 Id = context.Message.Id,

 DateTime = DateTime.Now

 };

 await context.Publish(greetingEvent);

2、新建控制臺程序Masstransit.RabbitMQ.GreetingEvent.SubscriberA用來訂閱GreetingEvent消息：

var bus = BusCreator.CreateBus((cfg, host) =>

 {

 cfg.ReceiveEndpoint(host, RabbitMqConstants.GreetingEventSubscriberAQueue, e =>

 {

 e.Consumer();

 });

 });

 bus.Start();

定義GreetingEventConsumer：

如果想學(xué)習(xí)Java工程化、高性能及分布式、深入淺出。微服務(wù)、Spring，MyBatis，Netty源碼分析的朋友可以加我的Java高級交流：854630135，群里有阿里大牛直播講解技術(shù)，以及Java大型互聯(lián)網(wǎng)技術(shù)的視頻免費分享給大家。

public class GreetingEventConsumer:IConsumer

 {

 public async Task Consume(ConsumeContext context)

 {

 await Console.Out.WriteLineAsync($"receive greeting event: id {context.Message.Id}");

 }

 }

這一代碼跟Masstransit.RabbitMQ.GreetingServer接受一個命令幾乎一模一樣，唯一的區(qū)別在于：

在Send/Receive模式中Client首先要獲得對方(Server)的終結(jié)點(endpoint)，直接向該終結(jié)點發(fā)送命令。Server方監(jiān)聽自己的終結(jié)點并消費命令。

而Publish/Subscribe模式中Client publish一個事件，SubscriberA在自己的終結(jié)點(endpointA)監(jiān)聽事件，SubscriberB在自己的終結(jié)點(endpointB)監(jiān)聽事件。

3、根據(jù)上面的分析再定義一個Masstransit.RabbitMQ.GreetingEvent.SubscriberB

4、將4個控制臺應(yīng)用程序跑起來看看

六、實現(xiàn)RPC模式

這一模式在Masstransit中被稱作Request/Response模式，通過IRequestClient 接口來實現(xiàn)相關(guān)操作。一個相關(guān)的例子在官方的github。

到此，相信大家對“怎么優(yōu)雅的使用RabbitMQ”有了更深的了解，不妨來實際操作一番吧！這里是創(chuàng)新互聯(lián)網(wǎng)站，更多相關(guān)內(nèi)容可以進入相關(guān)頻道進行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！