RabbitMQ 消息中間件

RabbitMQ 是使用 Erlang 語言開發的消息中間件, 其遵循了高級消息隊列協議(Advanced Message Queuing Protocol, AMQP)。

與 Kafka 等消息隊列相比，RabbitMQ 最大的優點在於其較高的可靠性:

• 提供確認(ACK)和重傳機制保證消息完成消費, 消費者異常不會致使消息丟失

• 提供消息持久化機制, broker 崩潰不會致使消息丟失

• 集羣模式下工做, 保證高可用

由於具備較高可靠性和一致性, RabbitMQ 能夠勝任訂單處理、秒殺等一致性要求較高的業務場景。

RabbitMQ 概念與機制

RabbitMQ 中的概念模型:

• Broker: 消息中間件實例, 多是單個節點也多是運行在多節點集羣上的邏輯實體
• 消息(Message): 消息由消息頭和消息體兩部分組成。消息頭中包括routing-key、priority等標準消息頭以及其它自定義消息頭，用於定義RabbitMQ對消息行爲。消息體是字節流，包含消息內容。
• 鏈接(Connection): 客戶端與 Broker 之間的 TCP鏈接
• 信道(Channel): Channel 是創建在 TCP 鏈接上的邏輯(虛擬)鏈接。多個 Channel 複用同一個 TCP 鏈接, 以免創建 TCP 鏈接的巨大開銷。 RabbitMQ 官方要求每一個線程使用獨立的 Channel, 禁止多個線程共用 Channel。
• 生產者(Publisher): 發送消息的客戶端線程
• 消費者(Consumer): 處理消息的客戶端線程
• 交換機(Exchange): 交換機負責將消息投遞到相應的隊列
• 隊列(Queue): 接收並保存交換機投遞的消息，直至被消費者成功消費。邏輯結構遵循先進先出FIFO。
• 綁定(Binding): 將隊列(Queue)註冊到交換機(Exchange)的路由表
• 虛擬主機(Vhost): 每一個Broker下可創建多個vhost, 每一個 vhost 可創建獨立的 Exchange、Queue、綁定及權限系統。同一個 Broker 下的 vhost 共享 Connection、Channel 和 用戶系統，就是說可使用同一個用戶身份使用同一個 Channel 訪問不一樣 vhost。

交換機(Exchange)

生產者發送的消息會首先送到交換機(Exchange), 交換機根據自身類型和消息的 routing-key 等信息將消息投遞到綁定的消息隊列中。

RabbitMQ中的四種標準交換機:

• direct: 若是消息的 routing-key 與隊列的 binding-key 徹底相同，direct類型的交換機則會將消息投遞到該隊列中。
  • 多個隊列可使用相同的 binding-key 綁定到同一個 direct 交換機，direct 交換機會把消息投遞到全部 binding-key 與消息 routing-key 相同的隊列
• topic: 容許隊列的 binding-key 中包含通配符*和#, topic 交換機會將消息投遞到 binding-key 與 routing-key 匹配的隊列中。
  • 通配符按照關鍵字進行匹配，如news.cn.a中的關鍵字是news、cn和a，即關鍵字按照.分割
  • #通配符匹配0個或多個關鍵字, news.#.a能夠匹配news.a, news.cn.a和news.asia.cn.a等
  • *通配符匹配一個關鍵字, news.*.a匹配news.cn.a不匹配news.a、news.asia.cn.a
• fanout: fanout 交換機不進行任何匹配, 將消息投遞到全部綁定的隊列
• header: header 交換機根據消息頭進行投遞，如今已較少使用

咱們可使用 RabbitMQ 的插件機制使用第三方交換機或自行開發交換機。如實現延時投遞的delayed-message-exchange。

消息頭中的delivery-mode能夠設置爲 persistent(持久化) 或者 transient(易失)。 Exchange 和 Queue 在處理持久化的消息時都會先將消息寫入磁盤中再進行下一步處理, 即便 RabbitMQ 崩潰也不會丟失。

消費者客戶端一般使用的channel.basicConsume使用推(push)模式投遞消息, 即當有新消息時 Broker 經過 channel 主動向客戶端發送消息。客戶端也可使用channel.basicGet從 Broker 拉取消息。

ACK機制

RabbitMQ 提供了確認送達(acknowledge)機制保證消息被正確處理不會丟失。

確認送達的回執有三種:

• ACK: 消息已被成功處理
• NACK: 消息處理異常, 須要從新投遞
• REJECT: 消息非法, 丟棄消息

RabbitMQ 的 Queue 能夠設置 no_ack=true, 則消息被投遞後即刪除不等待回執。

channel.basicConsume 能夠指定auto_ack模式，若auto_ack=true當客戶端收到完整消息後即會自動發出ACK回執，不然必須顯式的發出回執。

Java 代碼示例

首先安裝並啓動RabbitMQ實例, Mac用戶可使用 Homebrew 進行安裝:

brew install rabbitmq

啓動服務:

brew services start rabbitmq

或者使用官方docker鏡像:

docker run -d --hostname my-rabbit --name some-rabbit rabbitmq:3-management

RabbitMQ官網提供了Ubuntu、RPM以及Windows等多種平臺安裝方式。

RabbitMQ默認TCP端口爲5672, Web控制檯默認端口15672。

在Maven中添加依賴:

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.5.1</version>
</dependency>

編寫生產者:

package rabbit;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

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

/**
 * @author finley
 */
public class RabbitProducer {

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setHost("localhost");
        try (Connection conn = factory.newConnection();
             Channel channel = conn.createChannel()) {
            String exchangeName = "test-exchange";
            channel.exchangeDeclare(exchangeName, "direct", true);

            String routingKey = "hello";

            byte[] msg = "hello world".getBytes();
            AMQP.BasicProperties.Builder propsBuilder = new AMQP.BasicProperties.Builder();
            propsBuilder.deliveryMode(2); // persistent
            propsBuilder.priority(0); // normal
            propsBuilder.contentType("text/plain");
            channel.basicPublish(exchangeName, routingKey, propsBuilder.build(), msg);
        }
    }
}

編寫消費者:

package rabbit;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.*;

/**
 * @author finley
 */
public class RabbitConsumer {

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setHost("localhost");
        try (Connection conn = factory.newConnection();
             Channel channel = conn.createChannel()) {
            String exchangeName = "test-exchange";
            channel.exchangeDeclare(exchangeName, "direct", true);

            String queueName = channel.queueDeclare().getQueue();
            String bindingKey = "hello";
            channel.queueBind(queueName, exchangeName, bindingKey);

            while(true) {
                channel.basicConsume(queueName, false, "", new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        String routingKey = envelope.getRoutingKey();
                        String contentType = properties.getContentType();
                        String bodyStr = new String(body, "UTF-8");
                        System.out.println("routingKey: " + routingKey + ", contentType: " + contentType + ", body: " + bodyStr);
                        long deliveryTag = envelope.getDeliveryTag();
                        channel.basicAck(deliveryTag, false);
                    }
                });
            }
        }
    }

}

RabbitMQ 的消息爲字節, 能夠將 Java 對象序列化後做爲消息體發送。