從本節開始稱Sender為生產者 ， Recv為消費者

一、訊息確認

為了確保訊息一定被消費者處理，rabbitMQ提供了訊息確認功能，就是在消費者處理完任務之後，就給伺服器一個回饋，伺服器就會將該訊息刪除，如果消費者超時不回饋，那麼伺服器將就將該訊息重新發送給其他消費者

預設是開啟的，在消費者端通過下面的方式開啟訊息確認,  首先將autoAck自動確認關閉，等我們的任務執行完成之後，手動的去確認，類似JDBC的autocommit一樣

QueueingConsumer consumer = new QueueingConsumer(channel);
boolean autoAck =
 
 false;
channel.basicConsume("hello", autoAck, consumer);

在前面的例子中使用的是channel.basicConsume(channelName, true, consumer) ; 在接收到訊息後，就會自動反饋一個訊息給伺服器。

Sender03.java

package com.zf.rabbitmq03;

import java.io.IOException;

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

/**
 * 傳送訊息
 * @author zhoufeng
 *
 */
public class Sender03 {
	
	public static void main(String[] args) throws IOException {
		
		
		ConnectionFactory connFac = new ConnectionFactory() ;
		
		//RabbitMQ-Server安裝在本機，所以直接用127.0.0.1
		connFac.setHost("127.0.0.1");
		
		//建立一個連線
		Connection conn = connFac.newConnection() ;
		
		//建立一個渠道
		Channel channel = conn.createChannel() ;
		
		//定義Queue名稱
		String queueName = "queue01" ;
		
		//為channel定義queue的屬性，queueName為Queue名稱
		channel.queueDeclare( queueName , false, false, false, null) ;
		
		String msg = "Hello World!";
		
		//傳送訊息
		channel.basicPublish("", queueName , null , msg.getBytes());
		
		System.out.println("send message[" + msg + "] to "+ queueName +" success!");
		
		channel.close(); 
		conn.close(); 
		
	}

}
 

Recv03.java

package com.zf.rabbitmq03;

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.QueueingConsumer.Delivery;
import com.rabbitmq.client.ShutdownSignalException;

/**
 * 接收訊息
 * @author zhoufeng
 *
 */
public class Recv03 {

	public static void main(String[] args) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
		
		ConnectionFactory connFac = new ConnectionFactory() ;
		
		connFac.setHost("127.0.0.1");
		
		Connection conn = connFac.newConnection() ;
		
		Channel channel = conn.createChannel() ;
		
		String channelName = "channel01";
		
		channel.queueDeclare(channelName, false, false, false, null) ;
		
		
		//配置好獲取訊息的方式
		QueueingConsumer consumer = new QueueingConsumer(channel) ;
		

		//取消 autoAck
		boolean autoAck = false ;
		
		channel.basicConsume(channelName, autoAck, consumer) ;
		
		//迴圈獲取訊息
		while(true){
			
			//獲取訊息，如果沒有訊息，這一步將會一直阻塞
			Delivery delivery = consumer.nextDelivery() ;
			
			String msg = new String(delivery.getBody()) ;  
			
			//確認訊息，已經收到
			channel.basicAck(delivery.getEnvelope().getDeliveryTag()
					, false);
			
			System.out.println("received message[" + msg + "] from " + channelName);
		}
		
	}
	
}
 

如果將上面的channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);註釋掉， Sender03.java只需要執行一次 ， Recv03.java每次執行將都會收到HelloWorld訊息

注意：

但是這樣還是不夠的，如果rabbitMQ-Server突然掛掉了，那麼還沒有被讀取的訊息還是會丟失 ，所以我們可以讓訊息持久化。 只需要在定義Queue時，設定持久化訊息就可以了，方法如下：

booleandurable=true;channel.queueDeclare(channelName,durable,false,false,null);

二、公平排程

上一個例子能夠實現傳送一個Message與接收一個Message

從上一個Recv01中可以看出，必須處理完一個訊息，才會去接收下一個訊息。如果生產者眾多，那麼一個消費者肯定是忙不過來的。此時就可以用多個消費者來對同一個Channel的訊息進行處理，並且要公平的分配任務給多個消費者。不能部分很忙  部分總是空閒

實現公平排程的方式就是讓每個消費者在同一時刻會分配一個任務。 通過channel.basicQos(1);可以設定