RabbitMQ之Work Queues

Work Queues

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

轮询分发消息

在多个线程接收消息的时候，默认为轮询的方式接收消息。下面创建一个小的demo来瞅瞅

首先呢，每次都创建channel十分的麻烦，我们可以像创建数据库驱动一样创建一个工具类

package com.bestrookie.rabbitmq.utils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 * @author bestrookie
 * @version 1.0
 * @date 2021/12/25 20:07
 */
public class RabbitUtils {
    public static Channel getChannel() throws IOException, TimeoutException {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("47.102.118.110");
    factory.setUsername("rookie");
    factory.setPassword("123456");
    Connection connection = factory.newConnection();
    return connection.createChannel();
}
}

然后创建一个work线程

package com.bestrookie.rabbitmq.workers;

import com.bestrookie.rabbitmq.utils.RabbitUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 * @author bestrookie
 * @version 1.0
 * @date 2021/12/25 20:21
 */
public class Work01 {
    public static String QUEUE_NAME = "hello";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitUtils.getChannel();
        //消息的接收
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("消息被接收："+new String(message.getBody()));
        };
        CancelCallback cancelCallback = (consumerTag)->{
            System.out.println(consumerTag+" 消息接收失败，取消消息");
        };
        System.out.println("C等待消费");
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}

然后idea可以将一段代码重复开启

然后勾选

然后修改一下打印的名称，就可以创建一个新的线程

然后我们创建一个消息发送者就叫boss吧

package com.bestrookie.rabbitmq.workers;
import com.bestrookie.rabbitmq.utils.RabbitUtils;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;

/**
 * @author bestrookie
 * @version 1.0
 * @date 2021/12/26 14:25
 */
public class Boss {
    private static final String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception{
        Channel channel = RabbitUtils.getChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        Scanner scanner = new Scanner(System.in);

        while (scanner.hasNextLine()){
            System.out.println("请输入消息：");
            String msg = scanner.next();
            channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
            System.out.println("发送消息完成-->"+msg);
        }
    }
}

然后运行这个消息发送者。

通过结果不难看出轮询的方式接收消息。

消息应答

消费者完成一个任务可能需要一段时间，如果其中一个消费者处理一个长的任务，在还没有执行完这个任务时突然挂掉了，会发生什么情况呢？RabbitMQ一旦向消费者传递了一条消息，便立即将该消息标记为删除。在这种情况下，我们就会丢失正在处理的这个消息，以及后续发送给该消费者的消息，因为这个消费者挂掉了。

为了保证消息在发送过程中不丢失，引入了消息应答机制，消息应答就是：消费者在就收并且处理该消息之后，告诉RabbitMQ它已经处理了，RabbitMQ可以把该消息删除了

自动应答：

消息发送后立即被认为已经传送成功，这种模式需要在高吞吐量和数据传输安全性方面做权衡，因为这这种模式如果在接收到之前消费者那边出现链接或者channel关闭，那么消息就丢失了，当然另一方面这种模式消费者那边可以传递过载的消息，没有对传递的消息数量进行限制，当然这样有可能使得消费者这边由于接收太多还来不及处理的消息，导致这些消息的积压，最终使得内存耗尽，最终这些消费者线程被操作系统杀死，所以这种模式仅适合在消费者可以高效并以某种速率能够处理这些消息的情况下使用

手动应答

1
2
3

channel.basicAck//(用于肯定) RabbitMQ已知道该消息并且成功的处理消息，可以丢弃了
channel.basicNack//(用于否定确认)
channel.basicReject//(用于否定) 与basicNack相比少一个参数 不处理该消息，直接丢弃

消息重新应答

如果消费者由于某些原因失去连接（其通道关闭，连接已关闭或者TCP连接丢失），导致消息未发送ACK确认，RabbitMQ将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。

实战代码

首先创建一个生产者

package com.bestrookie.rabbitmq.three;
import com.bestrookie.rabbitmq.utils.RabbitUtils;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;
/**
 * @author bestrookie
 * @version 1.0
 * @date 2021/12/26 21:03
 */
public class Product {
    public static final String TASK_QUEUE_NAME = "ack_queue";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitUtils.getChannel();

        channel.queueDeclare(TASK_QUEUE_NAME,false,false,false,null);
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()){
            String msg = scanner.next();
            channel.basicPublish("",TASK_QUEUE_NAME,null,msg.getBytes(StandardCharsets.UTF_8));
            System.out.println("消息发送成功："+msg);
        }

    }
}

然后我们分别创建两个消费者，两者的区别就是线程的睡眠时间不同，一个睡眠时间长，一个睡眠时间短，我们可以在线程睡眠的时间内断开线程的运行，模拟真实场景中的连接断开。以此来观察结果

package com.bestrookie.rabbitmq.three;

import com.bestrookie.rabbitmq.utils.RabbitUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.util.concurrent.TimeUnit;

/**
 * @author bestrookie
 * @version 1.0
 * @date 2021/12/26 21:03
 */
public class CustomerOne {
    public static final String TASK_QUEUE_NAME = "ack_queue";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitUtils.getChannel();
        System.out.println("C1等待时间短");
        DeliverCallback deliverCallback = (consumerTag,msg)->{
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("接收消息："+new String(msg.getBody()));
            //手动应答
            channel.basicAck(msg.getEnvelope().getDeliveryTag(),false);
        };

        CancelCallback cancelCallback = (consumerTag)->{
            System.out.println("消费信息被取消");
        };
        //采用手动应答
        channel.basicConsume(TASK_QUEUE_NAME, false,deliverCallback,cancelCallback);
    }
}

线程二

package com.bestrookie.rabbitmq.three;

import com.bestrookie.rabbitmq.utils.RabbitUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.util.concurrent.TimeUnit;

/**
 * @author bestrookie
 * @version 1.0
 * @date 2021/12/26 21:31
 */
public class CustomerTwo {
    public static final String TASK_QUEUE_NAME = "ack_queue";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitUtils.getChannel();
        System.out.println("C2等待时间长");
        DeliverCallback deliverCallback = (consumerTag, msg)->{
            try {
                TimeUnit.SECONDS.sleep(30);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("接收消息："+new String(msg.getBody()));
            //手动应答
            channel.basicAck(msg.getEnvelope().getDeliveryTag(),false);
        };

        CancelCallback cancelCallback = (consumerTag)->{
            System.out.println("消费信息被取消");
        };
        //采用手动应答 若果不设置的话默认为 false
        channel.basicConsume(TASK_QUEUE_NAME, false,deliverCallback,cancelCallback);
    }
}

RabbitMQ持久化

持久化大家都很熟悉，一般情况下RabbitMQ会把消息存到内存当中，但是当Rabbit服务突然宕掉，这种情况消息就会丢失。所以为了保证消息不丢失我们需要进行持久化处理：我们需要将队列和消息都标记为持久化。

队列持久化

我们一开始创建的队列都是非持久化的，RabbitMQ如果重启的话，该队列就会被删除，如果需要实现持久化就需要在声明队列的时候吧durable参数设置为true

但是注意一点，之前声明的队列不是持久化的，需要将原来的队列删除，或者重新创建一个持久化的队列，不然就会出现错误。

持久化成功，我们就会看到

消息持久化

消息持久化就需要在消息发送的时候进行设置，在发送消息的第三个参数，就代表发送消息的设置MessageProperties.PERSISTENT_TEXT_PLAIN就代表了消息的持久化。

将消息标记为持久化并不能完全保证不会丢失消息。尽管它告诉RabbitMQ将消息保存到磁盘，但是这里依然存在当消息刚准备存储在磁盘但是还没有存储完成，还是存在一个间隔点。此时并没有真正的写入磁盘。持久性保证并不强，但是对于我们的简单任务而言，这已经绰绰有余了。如果需要更有力的持久化，就需要发布确认。

不公平分发

刚开始我们学习RabbitMQ分发消息采用的是轮询分发，但是在某种场景下这种的策略并不是最优选择，比如，有两个消费者再处理任务，其中一个消费者的处理速度非常快，另一个处理却非常慢，这个时候我们如果采用轮询的方式分发，就会浪费掉多余的性你那个，这种分配的方式是不符合这个场景的。

我们可以通过参数channel.basicQos来设置。

1
2
3

//采用手动应答
//接收方式设置 0轮询 1不公平分发 其他预取值
channel.basicQos(1);

预取值

消息的发送就是异步发送的，所以在任何时候，channel上肯定不止只有一个消息，所以在消费接收之前就存在一个消息缓冲区，这个预取值在本质上就是限制这个消息缓冲区中消息的数量。

这个设置的方式和是否公平分发是一样的都是通过参数channel.basicQos来设置。