1. 程式人生 > >RabbitMQ-Work模式

RabbitMQ-Work模式

一、簡介

在《RabbitMQ-簡單佇列》中,只有一個生產者一個消費者。今天我們一起學習Work模式,即一個生產者,多個消費者,如下圖所示:

需要注意的是:

(1)生產者的訊息是傳送到一個佇列裡,所以即使有兩個消費者,一個訊息只能被一個消費者消費。

(2)Work模式中可以分為兩種模式:一兩個消費者平均消費佇列中的訊息,即使它們的消費能力是不一樣的;二能者多勞模式,消費能力強的消費者會獲取更多的訊息。

二、編碼實現

2.1、生產者

向佇列傳送50條訊息,生產者每生產一條訊息後都會休眠一段時間。

public class Producer {

  private final static String QUEUE_NAME = "work_queue";

  public static void main(String[] argv) throws Exception {
    // 獲取到連線以及mq通道
    Connection connection = ConnectionUtil.getConnection();
    Channel channel = connection.createChannel();
    // 宣告佇列
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);

    for (int i = 0; i < 50; i++) {
      // 訊息內容
      String message = "訊息:" + i;
      channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
      // 休眠
      Thread.sleep(i * 10);
    }

    channel.close();
    connection.close();
  }
}

2.2、消費者1

public class Consumer1 {

  public static void main(String[] argv) throws Exception {
    // 獲取到連線以及mq通道
    Connection connection = ConnectionUtil.getConnection();
    Channel channel = connection.createChannel();
    // 宣告佇列
    channel.queueDeclare(QueueUtil.QUEUE_NAME_WORK, false, false, false, null);
    // 定義佇列的消費者
    QueueingConsumer consumer = new QueueingConsumer(channel);
    // 監聽佇列,手動返回完成
    channel.basicConsume(QueueUtil.QUEUE_NAME_WORK, false, consumer);
    // 獲取訊息
    while (true) {
      QueueingConsumer.Delivery delivery = consumer.nextDelivery();
      String message = new String(delivery.getBody());
      System.out.println("Consumer1 Received:" + message);
      // 休眠
      Thread.sleep(10);
      // 返回確認狀態
      channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }
  }
}

2.3、消費者2

public class Cunsumer2 {

  public static void main(String[] argv) throws Exception {
    // 獲取到連線以及mq通道
    Connection connection = ConnectionUtil.getConnection();
    Channel channel = connection.createChannel();
    // 宣告佇列
    channel.queueDeclare(QueueUtil.QUEUE_NAME_WORK, false, false, false, null);
    // 定義佇列的消費者
    QueueingConsumer consumer = new QueueingConsumer(channel);
    // 監聽佇列,手動返回完成狀態
    channel.basicConsume(QueueUtil.QUEUE_NAME_WORK, false, consumer);
    // 獲取訊息
    while (true) {
      QueueingConsumer.Delivery delivery = consumer.nextDelivery();
      String message = new String(delivery.getBody());
      System.out.println("Consumer2 Received:" + message);
      // 休眠1秒
      Thread.sleep(1000);
      channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }
  }
}

2.4、測試

(1)啟動消費者1和消費者2。

(2)啟動生產者,佇列work_queue中初始有50條訊息待消費,隨後被消費者1和消費者2消費。

(3)觀察控制檯。

  • 消費者1和消費者2獲取到的訊息內容是不同的,同一個訊息只能被一個消費者獲取。
  • 消費者1和消費者2獲取到的訊息的數量是相同的,一個是奇數一個是偶數。

 

大家看到這,可能會問,既然消費者1的能力大於消費者2(消費者1的休眠時間更短),那消費者1的消費數量是不是應該更多些呢?從一定角度上來說,應該是這樣的。這也是我們接下來要描述的“能者多勞模式”,如下:

public class Consumer1 {

  public static void main(String[] argv) throws Exception {
    // 獲取到連線以及mq通道
    Connection connection = ConnectionUtil.getConnection();
    Channel channel = connection.createChannel();
    // 宣告佇列
    channel.queueDeclare(QueueUtil.QUEUE_NAME_WORK, false, false, false, null);
    // 同一時刻伺服器只會發一條訊息給消費者(上一條訊息消費完了,才會消費下一條)
    channel.basicQos(1);
    // 定義佇列的消費者
    QueueingConsumer consumer = new QueueingConsumer(channel);
    // 監聽佇列,手動返回完成
    channel.basicConsume(QueueUtil.QUEUE_NAME_WORK, false, consumer);
    // 獲取訊息
    while (true) {
      QueueingConsumer.Delivery delivery = consumer.nextDelivery();
      String message = new String(delivery.getBody());
      System.out.println("Consumer1 Received:" + message);
      // 休眠
      Thread.sleep(10);
      // 返回確認狀態
      channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }
  }
}

不知道大家有沒有注意到,我們在消費者1這裡加了channel.basicQos(1)這行程式碼,用於表示同一時刻伺服器只會傳送一條訊息給消費者。同理,我們在消費者2也加上這行程式碼。然後重新進行測試: