RabbitMQ——消息手动应答、队列/消息持久化、不公平分发、预取值的概念理解及应用举例

/**
  * 第一个参数：消息标记 tag
  * 第二个参数：是否对消息进行批量应答
  */
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);

package com.szh.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;
/**
 *
 */
public class RabbitMqUtils {
    public static Channel getChannel() {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.40.130");
        factory.setPort(5672);
        factory.setUsername("root");
        factory.setPassword("root");
        Connection connection = null;
        Channel channel = null;
        try {
            connection = factory.newConnection();
            channel = connection.createChannel();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
        return channel;
    }
}

package com.szh.rabbitmq.utils;
/**
 *
 */
public class SleepUtils {
    public static void sleep(int second) {
        try {
            Thread.sleep(1000*second);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

package com.szh.rabbitmq.second;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;
import com.szh.rabbitmq.utils.RabbitMqUtils;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
/**
 * 消息在手动应答时不会丢失，而是放回队列中重新消费
 */
public class ProducerDemo {
    public static final String QUEUE_NAME = "ack_queue";
    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMqUtils.getChannel();
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        //从控制台中输入信息
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()) {
            String message = scanner.next();
            //MessageProperties.PERSISTENT_TEXT_PLAIN 设置消息为持久化
            channel.basicPublish("",QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes(StandardCharsets.UTF_8));
            System.out.println("生产者发出消息：" + message);
        }
    }
}

package com.szh.rabbitmq.second;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;
import com.szh.rabbitmq.utils.RabbitMqUtils;
import com.szh.rabbitmq.utils.SleepUtils;
import java.io.IOException;
/**
 * 消息在手动应答时不会丢失，而是放回队列中重新消费
 */
public class ConsumerDemo1 {
    public static final String QUEUE_NAME = "ack_queue";
    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMqUtils.getChannel();
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        System.out.println("C1消费者等待接收消息，该处理时间较短，仅需1s....");
        DeliverCallback deliverCallback = (consumerTag,message) -> {
            String messageBody = new String(message.getBody());
            SleepUtils.sleep(1);
            System.out.println("C1消费者接收到消息：" + messageBody);
            /**
             * 第一个参数：消息标记 tag
             * 第二个参数：是否对消息进行批量应答
             */
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        };
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println(consumerTag + "消费者取消消费接口的回调逻辑....");
        };
        //设置不公平分发，prefetchCount的值为1
        //int prefetchCount = 1;
        //prefetchCount的值为其他值，则表示为预取值
        int prefetchCount = 2;
        channel.basicQos(prefetchCount);
        //采用手动应答
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME,autoAck,deliverCallback,cancelCallback);
    }
}

package com.szh.rabbitmq.second;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.szh.rabbitmq.utils.RabbitMqUtils;
import com.szh.rabbitmq.utils.SleepUtils;
import java.io.IOException;
/**
 * 消息在手动应答时不会丢失，而是放回队列中重新消费
 */
public class ConsumerDemo2 {
    public static final String QUEUE_NAME = "ack_queue";
    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMqUtils.getChannel();
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        System.out.println("C2消费者等待接收消息，该处理时间较长，大概需要10s....");
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String messageBody = new String(message.getBody());
            SleepUtils.sleep(10);
            System.out.println("C2消费者接收到消息：" + messageBody);
            /**
             * 第一个参数：消息标记 tag
             * 第二个参数：是否对消息进行批量应答
             */
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        };
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println(consumerTag + "消费者取消消费接口的回调逻辑....");
        };
        //设置不公平分发，prefetchCount的值为1
        //int prefetchCount = 1;
        //prefetchCount的值为其他值，则表示为预取值
        int prefetchCount = 5;
        channel.basicQos(prefetchCount);
        //采用手动应答
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME,autoAck,deliverCallback,cancelCallback);
    }
}