RabbitMQ学习（三）：工作队列

2023-02-22 91

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务，而不得不等待它完成。相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务

一、工作队列原理

工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务，而不得不等待它完成。相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务

二、案例演示

在这个案例中我将启动两个工作线程、一个消息发送线程，看看这两个工作线程是如何工作的。

1、抽取工具类

public class RabbitMQUtils {
    public static Channel getChannel() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("47.242.xx.xx");    //填写自己的ip地址
        factory.setUsername("root");        //填写自己的用户名和密码
        factory.setPassword("root");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        return channel;
    }
}

2、工作线程代码准备

package com.shang.two;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.shang.utils.RabbitMQUtils;
/**
 * 这是一个工作线程（相当于之前的消费者）
 */
public class Worker01 {
    private static final String QUEUE_NAME="hello";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        DeliverCallback deliverCallback=(consumerTag, delivery)->{
            String receivedMessage = new String(delivery.getBody());
            System.out.println("接收到消息:"+receivedMessage);
        };
        CancelCallback cancelCallback=(consumerTag)->{
            System.out.println(consumerTag+"消费者取消消费接口回调逻辑");
        };
        System.out.println("C1 消费者启动等待消费......");
        /**
         * 消费者接受消息 channel.basicConsume()
         * 参数：
         * 1 消费哪个队列
         * 2 消费成功后是否要自动应答
         * 3 消费者未成功消费的回调
         * 4 消费者取录消费的回调
         */
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}

启动之后，开启多线程

为了区别两个工作线程，将工作线程的代码改为C2，再次启动

此时我们可以看到下方有两个线程在运行，一个是C1，一个是C2：

3、消息发送线程代码准备

public class Task01 {
    private static final String QUEUE_NAME="hello";
    public static void main(String[] args) throws Exception {
        try(Channel channel= RabbitMQUtils.getChannel();) {
            /**
             * 生成一个队列
             * 参数：
             * 1 队列名称
             * 2 队列里的消息是否持久化，默认情况消息存储在内存中
             * 3 该队列是否供多个消费者进行消费，是否进行消息共享
             * 4 是否自动删除
             * 5 其他参数
             */
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
            //从控制台当中接受信息
            Scanner scanner = new Scanner(System.in);
            while (scanner.hasNext()){
                String message = scanner.next();
                channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
                System.out.println("发送消息完成:"+message);
            }
        }
    } }