RabbitMQ之消息發布訂閱與信息持久化技術
信息發布與訂閱
Rabbit的核心組件包含Queue(消息隊列)和Exchanges兩部分，Exchange的主要部分就是對信息進行路由，通過將消息隊列綁定到Exchange上，則可以實現訂閱形式的消息發布及Publish/Subscribe在這種模式下消息發布者只需要將信息發布到相應的Exchange中，而Exchange則自動將信息分發到不同的Queue當中。
這種模式下Exchange充當的角色
在命令行中可以使用
sudo rabbitmqctl list_exchanges
sudo rabbitmqctl list_bindings
分別查看當前系統種存在的Exchange和Exchange上綁定的Queue信息。
消息發布者EmitLog.java
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class EmitLog {
private static final String EXCHANGE_NAME="logs";
public static void main(String[] args) throws java.io.IOException{
//創建鏈接工廠
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
//創建鏈接
Connection connection = factory.newConnection();
//創建信息管道
Channel channel = connection.createChannel();
//生命Exchange 非持久化
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
String message = "Message "+Math.random();
//第一個參數是對應的Exchange名稱,如果為空則使用默認Exchange
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
System.out.println("[x] Sent '"+message+"'");
//關閉鏈接
channel.close();
connection.close();
}
}
消息消費者ReceiveLogs.java
import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;
public class ReceiveLogs {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] args) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
//創建鏈接工廠
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
//創建鏈接
Connection connection = factory.newConnection();
//創建消息管道
Channel channel = connection.createChannel();
//聲明Exchange
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
//利用系統自動聲明一個非持久化的消息隊列，并返回唯一的隊列名稱
String queueName = channel.queueDeclare().getQueue();
//將消息隊列綁定到Exchange
channel.queueBind(queueName, EXCHANGE_NAME, "");
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
//聲明一個消費者
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, consumer);
while (true) {
//循環獲取信息
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" [x] Received '" + message + "'");
}
}
}
運行時啟動一個EmitLog.java多個ReceiveLogs.java則可以看到發布者每次發布信息，只要綁定到了相應Exchange的消費者都可以獲取到信息。
RabbitMQ信息持久化技術
上面的例子中我們實現了Publisher/Subscribe的消息分發方式，但是其中存在一些問題。比如當我們運行一個ReceiveLog都對應了一個特定的消息隊列，可以利用list_queues進行查看，同時這些消息隊列是幫到到名為logs的Exchange中，這是發布消息每個消費者都可以接收到，可以當關閉ReceiveLog程序后這些消息隊列就都會自動銷毀，因為他們是非持久化的。同樣對于EmitLog程序也一樣，每次關閉后之前生命的Exchange也將自動銷毀。
這就產生了一些問題。如果當ReceiveLog為運行時，此時就并沒有一個消息隊列是綁定到Exchange上的，在發布消息后再啟動ReceiveLog程序是無法接受到之前發布的信息。這就是為什么要進行消息的持久化。
通過持久化技術，我們可以生命一個持久化的Exchange，以及持久化的Queue這樣，在把Queue綁定到Exchange后，即使沒有消費者程序運行，信息依然能保存在Queue當中，當下次啟動消費者程序時依然能獲取到發布的所有信息。就好比當一個消費者程序在執行消息序列中的任務時，如果突然出現了異常那么重新啟動后，依然能從上一次發生錯誤的位置繼續運行，對于某些需要一個有序性和連續性的操作，這點顯的尤為重要。
下面還是給出一個例子，在持久化過程中，可以借助list_exchanges,list_bindings,list_queues來查看服務器中相關信息來幫組分析過程。
Publisher.java
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
public class Publisher {
private static final String EXCHANGE_NAME="persi";//定義Exchange名稱
private static final boolean durable = true;//消息隊列持久化
public static void main(String[] args) throws java.io.IOException {
ConnectionFactory factory = new ConnectionFactory();//創建鏈接工廠
factory.setHost("localhost");
Connection connection = factory.newConnection();//創建鏈接
Channel channel = connection.createChannel();//創建信息通道
channel.exchangeDeclare(EXCHANGE_NAME, "fanout", durable);//創建交換機并生命持久化
String message = "Hello Wrold "+Math.random();
//消息的持久化
channel.basicPublish(EXCHANGE_NAME, "", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
System.out.println("[x] Sent '" + message + "'");
channel.close();
connection.close();
}
}
Subscriber.java
public class Subscriber {
//private static final String[] QUEUE_NAMES= {"que_001","que_002","que_003","que_004","que_005"};
private static final String[] QUEUE_NAMES= {"que_006","que_007","que_008","que_009","que_0010"};
public static void main(String[] args){
for(int i=0;i<QUEUE_NAMES.length;i++){
SubscriberThead sub = new SubscriberThead(QUEUE_NAMES[i]);
Thread t = new Thread(sub);
t.start();
}
}
}
SubscriberThead.java
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.AMQP.Queue.DeclareOk;
public class SubscriberThead implements Runnable {
private String queue_name = null;
private static final String EXCHANGE_NAME = "persi";// 定義交換機名稱
private static final boolean durable = true;//消息隊列持久化
public SubscriberThead(String queue_name) {
this.queue_name = queue_name;
}
@Override
public void run() {
try{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "fanout", durable);
DeclareOk ok = channel.queueDeclare(queue_name, durable, false,
false, null);
String queueName = ok.getQueue();
channel.queueBind(queueName, EXCHANGE_NAME, "");
System.out.println(" ["+queue_name+"] Waiting for messages. To exit press CTRL+C");
channel.basicQos(1);//消息分發處理
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, false, consumer);
while (true) {
Thread.sleep(2000);
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" ["+queue_name+"] Received '" + message + "'");
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}catch(Exception e){
e.printStackTrace();
}
}
}
通過持久化處理后rabbitMQ將保存Exchange信息以及Queue信息，甚至在rabbitMQ服務器關閉后信息依然能保存，這樣就提供了消息傳遞的可靠性