工作队列 - 图1

工作队列

工作队列又称为任务队列，其主要思想是避免立即执行资源密集型任务，而将这些任务分发给多个线程，在后台运行的工作中将分发的任务最后执行完毕。当我们有多个工作线程的时候，这些工作线程将一起处理这些任务。

轮询分发消息

在默认情况下，假设我们有三个消费者，也就是工作线程，那么这三个工作线程从队列中获取消息的能力（几率）是一样的，会依次从队列中获取消息并执行相应的任务。

抽取工具类

由于我们创建发送线程和工作线程，都需要获取RabbitMQ的连接，而这一部分是重复的
因此我们可以将这一部分抽取出来称为公共的工具部分

public class MQUtil {
    public static Channel getChannel() throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        // 获取连接
        Connection connection = connectionFactory.newConnection();
        // 获取发送信息的通道
        return connection.createChannel();
    }
}

启动两个工作线程

我们可以将某个代码重复运行两次来实现启动两个工作线程，但个人感觉不太优雅（
所以采用多线程的方式来启动两个工作线程

/**
 * Created By Intellij IDEA
 *
 * @author ssssheep
 * @package com.ctgu.sheep.two
 * @datetime 2022/9/16 星期五
 */
public class Worker1 {
    public static final String QUEUE_NAME = "hello";
    public static void main(String[] args) throws Exception {
        Channel channel = MQUtil.getChannel();
        for(int i = 1;i <= 3;i++) {
            WorkThread workThread = new WorkThread(i + "", channel);
            workThread.start();
        }
    }
}
class WorkThread extends Thread {
    public static final String QUEUE_NAME = "hello";
    public String name;
    public Channel channel;
    public WorkThread(String name, Channel channel) {
        this.name = name;
        this.channel = channel;
    }
    @Override
    public void run() {
        System.out.println("工作线程" + name + "等待接受消息...");
        try {
            channel.basicConsume(QUEUE_NAME, true, (consumerTag, message) -> {
                System.out.println("工作线程" + consumerTag + "接收到消息: " + new String(message.getBody()));
            }, consumerTag -> {
                System.out.println("工作线程：" + consumerTag + "取消消费的逻辑");
            });
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

启动一个发送线程

/**
 * Created By Intellij IDEA
 *
 * @author ssssheep
 * @package com.ctgu.sheep.two
 * @datetime 2022/9/16 星期五
 */
public class Producer {
    public static final String QUEUE_NAME = "hello";
    public static void main(String[] args) throws Exception {
        Channel channel = MQUtil.getChannel();
        /**
         * 1. 队列名称
         * 2. 是否持久化
         * 3. 是否独占队列
         * 4. 是否在消费完成后自动删除队列
         * 5. 队列其他参数
         */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        for(int i = 1;i <= 20;i++) {
            String message = "hello rabbitmq" + i;
            /**
             * 1. 交换机名称
             * 2. 队列名称
             * 3. 消息其他属性
             * 4. 消息内容
             */
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("生产者发送消息：" + message);
//            Thread.sleep(1500);
        }
    }
}

结果

我们一共启动了三个工作线程和一个发送线程，三个工作线程会依次获取到消息并打印在控制台中

消息应答

概念

同计算机网络中数据的传输一样，在消息队列中，我们同样需要对收到的数据进行ACK确认，从而告诉消息队列：消息已经被正常处理掉了。

消费者完成一个任务可能需要一段时间。如果其中的一个消费者处理一个长的任务时，中途因为某种原因宕机。假设RabbitMQ一旦向消费者传递一条信息后就将其标记为删除，那么在这种情况下，消息就会出现丢失的情况。
为了保证消息在发送过程中不会丢失，我们需要引入消息应答机制。消息应答就是：消费者在接收到消息并且处理消息之后，通知rabbitmq此条消息已经被处理，可以进行删除了。