RabbitMQ的work queues模式_rabbitmq work queues模式

RabbitMQ的work queues模式

• workQueues模式图片
  
• Work Queues：与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。
• 应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

• 代码编写生产者步骤 (与简单模式生产者一模一样)
  1. 创建工厂
  2. 设置参数
  3. 新建连接
  4. 创建channel(管道)
  5. 创建队列
  6. 发送消息
  7. 关闭连接

package com.yang;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Producer_WorkQueues {
    public static void main(String[] args) throws Exception {
        //1. 创建工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2. 设置参数
        factory.setHost("192.168.20.146");
        factory.setPort(5672);
        factory.setUsername("test");
        factory.setPassword("test");
        factory.setVirtualHost("/test");
        //3. 创建连接
        Connection connection = factory.newConnection();
        //4. 创建channel管道
        Channel channel = connection.createChannel();
        //5. 创建队列
        channel.queueDeclare("work_queues",true,false,false,null);
        //6.发送消息
        String body = "work_queues....";
        channel.basicPublish("","work_queues",null,body.getBytes());
        //7. 关闭连接
        channel.close();
        connection.close();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

• 尝试执行
  

• 代码编写消费步骤 (与简单模式消费者一模一样,只是有两个)
  1. 创建工厂
  2. 设置参数
  3. 新建连接
  4. 创建channel(管道)
  5. 创建队列 (生产者已经创建了,我们就不需要了!)
  6. 接受消息
  7. 关闭连接 (肯定不止接受一次,要持续执行着的)

package com.yang;

import com.rabbitmq.client.*;

import java.io.IOException;


public class Consumer_WorkQueues1 {
    public static void main(String[] args) throws Exception {
        //1. 创建工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2. 设置参数
        factory.setHost("192.168.20.146");
        factory.setPort(5672);
        factory.setUsername("test");
        factory.setPassword("test");
        factory.setVirtualHost("/test");
        //3. 创建连接
        Connection connection = factory.newConnection();
        //4. 创建管道
        Channel channel = connection.createChannel();
        //5. 创建队列
        //channel.queueDeclare("work_queues",true,false,false,null);
        //6.接受消息
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println(new String(body));
            }
        };
        channel.basicConsume("work_queues",true,consumer);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

• 尝试执行
  
• 另外一个直接复制粘贴即可
  
• 此时先将两个消费者开启,然后将生产者循环生成10次,看看输出
  
• 查看两个消费者
  1. 消费者1
     
  2. 消费者2
     
     可以看出两个消费者每次都是从生产者拿一个进行消费！

• 小结
  1. 在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系。
  2. Work Queues 对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。例如：短信服务部署多个，只需要有一个节点成功发送即可。

以上就是work queues模式的全部内容