RabbitMQ簡介

RabbitMQ是消費-生產者模型的一個典型的代表，一端往訊息佇列中不斷寫入訊息，而另一端則可以讀取或者訂閱佇列中的訊息。它是一個實現了AMQP協議的可複用的企業訊息系統。RabbitMQ是用Erlang開發。那麼RabbitMQ也具備容錯，熱更新，分散式快取，持久化等特點。RabbitMQ可以輕鬆的搭建出大型可用的分散式佇列訊息叢集。

RabbitMQ的結構圖：

RabbitMQ的一些概念：

Broker：簡單來說就是訊息佇列伺服器實體。
Exchange：訊息交換機，它指定訊息按什麼規則，路由到哪個佇列。
Queue：訊息佇列載體，每個訊息都會被投入到一個或多個佇列。
Binding：繫結，它的作用就是把exchange和queue按照路由規則繫結起來。
Routing Key：路由關鍵字，exchange根據這個關鍵字進行訊息投遞。
vhost：虛擬主機，一個broker裡可以開設多個vhost，用作不同使用者的許可權分離。
producer：訊息生產者，就是投遞訊息的程式。
consumer：訊息消費者，就是接受訊息的程式。
channel：訊息通道，在客戶端的每個連線裡，可建立多個channel，每個channel代表一個會話任務。

RabbitMQ是用Erlang開發的，但對於主要的程式語言都有驅動或者客戶端。我們這裡要用的是Java來作為客戶端語言。假設都已經搭建好環境了（沒有的話看這裡），首先我們下載Java客戶端類庫，點選這裡，下圖是例子的模型：

首先建立佇列，生產者向佇列傳送訊息，消費者向佇列取出訊息，過程很簡單，下面正式開始吧~~

步驟如下：

（一）建立訪問使用者

[[email protected] sbin]# ./rabbitmqctl add_user test test
Creating user "test" ...

[[email protected]
 
 sbin]# ./rabbitmqctl set_permissions -p "/" test ".*" ".*" ".*"          
Setting permissions for user "test" in vhost "/" ...

（三）編寫生產者程式並啟動

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

public class HelloWorldProducer {
    
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception{
        ConnectionFactory cf = new ConnectionFactory();
        //rabbitmq監聽IP
        cf.setHost("192.168.1.96");
        //rabbitmq預設監聽埠，注意要記得開啟埠
        cf.setPort(5672);
        
        //設定訪問的使用者
        cf.setUsername("test");
        cf.setPassword("test");
        //建立連線
        Connection conn = cf.newConnection();
        //建立訊息通道
        Channel channel = conn.createChannel();

        String msg = "hello world!!!! 你好啊~";
        //建立hello佇列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        //傳送訊息
        channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
        System.out.println("send msg "+ msg + " to ["+ QUEUE_NAME +"] queue !");
        
        channel.close();
        conn.close();

    }

}
 

[[email protected] sbin]# ./rabbitmqctl list_queues -p "/"
Listing queues ...
hello   1
[[email protected] sbin]# 

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

public class HelloWorldConsumer {

    private final static String QUEUE_NAME = "hello";
    public static void main(String[] args) throws IOException, TimeoutException {

        ConnectionFactory cf = new ConnectionFactory();
        //rabbitmq監聽IP
        cf.setHost("192.168.1.96");
        //rabbitmq預設監聽埠，注意要記得開啟埠
        cf.setPort(5672);
        
        //設定訪問的使用者
        cf.setUsername("test");
        cf.setPassword("test");
        //建立連線
        Connection conn = cf.newConnection();
        //建立訊息通道
        Channel channel = conn.createChannel();

        //建立hello佇列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" Waiting for msg....");
        //建立消費者，並接受訊息
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                String msg = new String(body, "UTF-8");
                System.out.println("Received is = '" + msg + "'");
            }
        };
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }

}

[[email protected] sbin]# ./rabbitmqctl list_queues -p "/"
Listing queues ...
hello   0
[[email protected] sbin]# 

• 如果連線不成功可能是IP不對或者是5672埠沒開啟，可以在rabbitmq.config中修改(預設是{tcp_listeners, [{"127.0.0.1",5672}]})
  
• 如果用guest(預設使用者)訪問，監聽Host則必須是localhost
  
• 如果是新建立的使用者，記得給使用者繫結vhost並新增許可權
  
• 可以從RabbitMQ的日誌檔案 [email protected] ，獲取更多詳細的錯誤資訊