消费者2
package com.rabbitmq7;import com.rabbitmq.client.CancelCallback;import com.rabbitmq.client.Channel;import com.rabbitmq.client.DeliverCallback;import com.rabbitmqUtils.RabbitmqUtils;import java.io.IOException;import java.util.concurrent.TimeoutException;/*** @author 天真热* @create 2022-02-08 14:52* @desc**/public class Consume02 {//接收消息public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//获取信道Channel channel = RabbitmqUtils.getChannel();//声明一个交换机channel.exchangeDeclare(RabbitmqUtils.RABBITMQ_EXCHANGE, "topic");//声明一个临时队列String queueName = channel.queueDeclare().getQueue();//绑定交换机和队列的channel.queueBind(queueName, RabbitmqUtils.RABBITMQ_EXCHANGE, "queue.*.*");//消费者成功消费时候的回调方法DeliverCallback deliverCallback = (consumerTag, message) -> {System.out.println("消费者成功消费时候的回调" + new String(message.getBody()));};//消费者取消消费的回调方法CancelCallback cancelCallback = consumerTag -> {System.out.println("消费者取消消费的回调方法");};//消费消息//1.队列名//2.消费成功后是否自动应答（是则默认成功 ， 否则需要）//3.消费者未成功消费时候的回调//4.消费者取消消费的回调方法channel.basicConsume(queueName, true, deliverCallback, cancelCallback);System.out.println("消费者|queue.*.*|等待中....");}}
效果
十一、代码实现-优先级队列
队列的消费顺序一般是先进先出 。但是在某些订单中业务中 ， 我们需要给vip用户后下单 ， 先出货的特殊权限 ， 这时候就需要用到优先级队列 。
原理 ， 在原来先进先出的逻辑上 ， 给队列备注优先级 ， 最后的顺序如下：
优先级高–>优先级低–>没有备注优先级
ps:优先级的范围为0-255
生产者
package com.rabbitmq12;import com.rabbitmq.client.AMQP;import com.rabbitmq.client.Channel;import com.rabbitmqUtils.RabbitmqUtils;import java.io.IOException;import java.util.HashMap;import java.util.Map;import java.util.concurrent.TimeoutException;/*** @author 天真热* @create 2022-02-08 14:52* @desc**/public class Product {//队列名称public static final String ORDER_QUEUE = "ORDER_QUEUE";//发消息public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//获取信道Channel channel = RabbitmqUtils.getChannel();//参数Map argument = new HashMap<>();argument.put("x-max-priority", 10);//设置优先级范围0-10 ， 官方允许值是0-255 。设置过大会浪费内存//生成队列,//不创建交换机 ， 走默认的交换机//1.名称//2.队列消息是否持久化（否：存内存 ， 是：存磁盘 。默认否）//3.队列是否只供一个消费者消费 ， 默认否//4.最后一个消费者断开连接后 ， 是否自动删除 。//5.其他参数channel.queueDeclare(ORDER_QUEUE, true, false, false, argument);//发消息String message = "this is QUEUE_P";//持续发送消息for (int i = 0; i < 10; i++) {//1.交换机 ， 简单版本不考虑 ， 直接空字符串即可（默认/无名交换机）//2.路由key ， 直接写队列名即可//3.参数 ， 忽略//4.消息体if (i == 5) {AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().priority(5).build();channel.basicPublish("", ORDER_QUEUE, properties, (message + i).getBytes());} else {channel.basicPublish("", ORDER_QUEUE, null, (message + i).getBytes());}}System.out.println("消息发送成功");}}
消费者
package com.rabbitmq12;import com.rabbitmq.client.CancelCallback;import com.rabbitmq.client.Channel;import com.rabbitmq.client.DeliverCallback;import com.rabbitmqUtils.RabbitmqUtils;import java.io.IOException;import java.util.concurrent.TimeoutException;/*** @author 天真热* @create 2022-02-08 15:34* @desc**/public class Consume {//队列名称public static final String ORDER_QUEUE = "ORDER_QUEUE";//发消息public static void main(String[] args) throws IOException, TimeoutException {//获取信道Channel channel = RabbitmqUtils.getChannel();//消费者未成功消费时候的回调方法DeliverCallback deliverCallback = (consumerTag, message) -> {System.out.println("消费者成功消费时候的回调" + new String(message.getBody()));};//消费者取消消费的回调方法CancelCallback cancelCallback = consumerTag -> {System.out.println("消费者取消消费的回调方法");};//消费消息//1.队列名//2.消费成功后是否自动应答//3.消费者成功消费时候的回调//4.消费者取消消费的回调方法channel.basicConsume(ORDER_QUEUE, true, deliverCallback, cancelCallback);}}