大綱

1.訊息中介軟體概述,使用場景(日誌處理,非同步,系統解耦,流量削鋒)
2.Rabbitmq3.7.2安裝,控制檯簡介,管理員新增
3.使用者vhost配置以及介紹
4.java操作簡單佇列,生產者傳送訊息到佇列,消費者接收訊息
5.簡單佇列的缺陷,工作佇列work queues之 輪詢分發(Round-robin),以及輪詢分發現象
6.工作佇列work queues 公平分發(fair dispatch);prefetchCount = 1來限制RabbitMQ傳送的訊息,手動應答ack。
7.訊息應答ack與訊息持久化durable
8.publish/subscribe釋出訂閱模式 交換機(轉發器)匿名轉發Nameless exchange, Fanout Exchange不處理路由鍵 , Direct Exchange處理路由鍵, Topic Exchange將路由鍵和某模式進行匹配。佇列繫結交換機(Exchange) ;
9.routing路由模式
10.topic主題模式
11.Rabbitmq之訊息確認機制(AMQP事務機制)txSelect(), txCommit()以及txRollback(),事務機制的缺陷
12.Rabbitmq之訊息確認機制(Confirm機制序列) waitForConfirms
13.Rabbitmq之訊息確認機制(Confirm機制非同步) ConfirmListener deliveryTag unconfirm集合維護
14.spring整合rabbitmq-client,template 的使用
15.大資料日誌收集系統訊息中介軟體應用
16.搜尋系統DIH訊息中介軟體應用

工作佇列 公平分發

模型

輪詢分發解決了一部分壓力，但是有個問題就是：比如：現在有2個消費者，消費者1都是繁忙的，而消費者2則是輕鬆的。按照輪詢的方式，消費者1所以一直在忙個不停。消費者2則立即完成任務，然後很空閒。

而RabbitMQ則是不瞭解這些的。他是不知道你消費者的處理訊息的能力,這是因為當訊息進入佇列，RabbitMQ就會分派訊息。而rabbitmq只是盲目的將訊息輪詢的發給消費者。你一個我一個的這樣傳送.
為了解決這個問題，我們使用basicQos( prefetchCount = 1)方法，來限制RabbitMQ只發不超過1條的訊息給同一個消費者。當訊息處理完畢後，有了反饋ack，才會進行第二次傳送。(也就是說需要手動反饋給Rabbitmq )

還有一點需要注意，使用公平分發，必須關閉自動應答，改為手動應答。

生產者

public class Send {

    private final static String QUEUE_NAME  = "test_queue_work";

    public static void main(String[] argv) throws Exception {
        // 獲取到連線以及mq通道
        Connection connection = ConnectionUtils.getConnection();
          // 建立一個頻道  
        Channel channel = connection.createChannel();
        // 指定一個佇列  
 

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        int prefetchCount = 1; 

 //每個消費者傳送確認訊號之前，訊息佇列不傳送下一個訊息過來，一次只處理一個訊息 
        //限制發給同一個消費者不得超過1條訊息  
        channel.basicQos(prefetchCount);  

        // 傳送的訊息  
        for (int i = 0; i < 50; i++) {
            String message = "." + i;
              // 往佇列中發出一條訊息  
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
            Thread.sleep(i * 10);
        }
         // 關閉頻道和連線  
        channel.close();
        connection.close();
    }
}

消費者1

public class Recv1 {

    private final static String QUEUE_NAME  = "test_queue_work";

    public static void main(String[] args) throws Exception {
        // 獲取到連線以及mq通道
        Connection connection = ConnectionUtils.getConnection();
        final Channel channel = connection.createChannel();

         // 宣告佇列，主要為了防止訊息接收者先執行此程式，佇列還不存在時建立佇列。
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        channel.basicQos(1);//保證一次只分發一個  

        //定義一個訊息的消費者
        final Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println(" [1] Received '" + message + "'");
                try {
                    doWork(message);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    System.out.println(" [x] Done");
                    channel.basicAck(envelope.getDeliveryTag(), false);    
                }
            }

        };
        boolean autoAck = false; //手動確認訊息
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }

    private static void doWork(String task) throws InterruptedException {
        Thread.sleep(1000);
    }
}

消費者2

public class Recv2 {

    private final static String QUEUE_NAME  = "test_queue_work";

    public static void main(String[] args) throws Exception {
        // 獲取到連線以及mq通道
        Connection connection = ConnectionUtils.getConnection();
        final Channel channel = connection.createChannel();

        // 宣告佇列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.basicQos(1);//保證一次只分發一個  
        //定義一個訊息的消費者
        final Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println(" [2] Received '" + message + "'");
                try {
                    doWork(message);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    System.out.println(" [x] Done");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };


        boolean autoAck = false; //關閉自動 確認
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }

    private static void doWork(String task) throws InterruptedException {
        Thread.sleep(2000);
    }

    public static void oldAPi() throws Exception, TimeoutException {
        // 獲取到連線以及mq通道
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();

        // 宣告佇列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 同一時刻伺服器只會發一條訊息給消費者
        channel.basicQos(1);
        // 定義佇列的消費者
        QueueingConsumer consumer = new QueueingConsumer(channel);
        // 監聽佇列，手動返回完成狀態
        channel.basicConsume(QUEUE_NAME, false, consumer);

        // 獲取訊息
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" [x] Received '" + message + "'");
            // 休眠1秒
            Thread.sleep(1000);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}

這時候現象就是消費者1 速度大於消費者2