RabbitMQ通信模型中的work模型

上一篇文章中，簡單的介紹了一下RabbitMQ，以及安裝和hello world。

有的小伙伴留言說看不懂其中的方法參數(shù)，這里先解釋一下幾個基本的方法參數(shù)。

复制// 聲明隊列方法
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
/**
 * param1：queue 隊列的名字
 * param2：durable 是否持久化；比如現(xiàn)在發(fā)送到隊列里面的消息，如果沒有持久化，重啟這個隊列后數(shù) 據(jù)會丟失(false) true：重啟之后數(shù)據(jù)依然在
 * param3：exclusive 是否排外(是否是當前連接的專屬隊列)，排外的意思是：
 *            1：連接關(guān)閉之后 這個隊列是否自動刪除（false：不自動刪除）
 *            2：是否允許其他通道來進行訪問這個數(shù)據(jù)（false：不允許） 
 * param4：autoDelete 是否自動刪除
 *            就是當最后一個連接斷開的時候，是否自動刪除這個隊列（false：不刪除）
 * param5：arguments(map) 聲明隊列的時候，附帶的一些參數(shù)
 */

复制// 發(fā)送數(shù)據(jù)到隊列
channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, "第一個隊列消息...".getBytes());
/**
 * param1：exchange  交換機  沒有就設(shè)置為 "" 值就可以了
 * param2：routingKey 路由的key  現(xiàn)在沒有設(shè)置key，直接使用隊列的名字
 * param3：BasicProperties 發(fā)送數(shù)據(jù)到隊列的時候，是否要帶一些參數(shù)。
 *      MessageProperties.PERSISTENT_TEXT_PLAIN表示沒有帶任何參數(shù)
 * param4：body 向隊列中發(fā)送的消息數(shù)據(jù)
 */

Work模型

work模型稱為工作隊列或者競爭消費者模式，多個消費者消費的數(shù)據(jù)之和才是原來隊列中的所有數(shù)據(jù)，適用于流量的削峰。

img

演示

寫個簡單的測試：

1. 生產(chǎn)者

   复制public class Producer {
       private static final String QUEUE_NAME = "queue_work_1";
   
       public static void main(String[] args) throws IOException, TimeoutException {
           Connection connection = ConnectionUtils.getConnection();
           Channel channel = connection.createChannel();
           channel.queueDeclare(QUEUE_NAME, false, false, false, null);
           for (int i = 0; i < 100; i++) {
               channel.basicPublish("", QUEUE_NAME, null, ("work模型：" + i).getBytes());
           }
           channel.close();
           connection.close();
       }
   
   }
   

2. 消費者

   复制// 消費者1
   public class Consumer {
       private static final String QUEUE_NAME = "queue_work_1";
   
       public static void main(String[] args) throws IOException, TimeoutException {
           Connection connection = ConnectionUtils.getConnection();
           Channel channel = connection.createChannel();
           channel.queueDeclare(QUEUE_NAME, false, false, false, null);
           // channel.basicQos(0, 1, false);
           DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
               @Override
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                   System.out.println(System.currentTimeMillis() + "消費者1接收到信息：" + new String(body));
                   channel.basicAck(envelope.getDeliveryTag(), false);
               }
           };
           channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
       }
   
   }
   

   复制// 消費者2
   public class Consumer2 {
       private static final String QUEUE_NAME = "queue_work_1";
   
       public static void main(String[] args) throws IOException, TimeoutException {
           Connection connection = ConnectionUtils.getConnection();
           Channel channel = connection.createChannel();
           channel.queueDeclare(QUEUE_NAME, false, false, false, null);
           // channel.basicQos(0, 1, false);
           DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
               @Override
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                   System.out.println(System.currentTimeMillis() + "消費者2接收到信息：" + new String(body));
                   channel.basicAck(envelope.getDeliveryTag(), false);
                   // 這里加了個延遲，表示處理業(yè)務時間
                   try {
                       Thread.sleep(200);
                   } catch (InterruptedException e) {
                       e.printStackTrace();
                   }
               }
           };
           channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
       }
   }
   

3. 結(jié)果
   
   image-20221229210012145

image-20221229210046184

可以看出來：100條消息，消費者之間是平分的，消費者1 幾乎是瞬間完成，消費者2 則是慢慢吞吞的運行完畢，消費者1大量時間處于空閑狀態(tài)，消費者2則一直忙碌。這顯然是不適用于實際開發(fā)中。

我們需要遵從一個原則，就是 能者多勞 ，消費越快的人，消費的越多；

現(xiàn)在我們把消費者1和2的代碼中 复制// channel.basicQos(0, 1, false); 這行代碼取消注釋，再次運行；

image-20221229211317632

image-20221229211335782

現(xiàn)在的結(jié)果就比較符合能者多勞，雖然你干的多，但是工資是一樣的呀~

work模型的一個主要的方法是复制basicQos()；這里也解釋一下其參數(shù)：

复制// 設(shè)置限流機制
channel.basicQos(0, 1, false);
/**  
 *  param1: prefetchSize，消息本身的大小 如果設(shè)置為0  那么表示對消息本身的大小不限制
 *  param2: prefetchCount，告訴rabbitmq不要一次性給消費者推送大于N個消息
 *  param3：global，是否將上面的設(shè)置應用于整個通道，false表示只應用于當前消費者
 */

小結(jié)

本文到這里就結(jié)束了，主要介紹了RabbitMQ通信模型中的work模型，適用于限流、削峰等應用場景。