6.1 启动RabbitMQ

在虚拟机中执行命令systemctl start rabbitmq-server
启动之后查看下状态systemctl status rabbitmq-server

6.2 消息生产者

/**
 * 消息在手动应答时是不丢失的、放回队列中重新消费
 */
public class Task2 {

    //队列名称
    public static final String TASK_QUEUE_NAME="ack_queue";

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

        //声明队列
        boolean durable=true;   //需要让QUEUE进行持久化
        channel.queueDeclare(TASK_QUEUE_NAME,durable,false,false,null);
//        channel.queueDeclare(TASK_QUEUE_NAME,false,false,false,null);
        //从控制台中输入信息
        Scanner scanner=new Scanner(System.in);
        while(scanner.hasNext()){
            String message = scanner.next();
            channel.basicPublish("",TASK_QUEUE_NAME,null,message.getBytes("UTF-8"));
            System.out.println("生产者发出消息:"+message);
        }
    }
}

6.3 消费者01

package com.atguigu.rabbitmq.three;

import com.atguigu.rabbitmq.utils.RabbitMqUtils;
import com.atguigu.rabbitmq.utils.SleepUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

/**
 * 消息在手动应答时是不丢失的、放回队列中重新消费
 */
public class Work03 {

    public static final String TASK_QUEUE_NAME="ack_queue";

    //接收消息
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        System.out.println("C1等待接收消息处理时间较短");

        DeliverCallback deliverCallback=(consumerTag, message) -> {
            //沉睡1秒
            SleepUtils.sleep(1);

            System.out.println("接收到的消息:"+new String(message.getBody(),"UTF-8"));
            //手动应答
            /**
             * 参数1：消息的标记 tag
             * 参数2：是否批量应答 false：不批量应答信道中的消息 true:批量应答
             */
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);

        };

        //采用手动应答
        boolean autoAck=false;
        channel.basicConsume(TASK_QUEUE_NAME,autoAck,deliverCallback,(consumerTag -> {
            System.out.println(consumerTag+"消费者取消消费接口回调逻辑");
        }));
    }
}

6.4 消费者02

package com.atguigu.rabbitmq.three;

import com.atguigu.rabbitmq.utils.RabbitMqUtils;
import com.atguigu.rabbitmq.utils.SleepUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

/**
 * 消息在手动应答时是不丢失的、放回队列中重新消费
 */
public class Work04 {

    public static final String TASK_QUEUE_NAME="ack_queue";

    //接收消息
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        System.out.println("C2等待接收消息处理时间较长");

        DeliverCallback deliverCallback=(consumerTag, message) -> {
            //沉睡1秒
            SleepUtils.sleep(30);

            System.out.println("接收到的消息:"+new String(message.getBody(),"UTF-8"));
            //手动应答
            /**
             * 参数1：消息的标记 tag
             * 参数2：是否批量应答 false：不批量应答信道中的消息 true:批量应答
             */
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);

        };
        //采用手动应答
        boolean autoAck=false;
        channel.basicConsume(TASK_QUEUE_NAME,autoAck,deliverCallback,(consumerTag -> {
            System.out.println(consumerTag+"消费者取消消费接口回调逻辑");
        }));
    }
}

6.5 手动应答效果演示

我们在生产者中发送四条消息，观察消费者C1和C2的处理过程，其中
我们设置消费者C1每处理一条消息休眠1秒
消费者C2每处理一条消息休眠30秒
生产者先发送四条消息

观察消费者C1和C2(默认使用的是轮询分发)
消费者C1处理了两条

消费者C2,由于消费者C2每30秒才能接收一条消息，所以这里还看不到它处理完成的响应。

这个时候观察队列中的消息,从下图可看出还有两条消息没有被接收，这是由于C2接收消息的速度太慢(我们故意设置的慢)