上海网站设计工作室,深圳网站设计服务,贵州省建设厅官方网站,做网站需要用到的语言 只需要在消费者端#xff0c;添加Qos能力以及更改为手动ack即可让消费者#xff0c;根据自己的能力去消费指定的消息#xff0c;而不是默认情况下由RabbitMQ平均分配了#xff0c;生产者不变#xff0c;正常发布消息到默认的exchange 消费者指定Qoa和手动ack … 只需要在消费者端添加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_NAMEwork;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, (messagei).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_NAMEwork;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_NAMEwork;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 - {});}}
