搬磚的陳大師版權(quán)所有，轉(zhuǎn)載請注明：http://www.lenggirl.com/tool/RabbitMQ.html

手冊:http://www.rabbitmq.com/getstarted.html

安裝:http://www.rabbitmq.com/download.html

參考：http://blog.csdn.net/whycold/article/details/41119807

一.介紹

AMQP，即Advanced Message Queuing Protocol，高級消息隊列協(xié)議，是應(yīng)用層協(xié)議的一個開放標(biāo)準(zhǔn)，為面向消息的中間件設(shè)計。消息中間件主要用于組件之間的解耦，消息的發(fā)送者無需知道消息使用者的存在，反之亦然。

AMQP的主要特征是面向消息、隊列、路由（包括點(diǎn)對點(diǎn)和發(fā)布/訂閱）、可靠性、安全。

RabbitMQ是一個開源的AMQP實(shí)現(xiàn)，服務(wù)器端用Erlang語言編寫，支持多種客戶端，如：Go、Python、Ruby。用于在分布式系統(tǒng)中存儲轉(zhuǎn)發(fā)消息。

二.安裝

ubuntu直接下載deb文件安裝，默認(rèn)已經(jīng)啟動，sudo敲入：

sudo  rabbitmq-server start 
sudo lsof -i:5672

啟用插件，進(jìn)入UI：

sudo rabbitmq-plugins enable rabbitmq_management

登錄http://127.0.0.1:15672

用戶名:密碼=guest:guest

三.使用

# 敲入查看幫助sudo rabbitmqctl# 創(chuàng)建用戶
sudo rabbitmqctl add_user  登錄用戶名  密碼# 可以創(chuàng)建管理員用戶，負(fù)責(zé)整個MQ的運(yùn)維
sudo rabbitmqctl set_user_tags 登錄用戶名 administrator# 可以創(chuàng)建RabbitMQ監(jiān)控用戶，負(fù)責(zé)整個MQ的監(jiān)控
sudo rabbitmqctl set_user_tags 登錄用戶名 monitoring 
# 可以創(chuàng)建某個項目的專用用戶，只能訪問項目自己的virtual hosts
sudo rabbitmqctl set_user_tags 登錄用戶名 management# 查看用戶
sudo rabbitmqctl list_users 

# 授權(quán)# 該命令使用戶具有/這個virtual host中所有資源的配置、寫、讀權(quán)限以便管理其中的資源# set_permissions [-p <vhostpath>] <user> <conf> <write> <read># 其中，<conf> <write> <read>的位置分別用正則表達(dá)式來匹配特定的資源，如'^(amq\.gen.*|amq\.default)$'可以匹配server生成的和默認(rèn)的exchange，'^$'不匹配任何資源
sudo rabbitmqctl  set_permissions -p / 登錄用戶名 '.*' '.*' '.*'

四.概念入門

1.ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel都是RabbitMQ對外提供的API中最基本的對象。

Connection是RabbitMQ的socket鏈接，它封裝了socket協(xié)議相關(guān)部分邏輯。

ConnectionFactory為Connection的制造工廠。

Channel是我們與RabbitMQ打交道的最重要的一個接口，我們大部分的業(yè)務(wù)操作是在Channel這個接口中完成的，包括定義Queue、定義Exchange、綁定Queue與Exchange、發(fā)布消息等。

2.Queue

Queue（隊列）是RabbitMQ的內(nèi)部對象，用于存儲消息，用下圖表示。

RabbitMQ中的消息都只能存儲在Queue中，生產(chǎn)者（下圖中的P）生產(chǎn)消息并最終投遞到Queue中，消費(fèi)者（下圖中的C）可以從Queue中獲取消息并消費(fèi)。

多個消費(fèi)者可以訂閱同一個Queue，這時Queue中的消息會被平均分?jǐn)偨o多個消費(fèi)者進(jìn)行處理，而不是每個消費(fèi)者都收到所有的消息并處理。

3.消息的一些機(jī)制

3.1.消息確認(rèn)Message acknowledgment

在實(shí)際應(yīng)用中，可能會發(fā)生消費(fèi)者收到Queue中的消息，但沒有處理完成就宕機(jī)（或出現(xiàn)其他意外）的情況，這種情況下就可能會導(dǎo)致消息丟失。為了避免這種情況發(fā)生，我們可以要求消費(fèi)者在消費(fèi)完消息后發(fā)送一個回執(zhí)給RabbitMQ，RabbitMQ收到消息回執(zhí)（Message acknowledgment）后才將該消息從Queue中移除；如果RabbitMQ沒有收到回執(zhí)并檢測到消費(fèi)者的RabbitMQ連接斷開，則RabbitMQ會將該消息發(fā)送給其他消費(fèi)者（如果存在多個消費(fèi)者）進(jìn)行處理。這里不存在timeout概念，一個消費(fèi)者處理消息時間再長也不會導(dǎo)致該消息被發(fā)送給其他消費(fèi)者，除非它的RabbitMQ連接斷開。

這里會產(chǎn)生另外一個問題，如果我們的開發(fā)人員在處理完業(yè)務(wù)邏輯后，忘記發(fā)送回執(zhí)給RabbitMQ，這將會導(dǎo)致嚴(yán)重的bug——Queue中堆積的消息會越來越多；消費(fèi)者重啟后會重復(fù)消費(fèi)這些消息并重復(fù)執(zhí)行業(yè)務(wù)邏輯…

另外pub message是沒有ack的。（？？）

3.2.消息持久Message durability

如果我們希望即使在RabbitMQ服務(wù)重啟的情況下，也不會丟失消息，我們可以將Queue與Message都設(shè)置為可持久化的（durable），這樣可以保證絕大部分情況下我們的RabbitMQ消息不會丟失。但依然解決不了小概率丟失事件的發(fā)生（比如RabbitMQ服務(wù)器已經(jīng)接收到生產(chǎn)者的消息，但還沒來得及持久化該消息時RabbitMQ服務(wù)器就斷電了），如果我們需要對這種小概率事件也要管理起來，那么我們要用到事務(wù)。由于這里僅為RabbitMQ的簡單介紹，所以這里將不講解RabbitMQ相關(guān)的事務(wù)。

3.3.提前取機(jī)制Prefetch count

前面我們講到如果有多個消費(fèi)者同時訂閱同一個Queue中的消息，Queue中的消息會被平攤給多個消費(fèi)者。這時如果每個消息的處理時間不同，就有可能會導(dǎo)致某些消費(fèi)者一直在忙，而另外一些消費(fèi)者很快就處理完手頭工作并一直空閑的情況。我們可以通過設(shè)置prefetchCount來限制Queue每次發(fā)送給每個消費(fèi)者的消息數(shù)，比如我們設(shè)置prefetchCount=1，則Queue每次給每個消費(fèi)者發(fā)送一條消息；消費(fèi)者處理完這條消息后Queue會再給該消費(fèi)者發(fā)送一條消息。就是變慢而已。訂閱模式如何平攤？這種模式是一個消費(fèi)者一次性拿很多條消息？

4.Exchange

在上一節(jié)我們看到生產(chǎn)者將消息投遞到Queue中，實(shí)際上這在RabbitMQ中這種事情永遠(yuǎn)都不會發(fā)生。實(shí)際的情況是，生產(chǎn)者將消息發(fā)送到Exchange（交換器，下圖中的X），由Exchange將消息路由到一個或多個Queue中（或者丟棄）。

Exchange是按照什么邏輯將消息路由到Queue的？這個將在Binding一節(jié)介紹。

RabbitMQ中的Exchange有四種類型，不同的類型有著不同的路由策略，這將在Exchange Types一節(jié)介紹。

5.Routing key

生產(chǎn)者在將消息發(fā)送給Exchange的時候，一般會指定一個routing key，來指定這個消息的路由規(guī)則，而這個routing key需要與Exchange Type及binding key聯(lián)合使用才能最終生效。

在Exchange Type與binding key固定的情況下（在正常使用時一般這些內(nèi)容都是固定配置好的），我們的生產(chǎn)者就可以在發(fā)送消息給Exchange時，通過指定routing key來決定消息流向哪里。

RabbitMQ為routing key設(shè)定的長度限制為255 bytes。

6.Binding

RabbitMQ中通過Binding將Exchange與Queue關(guān)聯(lián)起來，這樣RabbitMQ就知道如何正確地將消息路由到指定的Queue了。

6.1.Binding key

在綁定（Binding）Exchange與Queue的同時，一般會指定一個binding key；消費(fèi)者將消息發(fā)送給Exchange時，一般會指定一個routing key；當(dāng)binding key與routing key相匹配時，消息將會被路由到對應(yīng)的Queue中。這個將在Exchange Types章節(jié)會列舉實(shí)際的例子加以說明。

在綁定多個Queue到同一個Exchange的時候，這些Binding允許使用相同的binding key。 binding key 并不是在所有情況下都生效，它依賴于Exchange Type，比如fanout類型(廣播)的Exchange就會無視binding key，而是將消息路由到所有綁定到該Exchange的Queue。

7.Exchange Types

RabbitMQ常用的Exchange Type有fanout、direct、topic、headers這四種（AMQP規(guī)范里還提到兩種Exchange Type，分別為system與自定義，這里不予以描述），下面分別進(jìn)行介紹。

7.1.fanout

fanout類型的Exchange路由規(guī)則非常簡單，它會把所有發(fā)送到該Exchange的消息路由到所有與它綁定的Queue中。

7.2.direct

direct類型的Exchange路由規(guī)則也很簡單，它會把消息路由到那些binding key與routing key完全匹配的Queue中。

以上圖的配置為例，我們以routingKey=”error”發(fā)送消息到Exchange，則消息會路由到Queue1（amqp.gen-S9b…，這是由RabbitMQ自動生成的Queue名稱）和Queue2（amqp.gen-Agl…）；如果我們以routingKey=”info”或routingKey=”warning”來發(fā)送消息，則消息只會路由到Queue2。如果我們以其他routingKey發(fā)送消息，則消息不會路由到這兩個Queue中。

7.3.topic

前面講到direct類型的Exchange路由規(guī)則是完全匹配binding key與routing key，但這種嚴(yán)格的匹配方式在很多情況下不能滿足實(shí)際業(yè)務(wù)需求。topic類型的Exchange在匹配規(guī)則上進(jìn)行了擴(kuò)展，它與direct類型的Exchage相似，也是將消息路由到binding key與routing key相匹配的Queue中，但這里的匹配規(guī)則有些不同，它約定：

1. routing key為一個句點(diǎn)號“. ”分隔的字符串（我們將被句點(diǎn)號“. ”分隔開的每一段獨(dú)立的字符串稱為一個單詞），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”
2. binding key與routing key一樣也是句點(diǎn)號“. ”分隔的字符串
3. binding key中可以存在兩種特殊字符“*”與“#”，用于做模糊匹配，其中“*”用于匹配一個單詞，“#”用于匹配多個單詞（可以是零個）

以上圖中的配置為例，routingKey=”quick.orange.rabbit”的消息會同時路由到Q1與Q2，routingKey=”lazy.orange.fox”的消息會路由到Q1與Q2，routingKey=”lazy.brown.fox”的消息會路由到Q2，routingKey=”lazy.pink.rabbit”的消息會路由到Q2（只會投遞給Q2一次，雖然這個routingKey與Q2的兩個bindingKey都匹配）；routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息將會被丟棄，因?yàn)樗鼈儧]有匹配任何bindingKey

7.4.headers

headers類型的Exchange不依賴于routing key與binding key的匹配規(guī)則來路由消息，而是根據(jù)發(fā)送的消息內(nèi)容中的headers屬性進(jìn)行匹配。

在綁定Queue與Exchange時指定一組鍵值對；當(dāng)消息發(fā)送到Exchange時，RabbitMQ會取到該消息的headers（也是一個鍵值對的形式），對比其中的鍵值對是否完全匹配Queue與Exchange綁定時指定的鍵值對；如果完全匹配則消息會路由到該Queue，否則不會路由到該Queue。

該類型的Exchange沒有用到過（不過也應(yīng)該很有用武之地），所以不做介紹。

8.RPC

MQ本身是基于異步的消息處理，前面的示例中所有的生產(chǎn)者（P）將消息發(fā)送到RabbitMQ后不會知道消費(fèi)者（C）處理成功或者失?。ㄉ踔吝B有沒有消費(fèi)者來處理這條消息都不知道）。

但實(shí)際的應(yīng)用場景中，我們很可能需要一些同步處理，需要同步等待服務(wù)端將我的消息處理完成后再進(jìn)行下一步處理。這相當(dāng)于RPC（Remote Procedure Call，遠(yuǎn)程過程調(diào)用）。在RabbitMQ中也支持RPC。

RabbitMQ中實(shí)現(xiàn)RPC的機(jī)制是：

  1. 客戶端發(fā)送請求（消息）時，在消息的屬性（MessageProperties，在AMQP協(xié)議中定義了14中properties，這些屬性會隨著消息一起發(fā)送）中設(shè)置兩個值replyTo（一個Queue名稱，用于告訴服務(wù)器處理完成后將通知我的消息發(fā)送到這個Queue中）和correlationId（此次請求的標(biāo)識號，服務(wù)器處理完成后需要將此屬性返還，客戶端將根據(jù)這個id了解哪條請求被成功執(zhí)行了或執(zhí)行失?。?/p>

  2. 服務(wù)器端收到消息并處理

  3. 服務(wù)器端處理完消息后，將生成一條應(yīng)答消息到replyTo指定的Queue，同時帶上correlationId屬性

  4. 客戶端之前已訂閱replyTo指定的Queue，從中收到服務(wù)器的應(yīng)答消息后，根據(jù)其中的correlationId屬性分析哪條請求被執(zhí)行了，根據(jù)執(zhí)行結(jié)果進(jìn)行后續(xù)業(yè)務(wù)處理

五.Go 接口

http://www.rabbitmq.com/tutorials/tutorial-one-go.html

請看官方示例：

https://github.com/rabbitmq/rabbitmq-tutorials/blob/master/go/send.go

上面的例子仔細(xì)看，有必要看源碼！

GO接口庫

六.實(shí)例解釋

四種模式

  1. DIRECT 默認(rèn)點(diǎn)對點(diǎn)模式

  2. TOPIC 話題模式

  3. FANOUT 廣播模式

  4. RPC RPC模式

工作隊列：默認(rèn)點(diǎn)對點(diǎn)模式

發(fā)布方，一個！

package mainimport (
        "fmt"        "log"        "os"        "strings"        "github.com/streadway/amqp"
)

func failOnError(err error, msg string) {        if err != nil {
                log.Fatalf("%s: %s", msg, err)
                panic(fmt.Sprintf("%s: %s", msg, err))
        }
}func main() {        // 撥號，下面例子都一樣
        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
        failOnError(err, "Failed to connect to RabbitMQ")
        defer conn.Close()        // 這個是最重要的
        ch, err := conn.Channel()
        failOnError(err, "Failed to open a channel")
        defer ch.Close()        // 申明一個隊列        // https://godoc.org/github.com/streadway/amqp#Channel.QueueDeclare
        q, err := ch.QueueDeclare(                "task_queue", // name  有名字！                true,         // durable  持久性的,如果事前已經(jīng)聲明了該隊列，不能重復(fù)聲明                false,        // delete when unused                false,        // exclusive 如果是真，連接一斷開，隊列刪除                false,        // no-wait
                nil,          // arguments
        )
        failOnError(err, "Failed to declare a queue")

        body := bodyFrom(os.Args)        
        // 發(fā)布
        err = ch.Publish(                "",           // exchange 默認(rèn)模式，exchange為空
                q.Name,       // routing key 默認(rèn)模式路由到同名隊列，即是task_queue                false,        // mandatory                false,
                amqp.Publishing{                        // 持久性的發(fā)布，因?yàn)殛犃斜宦暶鳛槌志玫?，發(fā)布消息必須加上這個（可能不用），但消息還是可能會丟，如消息到緩存但MQ掛了來不及持久化。
                        DeliveryMode: amqp.Persistent,
                        ContentType:  "text/plain",
                        Body:         []byte(body),
                })
        failOnError(err, "Failed to publish a message")
        log.Printf(" [x] Sent %s", body)
}func bodyFrom(args []string) string {        var s string        if (len(args) < 2) || os.Args[1] == "" {
                s = "hello"
        } else {
                s = strings.Join(args[1:], " ")
        }        return s
}

工作方，多個，拿發(fā)布方的消息

package mainimport (        "bytes"        "fmt"        "github.com/streadway/amqp"        "log"        "time"
)

func failOnError(err error, msg string) {        if err != nil {                log.Fatalf("%s: %s", msg, err)
                panic(fmt.Sprintf("%s: %s", msg, err))
        }
}

func main() {
        conn, err

搬磚的陳大師版權(quán)所有，轉(zhuǎn)載請注明：http://www.lenggirl.com/tool/RabbitMQ.html

更多精彩請轉(zhuǎn)到 http://www.lenggirl.com 文章發(fā)表在博客園只是為了有更多的人看到，文章格式在博客園不太好看，請到我的個人博客

http://www.cnblogs.com/nima/p/7126225.html