RabbitMQ實例詳解

RabbitMQ實例詳解

　　消息隊列中間件是分布式系統中重要的組件，主要解決應用解耦，異步消息，流量削鋒等問題，實現高性能，高可用，可伸縮和最終一致性架構。

Queue

Queue（隊列）是RabbitMQ的內部對象，用於存儲消息，用下圖表示。

RabbitMQ中的消息都只能存儲在Queue中，生產者（下圖中的P）生產消息並最終投遞到Queue中，消費者（下圖中的C）可以從Queue中獲取消息並消費。

多個消費者可以訂閱同一個Queue，這時Queue中的消息會被平均分攤給多個消費者進行處理，而不是每個消費者都收到所有的消息並處理。

Message acknowledgment

在實際應用中，可能會發生消費者收到Queue中的消息，但沒有處理完成就宕機（或出現其他意外）的情況，這種情況下就可能會導致消息丟失。為了避免這種情況發生，我們可以要求消費者在消費完消息后發送一個回執給RabbitMQ，RabbitMQ收到消息回執（Message acknowledgment）后才將該消息從Queue中移除；如果RabbitMQ沒有收到回執並檢測到消費者的RabbitMQ連接斷開，則RabbitMQ會將該消息發送給其他消費者（如果存在多個消費者）進行處理。這里不存在timeout概念，一個消費者處理消息時間再長也不會導致該消息被發送給其他消費者，除非它的RabbitMQ連接斷開。
這里會產生另外一個問題，如果我們的開發人員在處理完業務邏輯后，忘記發送回執給RabbitMQ，這將會導致嚴重的bug——Queue中堆積的消息會越來越多；消費者重啟后會重復消費這些消息並重復執行業務邏輯…

另外pub message是沒有ack的。

Message durability

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

Prefetch count

前面我們講到如果有多個消費者同時訂閱同一個Queue中的消息，Queue中的消息會被平攤給多個消費者。這時如果每個消息的處理時間不同，就有可能會導致某些消費者一直在忙，而另外一些消費者很快就處理完手頭工作並一直空閑的情況。我們可以通過設置prefetchCount來限制Queue每次發送給每個消費者的消息數，比如我們設置prefetchCount=1，則Queue每次給每個消費者發送一條消息；消費者處理完這條消息后Queue會再給該消費者發送一條消息。

Exchange

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

Exchange是按照什么邏輯將消息路由到Queue的？這個將在Binding一節介紹。
RabbitMQ中的Exchange有四種類型，不同的類型有着不同的路由策略，這將在Exchange Types一節介紹。

routing key

生產者在將消息發送給Exchange的時候，一般會指定一個routing key，來指定這個消息的路由規則，而這個routing key需要與Exchange Type及binding key聯合使用才能最終生效。
在Exchange Type與binding key固定的情況下（在正常使用時一般這些內容都是固定配置好的），我們的生產者就可以在發送消息給Exchange時，通過指定routing key來決定消息流向哪里。
RabbitMQ為routing key設定的長度限制為255 bytes。

Binding

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

Binding key

在綁定（Binding）Exchange與Queue的同時，一般會指定一個binding key；消費者將消息發送給Exchange時，一般會指定一個routing key；當binding key與routing key相匹配時，消息將會被路由到對應的Queue中。這個將在Exchange Types章節會列舉實際的例子加以說明。
在綁定多個Queue到同一個Exchange的時候，這些Binding允許使用相同的binding key。
binding key 並不是在所有情況下都生效，它依賴於Exchange Type，比如fanout類型的Exchange就會無視binding key，而是將消息路由到所有綁定到該Exchange的Queue。

Exchange Types

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

fanout

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

上圖中，生產者（P）發送到Exchange（X）的所有消息都會路由到圖中的兩個Queue，並最終被兩個消費者（C1與C2）消費。

direct

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

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

topic

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

• routing key為一個句點號“. ”分隔的字符串（我們將被句點號“. ”分隔開的每一段獨立的字符串稱為一個單詞），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”
• binding key與routing key一樣也是句點號“. ”分隔的字符串
• 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”的消息將會被丟棄，因為它們沒有匹配任何bindingKey。

headers

headers類型的Exchange不依賴於routing key與binding key的匹配規則來路由消息，而是根據發送的消息內容中的headers屬性進行匹配。
在綁定Queue與Exchange時指定一組鍵值對；當消息發送到Exchange時，RabbitMQ會取到該消息的headers（也是一個鍵值對的形式），對比其中的鍵值對是否完全匹配Queue與Exchange綁定時指定的鍵值對；如果完全匹配則消息會路由到該Queue，否則不會路由到該Queue。
該類型的Exchange沒有用到過（不過也應該很有用武之地），所以不做介紹。

RPC

MQ本身是基於異步的消息處理，前面的示例中所有的生產者（P）將消息發送到RabbitMQ后不會知道消費者（C）處理成功或者失敗（甚至連有沒有消費者來處理這條消息都不知道）。
但實際的應用場景中，我們很可能需要一些同步處理，需要同步等待服務端將我的消息處理完成后再進行下一步處理。這相當於RPC（Remote Procedure Call，遠程過程調用）。在RabbitMQ中也支持RPC。

RabbitMQ中實現RPC的機制是：

• 客戶端發送請求（消息）時，在消息的屬性（MessageProperties，在AMQP協議中定義了14中properties，這些屬性會隨着消息一起發送）中設置兩個值replyTo（一個Queue名稱，用於告訴服務器處理完成后將通知我的消息發送到這個Queue中）和correlationId（此次請求的標識號，服務器處理完成后需要將此屬性返還，客戶端將根據這個id了解哪條請求被成功執行了或執行失敗）
• 服務器端收到消息並處理
• 服務器端處理完消息后，將生成一條應答消息到replyTo指定的Queue，同時帶上correlationId屬性
• 客戶端之前已訂閱replyTo指定的Queue，從中收到服務器的應答消息后，根據其中的correlationId屬性分析哪條請求被執行了，根據執行結果進行后續業務處理。

RabbitMQ的簡單實例

 

package test;

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class SendTest {
    private final static String QUEUE_NAME = "hello";
    public static void main(String[] arg) throws Exception{
        //創建連接工程
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        //創建連接
        Connection connection = factory.newConnection();
        //創建channel
        Channel channel = connection.createChannel();
        //創建隊列，將消息發送到隊列，便於消費
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        String message = "hello World!!!!!!!!!!!!!!hello World!!!!!!!!";
        /* RabbitMQ默認有一個exchange，叫default exchange，它用一個空字符串表示，
         * 它是direct exchange類型，任何發往這個exchange的消息都會被路由到routing key
         * 的名字對應的隊列上，如果沒有對應的隊列，則消息會被丟棄。這就是為什么代碼中channel執行
         * basicPulish方法時，第二個參數本應該為routing key，卻被寫上了QUEUE_NAME。*/
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        channel.close();
        connection.close();
    }
}


package test;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

public class ReceiveTest {
    private final static String QUEUE_NAME = "hello";
    public static void main(String[] arg) throws Exception{
        //創建連接工程
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        //創建連接
        Connection connection = factory.newConnection();
        //創建channel
        Channel channel = connection.createChannel();
        //創建隊列，將消息發送到隊列，便於消費
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(QUEUE_NAME, true, consumer);
        while(true){
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(message);
        }
    }
}

Channel的基本函數詳解

1.1 channel.exchangeDeclare()：

type：有direct、fanout、topic三種
durable：true、false true：服務器重啟會保留下來Exchange。警告：僅設置此選項，不代表消息持久化。即不保證重啟后消息還在。原文：true if we are declaring a durable exchange (the exchange will survive a server restart)
autoDelete:true、false.true:當已經沒有消費者時，服務器是否可以刪除該Exchange。原文1：true if the server should delete the exchange when it is no longer in use。

 /**
     * Declare an exchange.
     * @see com.rabbitmq.client.AMQP.Exchange.Declare
     * @see com.rabbitmq.client.AMQP.Exchange.DeclareOk
     * @param exchange the name of the exchange
     * @param type the exchange type
     * @param durable true if we are declaring a durable exchange (the exchange will survive a server restart)
     * @param autoDelete true if the server should delete the exchange when it is no longer in use
     * @param arguments other properties (construction arguments) for the exchange
     * @return a declaration-confirm method to indicate the exchange was successfully declared
     * @throws java.io.IOException if an error is encountered
     */
    Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete,
                                       Map<String, Object> arguments) throws IOException;

1.2 chanel.basicQos()

prefetchSize：0 
prefetchCount：會告訴RabbitMQ不要同時給一個消費者推送多於N個消息，即一旦有N個消息還沒有ack，則該consumer將block掉，直到有消息ack
global：true\false 是否將上面設置應用於channel，簡單點說，就是上面限制是channel級別的還是consumer級別

備注：據說prefetchSize 和global這兩項，rabbitmq沒有實現，暫且不研究

    /**
     * Request specific "quality of service" settings.
     *
     * These settings impose limits on the amount of data the server
     * will deliver to consumers before requiring acknowledgements.
     * Thus they provide a means of consumer-initiated flow control.
     * @see com.rabbitmq.client.AMQP.Basic.Qos
     * @param prefetchSize maximum amount of content (measured in
     * octets) that the server will deliver, 0 if unlimited
     * @param prefetchCount maximum number of messages that the server
     * will deliver, 0 if unlimited
     * @param global true if the settings should be applied to the
     * entire channel rather than each consumer
     * @throws java.io.IOException if an error is encountered
     */
    void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;

1.3 channel.basicPublish()

routingKey：路由鍵，#匹配0個或多個單詞，*匹配一個單詞，在topic exchange做消息轉發用

mandatory：true：如果exchange根據自身類型和消息routeKey無法找到一個符合條件的queue，那么會調用basic.return方法將消息返還給生產者。false：出現上述情形broker會直接將消息扔掉

immediate：true：如果exchange在將消息route到queue(s)時發現對應的queue上沒有消費者，那么這條消息不會放入隊列中。當與消息routeKey關聯的所有queue(一個或多個)都沒有消費者時，該消息會通過basic.return方法返還給生產者。
BasicProperties ：需要注意的是BasicProperties.deliveryMode，0:不持久化 1：持久化 這里指的是消息的持久化，配合channel(durable=true),queue(durable)可以實現，即使服務器宕機，消息仍然保留
簡單來說：mandatory標志告訴服務器至少將該消息route到一個隊列中，否則將消息返還給生產者；immediate標志告訴服務器如果該消息關聯的queue上有消費者，則馬上將消息投遞給它，如果所有queue都沒有消費者，直接把消息返還給生產者，不用將消息入隊列等待消費者了。

  /**
     * Publish a message.
     *
     * Publishing to a non-existent exchange will result in a channel-level
     * protocol exception, which closes the channel.
     *
     * Invocations of <code>Channel#basicPublish</code> will eventually block if a
     * <a href="http://www.rabbitmq.com/alarms.html">resource-driven alarm</a> is in effect.
     *
     * @see com.rabbitmq.client.AMQP.Basic.Publish
     * @see <a href="http://www.rabbitmq.com/alarms.html">Resource-driven alarms</a>.
     * @param exchange the exchange to publish the message to
     * @param routingKey the routing key
     * @param mandatory true if the 'mandatory' flag is to be set
     * @param immediate true if the 'immediate' flag is to be
     * set. Note that the RabbitMQ server does not support this flag.
     * @param props other properties for the message - routing headers etc
     * @param body the message body
     * @throws java.io.IOException if an error is encountered
     */
    void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body)
            throws IOException;

1.4 channel.basicAck();

deliveryTag:該消息的index
multiple：是否批量.true:將一次性ack所有小於deliveryTag的消息。

 

    /**
     * Acknowledge one or several received
     * messages. Supply the deliveryTag from the {@link com.rabbitmq.client.AMQP.Basic.GetOk}
     * or {@link com.rabbitmq.client.AMQP.Basic.Deliver} method
     * containing the received message being acknowledged.
     * @see com.rabbitmq.client.AMQP.Basic.Ack
     * @param deliveryTag the tag from the received {@link com.rabbitmq.client.AMQP.Basic.GetOk} or {@link com.rabbitmq.client.AMQP.Basic.Deliver}
     * @param multiple true to acknowledge all messages up to and
     * including the supplied delivery tag; false to acknowledge just
     * the supplied delivery tag.
     * @throws java.io.IOException if an error is encountered
     */
    void basicAck(long deliveryTag, boolean multiple) throws IOException;

 

1.5 channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);

deliveryTag:該消息的index
multiple：是否批量.true:將一次性拒絕所有小於deliveryTag的消息。
requeue：被拒絕的是否重新入隊列

 

    /**
     * Reject one or several received messages.
     *
     * Supply the <code>deliveryTag</code> from the {@link com.rabbitmq.client.AMQP.Basic.GetOk}
     * or {@link com.rabbitmq.client.AMQP.Basic.GetOk} method containing the message to be rejected.
     * @see com.rabbitmq.client.AMQP.Basic.Nack
     * @param deliveryTag the tag from the received {@link com.rabbitmq.client.AMQP.Basic.GetOk} or {@link com.rabbitmq.client.AMQP.Basic.Deliver}
     * @param multiple true to reject all messages up to and including
     * the supplied delivery tag; false to reject just the supplied
     * delivery tag.
     * @param requeue true if the rejected message(s) should be requeued rather
     * than discarded/dead-lettered
     * @throws java.io.IOException if an error is encountered
     */
    void basicNack(long deliveryTag, boolean multiple, boolean requeue)
            throws IOException;

1.5 channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);

deliveryTag:該消息的index
requeue：被拒絕的是否重新入隊列

channel.basicNack 與 channel.basicReject 的區別在於basicNack可以拒絕多條消息，而basicReject一次只能拒絕一條消息

 /**
     * Reject a message. Supply the deliveryTag from the {@link com.rabbitmq.client.AMQP.Basic.GetOk}
     * or {@link com.rabbitmq.client.AMQP.Basic.Deliver} method
     * containing the received message being rejected.
     * @see com.rabbitmq.client.AMQP.Basic.Reject
     * @param deliveryTag the tag from the received {@link com.rabbitmq.client.AMQP.Basic.GetOk} or {@link com.rabbitmq.client.AMQP.Basic.Deliver}
     * @param requeue true if the rejected message should be requeued rather than discarded/dead-lettered
     * @throws java.io.IOException if an error is encountered
     */
    void basicReject(long deliveryTag, boolean requeue) throws IOException;

1.6 channel.basicConsume(QUEUE_NAME, true, consumer);

autoAck：是否自動ack，如果不自動ack，需要使用channel.ack、channel.nack、channel.basicReject 進行消息應答

    /**
     * Start a non-nolocal, non-exclusive consumer, with
     * a server-generated consumerTag.
     * @param queue the name of the queue
     * @param autoAck true if the server should consider messages
     * acknowledged once delivered; false if the server should expect
     * explicit acknowledgements
     * @param callback an interface to the consumer object
     * @return the consumerTag generated by the server
     * @throws java.io.IOException if an error is encountered
     * @see com.rabbitmq.client.AMQP.Basic.Consume
     * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
     * @see #basicConsume(String, boolean, String, boolean, boolean, Map, Consumer)
     */
    String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;

1.7 chanel.exchangeBind()

 

channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
用於通過綁定bindingKey將queue到Exchange，之后便可以進行消息接收

 

    /**
     * Bind an exchange to an exchange, with no extra arguments.
     * @see com.rabbitmq.client.AMQP.Exchange.Bind
     * @see com.rabbitmq.client.AMQP.Exchange.BindOk
     * @param destination the name of the exchange to which messages flow across the binding
     * @param source the name of the exchange from which messages flow across the binding
     * @param routingKey the routine key to use for the binding
     * @return a binding-confirm method if the binding was successfully created
     * @throws java.io.IOException if an error is encountered
     */
    Exchange.BindOk exchangeBind(String destination, String source, String routingKey) throws IOException;

1.8 channel.queueDeclare(QUEUE_NAME, false, false, false, null);

 

durable：true、false true：在服務器重啟時，能夠存活

exclusive ：是否為當前連接的專用隊列，在連接斷開后，會自動刪除該隊列，生產環境中應該很少用到吧。
autodelete：當沒有任何消費者使用時，自動刪除該隊列。this means that the queue will be deleted when there are no more processes consuming messages from it.

 

 /**
     * Declare a queue
     * @see com.rabbitmq.client.AMQP.Queue.Declare
     * @see com.rabbitmq.client.AMQP.Queue.DeclareOk
     * @param queue the name of the queue
     * @param durable true if we are declaring a durable queue (the queue will survive a server restart)
     * @param exclusive true if we are declaring an exclusive queue (restricted to this connection)
     * @param autoDelete true if we are declaring an autodelete queue (server will delete it when no longer in use)
     * @param arguments other properties (construction arguments) for the queue
     * @return a declaration-confirm method to indicate the queue was successfully declared
     * @throws java.io.IOException if an error is encountered
     */
    Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,
                                 Map<String, Object> arguments) throws IOException;