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服務器關閉後信息依然能保存，這樣就提供了消息傳遞的可靠性