RabbitMq-Work queues工作队列模式(二)

1、Work queues 工作队列模式概念:

Work Queues 与入门程序的 简单模式 相比,多了一个或一些消费端,多个消费端共同消费同一个队列中的消息。

工作队列模式:在同一个队列中可以有多个消费者,消费者之间对于消息的接收是竞争关系。
RabbitMq-Work queues工作队列模式(二)

Work Queues 与入门程序的 简单模式 的代码是几乎一样的;可以完全复制,并复制多一个消费者进行多个消费者同时消费消息的测试。

2、Work queues 工作队列模式 代码步骤:

RabbitMq-Work queues工作队列模式(二)

1)生产者:发送30个消息

package com.study.rabbitmq.work;

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

//work模式 发送消息

public class Producer {
    static final String QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws Exception {
// 1. 创建连接工厂(设置RabbitMQ的连接参数);
        ConnectionFactory connectionFactory= new ConnectionFactory();
        //主机;默认localhost
        connectionFactory.setHost("localhost");
        //连接端口;默认5672
        connectionFactory.setPort(5672);
        //虚拟主机;默认/
        connectionFactory.setVirtualHost("/");
        //用户名;默认guest
        connectionFactory.setUsername("guest");
        //密码;默认guest
        connectionFactory.setPassword("guest");
//   2. 创建连接
        Connection connection = connectionFactory.newConnection();
//   3. 创建频道;
        Channel channel = connection.createChannel();
//   4. 声明队列;
        /**
         * 参数1:队列名称
         * 参数2:是否定义持久化队列(消息会持久化保存在服务器上)
         * 参数3:是否独占本连接
         * 参数4:是否在不使用的时候队列自动删除
         * 参数5:其它参数
         */
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        for(int i=1;i<=30;i++){
            //5. 发送消息(改动);
            String message = "你好!work 模式---"+i;
            /**
             * 参数1:交换机名称;如果没有则指定空字符串(表示使用默认的交换机)
             * 参数2:路由key,简单模式中可以使用队列名称
             * 参数3:消息其它属性
             * 参数4:消息内容
             */
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            System.out.println("已发送消息:" + message);

        }
//        6. 关闭资源
        channel.close();
        connection.close();
    }

}

2)消费者:创建两个消费者监听同一个队列,查看两个消费者的接收消息是否存在重复。

package com.study.rabbitmq.work;
import com.rabbitmq.client.*;
import com.study.rabbitmq.util.ConnectionUtil;

import java.io.IOException;


/**
 * work模式;消费者接收消息
 */
public class Consumer1 {
    public static void main(String[] args) throws Exception {
//        1. 创建连接工厂;
//        2. 创建连接;(抽取一个获取连接的工具类)
        Connection connection = ConnectionUtil.getConnection();
//        3. 创建频道;
        final Channel channel = connection.createChannel();
//        4. 声明队列;
        /**
         * 参数1:队列名称
         * 参数2:是否定义持久化队列(消息会持久化保存在服务器上)
         * 参数3:是否独占本连接
         * 参数4:是否在不使用的时候队列自动删除
         * 参数5:其它参数
         */
        channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
        //每次可以预取多少个消息
        channel.basicQos(1);
//        5. 创建消费者(接收消息并处理消息);
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //路由key
                System.out.println("路由key为:" + envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为:" + envelope.getExchange());
                //消息id
                System.out.println("消息id为:" + envelope.getDeliveryTag());
                //接收到的消息
                System.out.println("消费者1-----接收到的消息为:" + new String(body, "utf-8"));
                try {
                    Thread.sleep(1000);
                    //确认消息
                    /*
                    参数1:消息id
                    参数2:是否确认,false表示只有当前这条消息被处理
                     */
                    channel.basicAck(envelope.getDeliveryTag(),false);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
//        6. 监听队列   (需要持续监听队列消息,所以不要关闭资源)
        /**
         * 参数1:队列名
         * 参数2:是否要自动确认;设置为true表示消息接收到自动向MQ回复接收到了,MQ则会将消息从队列中删除;
         * 如果设置为false则需要手动确认
         * 参数3:消费者
         */
        channel.basicConsume(Producer.QUEUE_NAME,true,defaultConsumer);
        //不关闭资源,应该一直监听消息
        // channel.close();
        // connection.close();


    }

}
  1. 先启动消费者,再启动生产者 结果:发现一个消息只能被一个消费者接收,其它消费者是不能接收到同一条消息的
    RabbitMq-Work queues工作队列模式(二)

3、工作队列模式小结

工作队列模式:一个消息只能被一个消费者接收,其它消费者是不能接收到同一条消息的。

应用场景:对于 任务过重或任务较多情况使用工作队列,可以提高任务处理的速度

添加对同一个队列的消费者来提高任务处理能力。

上一篇:RabbitMQ(四)——work queues工作模式


下一篇:python-select / tcp通信