上一節的兩個工程，一個負責傳送，一個負責接收，也就是一一對於的關係。

     只要訊息發出了，接收者就處理；當接收效率較低時，就會出現接收者處理不過來，我們就可能會處理不過來，於是我們就可能多配置接受者。這個模式就是"Work queues",它的結構如下

    多個接收者，它們會出現什麼情況呢？是否像大鍋飯，有的人撐死，有的人餓死。這個通過例子驗證。

一、再建一個接收者工程 HelloReceiving2

1、把HelloReceiver工程中的HelloRabbitConfig、HelloReceiver、logback.xml依次拷貝過去

2、修改application.properties為

#伺服器配置
spring.application.name=rabbitmq-hello-receiving
server.port=9092 
#rabbitmq連線引數
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=test
spring.rabbitmq.password=123456

二、執行工程

1、在工程HelloSending所在資料夾開啟cmd,執行mvn spring-boot:run

2、在工程HelloReceiving所在資料夾開啟cmd,執行mvn spring-boot:run

3、在工程HelloReceiving2所在資料夾開啟cmd,執行mvn spring-boot:run
4、在瀏覽器中輸入http://localhost:9080/send/上帝1，http://localhost:9080/send/上帝2，http://localhost:9080/send/上帝3
觀察兩個Receiving的日誌.

查看出不均衡吧，為了突出這個不公平，我們修改傳送程式碼如下

package com.example;

import java.util.Date;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class HelloSender {

	protected static Logger logger=LoggerFactory.getLogger(HelloSender.class); 
	
	@Autowired
    private AmqpTemplate rabbitTemplate;

    public String send(String name) {
    	String context = "hello "+name+" --" + new Date();
    	String sendStr;
    	for(int i=1;i<=100;i++){
    		sendStr="第["+i+"]個 hello "+name+" --" + new Date(); 
    		logger.debug("HelloSender: " + sendStr);
    		this.rabbitTemplate.convertAndSend("hello", sendStr);
    	}
		return context;
    }
}
 

再次http://localhost:9080/send/上帝，會發現更多的不公平。

三、Message acknowledgment 訊息確認

1、預設情況下，RabbitMQ 會順序的分發每個Message。當分發後，會將該Message刪除，然後將下一個Message分發到下一個Consumer。這種分發方式叫做round-robin

2、每個Consumer可能需要一段時間才能處理完收到的資料。如果在這個過程中，Consumer出錯了，異常退出了，而資料還沒有處理完成，那麼非常不幸，這段資料就丟失了。因為我們採用no-ack的方式進行確認，也就是說，每次Consumer接到資料後，而不管是否處理完成，RabbitMQ Server會立即把這個Message標記為完成，然後從queue中刪除了。

3、如果一個Consumer異常退出了，它處理的資料能夠被另外的Consumer處理，這樣資料在這種情況下就不會丟失了（注意是這種情況下）。

4、為了保證資料不被丟失，RabbitMQ支援訊息確認機制，即acknowledgments。為了保證資料能被正確處理而不僅僅是被Consumer收到，那麼我們不能採用no-ack。而應該是在處理完資料後傳送ack。

5、在處理資料後傳送的ack，就是告訴RabbitMQ資料已經被接收，處理完成，RabbitMQ可以去安全的刪除它了。

6、如果Consumer退出了但是沒有傳送ack，那麼RabbitMQ就會把這個Message傳送到下一個Consumer。這樣就保證了在Consumer異常退出的情況下資料也不會丟失。

7、這裡並沒有用到超時機制。RabbitMQ僅僅通過Consumer的連線中斷來確認該Message並沒有被正確處理。也就是說，RabbitMQ給了Consumer足夠長的時間來做資料處理。

    訊息確認，對於spring-boot來說，就是一個開關，它就是spring.rabbitmq.listener.acknowledge-mode

    acknowledgeMode有三值：

    A、NONE = no acks will be sent (incompatible with channelTransacted=true).

          RabbitMQ calls this "autoack" because the broker assumes all messages are acked without any action from the consumer.

    B、MANUAL = the listener must acknowledge all messages by calling Channel.basicAck().

    C、AUTO = the container will acknowledge the message automatically, unless the MessageListener throws an exception.

Note that acknowledgeMode is complementary to channelTransacted - if the channel is transacted then the broker requires a commit notification in addition to the ack. This is the default mode. See also txSize.

    非常簡單，在application.properties中增加spring.rabbitmq.listener.acknowledge-mode=AUTO

為了更好的演示異常，我們把生產者、消費者都做了sleep.程式碼如下：

sending

package com.example;

import java.util.Date;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class HelloSender {

	protected static Logger logger=LoggerFactory.getLogger(HelloSender.class); 
	
	@Autowired
    private AmqpTemplate rabbitTemplate;

    public String send(String name) throws InterruptedException {
    	String context = "hello "+name+" --" + new Date();
    	String sendStr;
    	for(int i=1;i<=100;i++){
    		sendStr="第["+i+"]個 hello "+name+" --" + new Date(); 
    		logger.debug("HelloSender: " + sendStr);
    		this.rabbitTemplate.convertAndSend("hello", sendStr);
    		Thread.sleep(1000);
    	}
		return context;
    }
}

package com.example;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {
	protected static Logger logger = LoggerFactory.getLogger(HelloReceiver.class);

	@RabbitHandler
	public void process(String hello) {
		logger.debug("HelloReceiver : " + hello);
		try {
			Thread.sleep(2000);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}

package com.example;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {
	protected static Logger logger = LoggerFactory.getLogger(HelloReceiver.class);

	@RabbitHandler
	public void process(String hello) {
		logger.debug("HelloReceiver : " + hello);
		try {
			Thread.sleep(3000);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}

四、訊息持久化

在上一節中我們知道了即使Consumer異常退出，Message也不會丟失。但是如果RabbitMQ Server退出呢？軟體都有bug，即使RabbitMQ Server是完美毫無bug的（當然這是不可能的，是軟體就有bug，沒有bug的那不叫軟體），它還是有可能退出的：被其它軟體影響，或者系統重啟了，系統panic了。。。

為了保證在RabbitMQ退出或者crash了資料仍沒有丟失，需要將queue和Message都要持久化。

queue持久化，就是在例項時呼叫具有引數durable的建構函式.

package com.example;

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class HelloRabbitConfig {

    @Bean
    public Queue helloQueue() {
    	return new Queue("hello",true);
    }
}

五、單消費者，排他佇列

訊息只能一個消費者接收處理，其它消費者只能看著，這也是佇列例項時呼叫具有引數exclusive 的建構函式。

    /**
     * Construct a new queue, given a name, durability, exclusive and auto-delete flags.
     * @param name the name of the queue.
     * @param durable true if we are declaring a durable queue (the queue will survive a server restart)
     * @param exclusive true if we are declaring an exclusive queue (the queue will only be used by the declarer's
     * connection)
     * @param autoDelete true if the server should delete the queue when it is no longer in use
     */
    public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete) {
        this(name, durable, exclusive, autoDelete, null);
    }

六、引數

• spring.rabbitmq.addresses指定client連線到的server的地址，多個以逗號分隔.

• spring.rabbitmq.dynamic是否建立AmqpAdmin bean. 預設為: true)

• spring.rabbitmq.host指定RabbitMQ host.預設為: localhost)

• spring.rabbitmq.listener.acknowledge-mode指定Acknowledge的模式.

• spring.rabbitmq.listener.auto-startup是否在啟動時就啟動mq，預設: true)

• spring.rabbitmq.listener.concurrency指定最小的消費者數量.

• spring.rabbitmq.listener.max-concurrency指定最大的消費者數量.

• spring.rabbitmq.listener.prefetch指定一個請求能處理多少個訊息，如果有事務的話，必須大於等於transaction數量.

• spring.rabbitmq.listener.transaction-size指定一個事務處理的訊息數量，最好是小於等於prefetch的數量.

• spring.rabbitmq.password指定broker的密碼.

• spring.rabbitmq.port指定RabbitMQ 的埠，預設: 5672)

• spring.rabbitmq.requested-heartbeat指定心跳超時，0為不指定.

• spring.rabbitmq.ssl.enabled是否開始SSL，預設: false)

• spring.rabbitmq.ssl.key-store指定持有SSL certificate的key store的路徑

• spring.rabbitmq.ssl.key-store-password指定訪問key store的密碼.

• spring.rabbitmq.ssl.trust-store指定持有SSL certificates的Trust store.

• spring.rabbitmq.ssl.trust-store-password指定訪問trust store的密碼.

• spring.rabbitmq.username指定登陸broker的使用者名稱.

• spring.rabbitmq.virtual-host指定連線到broker的Virtual host.