> 只需要在消费者端，添加Qos能力以及更改为手动ack即可让消费者，根据自己的能力去消费指定的消息，而不是默认情况下由RabbitMQ平均分配了，生产者不变，正常发布消息到默认的exchange

>  消费者指定Qoa和手动ack

生产者

package com.qf.mq2302.work;import com.qf.mq2302.utils.MQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;public class Send {public static final String QUEUE_NAME="work";public static void main(String[] args) throws Exception {//1.获取连接对象Connection conn = MQUtils.getConnection();//2. 创建一个channel对象,对于MQ的大部分操作，都定义在了channel对象上Channel channel = conn.createChannel();//3.声明了一个队列/*** queue – the name of the queue* durable – true代表创建的队列是持久化的（当mq重启后，该对立依然存在）* exclusive – 该队列是不是排他的 (该对立是否只能由当前创建该队列的连接使用)* autoDelete – 该队列是否可以被mq服务器自动删除* arguments – 队列的其他参数，可以为null*/channel.queueDeclare(QUEUE_NAME, false, false, false, null);String message = "Hello doubleasdasda!";//生产者如何发送消息，使用下面的方法即可/*** exchange – 交换机的名字 ,如果是空串，说明是把消息发给了默认交换机* routingKey – 路由的key,当发送消息给默认交换机时，routingkey代表队列的名字* other properties - 消息的其他属性，可以为null* body – 消息的内容，注意，要是有 字节数组*/for (int i = 0; i < 21; i++) {channel.basicPublish("", QUEUE_NAME, null, (message+i).getBytes());}System.out.println(" [x] Sent '" + message + "'");//关闭资源channel.close();conn.close();}
}

消费者一

package com.qf.mq2302.work;import com.qf.mq2302.utils.MQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;import java.io.IOException;public class Recv {private  final  static  String QUEUE_NAME="work";public static void main(String[] args) throws Exception {//1.获取连接对象Connection conn = MQUtils.getConnection();//2. 创建一个channel对象,对于MQ的大部分操作，都定义在了channel对象上Channel channel = conn.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);//3.该消费者收到消息之后的处理逻辑，写在DeliverCallback对象中DeliverCallback deliverCallback =new DeliverCallback() {@Overridepublic void handle(String consumerTag, Delivery message) throws IOException {//从Delivery对象中可以获取到生产者，发送的消息的字节数组byte[] body = message.getBody();String msg = new String(body, "utf-8");try {Thread.sleep(400);} catch (InterruptedException e) {e.printStackTrace();}//在这里写消费者的业务逻辑,例如,发送邮件System.out.println("消费者01:"+msg);//手动ack//从message对象中取long deliveryTag = message.getEnvelope().getDeliveryTag();/*** 第一个参数：消息编号* 第二个参数: false,代表只确认这一个消息*/channel.basicAck(deliveryTag,false);}};//设置该消费者，每次只能从mq中获取一条消息channel.basicQos(1);//4.让当前消费者开始消费（QUEUE_NAME）队列中的消息/***把消费者的确认模式，设置为 手动 ack**/channel.basicConsume(QUEUE_NAME,false,deliverCallback,consumerTag -> {});}}

消费者二

package com.qf.mq2302.work;import com.qf.mq2302.utils.MQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;import java.io.IOException;public class Recv02 {private  final  static  String QUEUE_NAME="work";public static void main(String[] args) throws Exception {//1.获取连接对象Connection conn = MQUtils.getConnection();//2. 创建一个channel对象,对于MQ的大部分操作，都定义在了channel对象上Channel channel = conn.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);//3.该消费者收到消息之后的处理逻辑，写在DeliverCallback对象中DeliverCallback deliverCallback =new DeliverCallback() {@Overridepublic void handle(String consumerTag, Delivery message) throws IOException {//从Delivery对象中可以获取到生产者，发送的消息的字节数组byte[] body = message.getBody();String msg = new String(body, "utf-8");try {Thread.sleep(200);} catch (InterruptedException e) {e.printStackTrace();}//在这里写消费者的业务逻辑,例如,发送邮件System.out.println("消费者02:"+msg);long deliveryTag = message.getEnvelope().getDeliveryTag();channel.basicAck(deliveryTag,false);}};//注意：这个是可以存三个，而不是一次发三个channel.basicQos(3);//4.让当前消费者开始消费（QUEUE_NAME）队列中的消息/*** queue – the name of the queue* autoAck – true 代表当前消费者是不是自动确认模式。true代表自动确认。* deliverCallback – 当有消息发送给该消费者时，消费者如何处理消息的逻辑* cancelCallback – 当消费者被取消掉时，如果要执行代码，写到这里*/channel.basicConsume(QUEUE_NAME,false,deliverCallback,consumerTag -> {});}}