摘要：知識庫前言舍棄了繁重的事務消息而使用了消息確認機制實現了分布式事務，實在是解耦之一大神器。但是其配置起來挺麻煩，各種參數，各種調整。消費確認消費確認用來確保消費者是否成功的消費了消息。

rabbitmq知識庫: http://rabbitmq.org.cn/

Rabbitmq舍棄了繁重的事務消息而使用了消息確認機制實現了分布式事務，實在是解耦之一大神器。但是其配置起來挺麻煩，各種參數，各種調整。但國內貌似資料很少，找來找去都找不到，自己擼一發先

發送確認用來確保消息是否已送達消息隊列。消息一旦到達消息服務，就會觸發確認機制，可以分為兩種情況：

對于無法被路由的消息，一旦沒法找到一個隊列來消費它，就會觸發確認無法消費，此時ack＝false。旦有一種情況例外，就是連exchange都沒法找到，如果設置了mandatory, 此時就會先觸發basic.return，就是會先觸發returncallback回調。

對于可以被路由的消息，當消息被（所有的？）queue接受時，會觸發ack=true;對于設置了持久化（persistent）的消息，當消息成功的持久化到硬盤上才會觸發；對于設置了鏡像（mirror）的消息，那么是當所有的mirror接受到這個消息。

消費確認用來確保消費者是否成功的消費了消息。一旦有消費者成功注冊到相應的消息服務，消息將會被消息服務通過basic.deliver推（push）給消費者，此時消息會包含一個deliver tag用來唯一的標識消息。如果此時是手動模式，就需要手動的確認消息已經被成功消費，否則消息服務將會重發消息（因為消息已經持久化到了硬盤上，所以無論消息服務是不是可能掛掉，都會重發消息）。而且必須確認，無論是成功或者失敗，否則會引起非常嚴重的問題

有三種情況可能進死信交換機

被reject或者nack，并且requeue設置為false

消息最大存活時間（TTL）超時

消息數量超過最大隊列長度

只需要設置一個args，就ok拉

channel.exchangeDeclare("some.exchange.name", "direct");
Map args = new HashMap();
args.put("x-dead-letter-exchange", "some.exchange.name");
channel.queueDeclare("myqueue", false, false, false, args);

Channel Prefetch Setting (QoS)，表示當前channel中未應答消息的數目，如果超過了，隊列中將不再接受新的消息。這里所謂的channel就是指從消息服務到消費者的一個通道，簡單來說就是指消息從消息隊列發送到消費者了，如果沒收到應答，就算是一個Qos

加大這個值會增加消息的發送速度（Throughput），但是會加重消息隊列的內存，所以100-300之間是一個比較理想的狀態，可參考：http://next.rabbitmq.com/conf... :Channel Prefetch Setting (QoS)。

暴力的設置微100-300是存在一些問題的，如果太大，可能消息全部都壓在消費者中而得不到消費，看起來隊列是空的，實際上全部積壓在客戶端；如果太小則得不到消費，浪費資源。具體該怎么設置要根據實際的網絡吞吐量、以及消費者的消費能力。比如果消費者很快，是內存操作，那么你設置很大，甚至不設置都可以；但是如果消費者很慢，比如是個數據庫操作，那么很可能將消息全部積壓到消費者而得不到響應
。可以參考http://www.rabbitmq.com/blog/...

Qos同時也會存在一個問題，一個channel是會被多個消費者的，所以必須計算出所有消費者中未應答的數目，這顯然是非常不合理的，而且很麻煩，所以可以改為設置每個消費者緩存（Prefetch Buf）可以允許的最大的數目。

例子：

// 1. 一下子設置所有的queue都是10條unacknowledged
Channel channel = ...;
Consumer consumer = ...;
channel.basicQos(10); // Per consumer limit
channel.basicConsume("my-queue", false, consumer);

//2. 分別設置10條
Channel channel = ...;
Consumer consumer1 = ...;
Consumer consumer2 = ...;
channel.basicQos(10); // Per consumer limit
channel.basicConsume("my-queue1", false, consumer1);
channel.basicConsume("my-queue2", false, consumer2);

//3. 分別設置channel和consume，個人不推薦
Channel channel = ...;
Consumer consumer1 = ...;
Consumer consumer2 = ...;
channel.basicQos(10, false); // Per consumer limit
channel.basicQos(15, true);  // Per channel limit
channel.basicConsume("my-queue1", false, consumer1);
channel.basicConsume("my-queue2", false, consumer2);

當前最大允許空閑的最大channel數。如果在高并發的環境中，如果值過小的話channel會關關開開非常頻繁。所以在1.6的版本中,spring amqp將這個值從1提高到了25。同時你也可以從RabbitMQ Admin中心觀察到channel關關開開，那么就可以考慮增大cache的值了。

當你遇到 connetion error的錯誤時，就可以考慮增大channel cache size了。

通道（channel）中的消息標志，按照正數遞增，消息隊列中用來標識消息的唯一標識

當消息服務器資源不足時，會向所有的生產者發送這個消息，我們可以捕獲這個消息并做處理，資源可以是內存不足，cpu負載過重等等。

ConnectionFactory factory = new ConnectionFactory();
Connection connection = factory.newConnection();
connection.addBlockedListener(new BlockedListener() {
    public void handleBlocked(String reason) throws IOException {
        // Connection is now blocked
    }

    public void handleUnblocked() throws IOException {
        // Connection is now unblocked
    }
});

話說真出現block了也應該是在管理臺直接給出警告，不過也可以做一下避免異常

全應答標識。如果設置為true，一條消息應答了，那么之前的全部消息將被應答。比如目前channel中有delivery tags為5,6,7,8的消息，那么一旦8被應答，那么5，6，7將都被應答，如果設置為false，那么5，6，7將不會被應答。（不建議設置，畢竟一個channel中會綁定好多consumer）

當消息服務發生異常時，不會發送basic.ack，反而會發送一個basic.nack，而且不會自動requeue，此時需要消息發送方手動處理，進行重發。只有一種情況會發送nack：“basic.nack will only be delivered if an internal error occurs in the Erlang process responsible for a queue”

1 mq配置的listener到什么地方，是配置到每個微服務、或者是配置到mq中？如果在每個服務里面都配置concurency，是不是隨著節點的增加，listener數量也會無限增加？
2 ssl和max-queue-length的配置