原理及架構

1. 核心概念
- producer&Consumer
- Broker:它提供一種傳輸服務,它的角色就是維護一條從生產者到消費者的路線，保證資料能按照指定的方式進行傳輸
- Queue

訊息佇列，提供了FIFO的處理機制，具有快取訊息的能力。
rabbitmq中，佇列訊息可以設定為持久化，臨時或者自動刪除。

設定為持久化的佇列，queue中的訊息會在server本地硬碟儲存一份，防止系統crash，資料丟失
設定為臨時佇列，queue中的資料在系統重啟之後就會丟失
設定為自動刪除的佇列，當不存在使用者連線到server，佇列中的資料會被自動刪除

• Exchange

Exchange類似於資料通訊網路中的交換機，提供訊息路由策略。rabbitmq中，producer不是通過通道直接將訊息傳送給queue，而是先發送給Exchange。一個Exchange可以和多個Queue進行繫結，producer在傳遞訊息的時候，會傳遞一個ROUTING_KEY，Exchange會根據這個ROUTING_KEY按照特定的路由演算法，將訊息路由給指定的queue。和Queue一樣，Exchange也可設定為持久化，臨時或者自動刪除。

Exchange有4種類型：direct(預設)，fanout, topic, 和headers，不同型別的Exchange轉發訊息的策略有所區別：

Direct
直接交換器，工作方式類似於單播，Exchange會將訊息傳送完全匹配ROUTING_KEY的Queue

fanout
廣播是式交換器，不管訊息的ROUTING_KEY設定為什麼，Exchange都會將訊息轉發給所有繫結的Queue。

topic
主題交換器，工作方式類似於組播，Exchange會將訊息轉發和ROUTING_KEY匹配模式相同的所有佇列，
比如，ROUTING_KEY為user.stock的Message會轉發給繫結匹配模式為 * .stock,user.stock， * . * 和#.user.stock.#的佇列。（ * 表是匹配一個任意片語，#表示匹配0個或多個片語）

headers
訊息體的header匹配（ignore）
 

• Binding

所謂繫結就是將一個特定的 Exchange 和一個特定的 Queue 繫結起來。
Exchange 和Queue的繫結可以是多對多的關係。

• virtual host

在rabbitmq server上可以建立多個虛擬的message broker，又叫做virtual hosts (vhosts)。
每一個vhost本質上是一個mini-rabbitmq server，
分別管理各自的exchange和bindings。vhost相當於物理的server，
可以為不同app提供邊界隔離，使得應用安全的執行在不同的vhost例項上，相互之間不會干擾。
producer和consumer連線rabbit server
 
需要指定一個vhost。
一個broker裡可以有多個vhost，用作不同使用者的許可權分離。

• Channel:訊息通道,在客戶端的每個連線裡,可建立多個channel.

  1. 通訊過程
• 假設P1和C1註冊了相同的Broker，Exchange和Queue。P1傳送的訊息最終會被C1消費。基本的通訊流程大概如下所示：

1. P1生產訊息，傳送給伺服器端的Exchange
2. Exchange收到訊息，根據ROUTINKEY，將訊息轉發給匹配的Queue1
3. Queue1收到訊息，將訊息傳送給訂閱者C1
4. C1收到訊息，傳送ACK給佇列確認收到訊息
5. Queue1收到ACK，刪除佇列中快取的此條訊息

• Consumer收到訊息時需要顯式的向rabbit broker傳送basic.ack訊息或者consumer訂閱訊息時設定auto_ack引數為true。在通訊過程中，佇列對ACK的處理有以下幾種情況：

1. 如果consumer接收了訊息，傳送ack,rabbitmq會刪除佇列中這個訊息，傳送另一條訊息給consumer。
2. 如果cosumer接受了訊息, 但在傳送ack之前斷開連線，rabbitmq會認為這條訊息沒有被deliver,在consumer在次連線的時候，這條訊息會被redeliver。
3. 如果consumer接受了訊息，但是程式中有bug,忘記了ack,rabbitmq不會重複傳送訊息。
4. rabbitmq2.0.0和之後的版本支援consumer reject某條（類）訊息，可以通過設定requeue引數中的reject為true達到目地，那麼rabbitmq將會把訊息傳送給下一個註冊的consumer。

一、安裝

1. 安裝erlang，rpm -ivh erlang-19.0.4-1.el7.centos.x86_64.rpm
2. 安裝rabbitmq，rpm -ivh rabbitmq-server-3.6.12-1.el7.noarch.rpm

二、啟動

1. systemctl start rabbitmq-server
2. 設定開機自啟動，systemctl enable rabbitmq-server

三、配置

1. 建立使用者，因為預設的guest/guest使用者只能在本地登入

例如：建立一個admin/admin123，並讓他成為管理員
1. rabbitmqctl add_user admin admin123
2. rabbitmqctl set_user_tags admin administrator

四、編寫Java Client

1. maven引入amqp-client

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

1. 封裝ConnectionFactory

    public static Connection getConnection(String username,
        String password,
        String host,
        String virtualHost,
        int port) throws IOException, TimeoutException{
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUsername(username);
            factory.setPassword(password);
            factory.setHost(host);
            factory.setVirtualHost(virtualHost);
            factory.setPort(port);
            return factory.newConnection();
    }

1. 訊息生產者

import java.util.concurrent.TimeUnit;

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

public class RabbitMqPublish {

    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMqFactory.getConnection("root", "[email protected]", "192.168.1.8", "/", 5672);

        Channel channel =  connection.createChannel();

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

        while(true){
            try {
                TimeUnit.SECONDS.sleep(5);
                publish(channel);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void publish(Channel channel) throws Exception{
        String message =  System.currentTimeMillis()+"";
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println("Publish: " + message);
    }
}

1. 訊息消費者

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.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class RabbitMqConsumer {

    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws IOException, TimeoutException {

        Connection connection = RabbitMqFactory.getConnection("root", "[email protected]", "192.168.1.8", "/", 5672);

        Channel channel =  connection.createChannel();

        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("Received" + message);
            }
        };

        channel.basicConsume(QUEUE_NAME, true, consumer);

    }

}

1. 先執行生產者，後執行消費者。控制檯會每隔5s列印一次

五、相關參考