RabbitMQ 发布确认 交换机 死信队列 延迟队列

news/2024/4/23 19:37:49/

RabbitMQ

  • 发布确认
    • 开启发布确认的方法
    • 单个确认发布
    • 批量消息确认发布
    • 异步确认发布
      • 如何处理异步未确认消息
  • 交换机
    • 绑定
    • Fanout交换机
      • Fannout交换机(消费者)
      • Fannout交换机(生产者)
    • Direct exchage(直接交换机)
      • 生产者
      • 消费者
    • Topic交换机
      • Topic要求
      • Topic交换机(消费者)
      • Topic交换机(生产者)
  • 死信队列
    • 死信的来源
    • 死信实战
      • 死信实战(消费者1)
      • 死信实战(生产者)
      • 死信实战(消费者2)
  • 延迟队列
    • 延迟队列使用场景
    • 整合SpringBoot
      • 队列TTL
      • 基于死信存在的问题
        • Rabbitmq插件实现延迟队列

发布确认

1.设置要求队列必须持久化
2.设置要求队列中的消息
3.发布确认
什么是发布确认?
只有当消息完完整整的发送完成发布确认之后,消息才算在磁盘上保存好了,数据再怎么服务器开关都不会丢失

开启发布确认的方法

发布确认默认是没有开启的,如果要开启需要调用方法 confirmSelect,每当你要想使用发布确认,都需要在channel 上调用该方法
在这里插入图片描述

单个确认发布

这是一种简单的确认方式,它是一种同步确认发布的方式,也就是发布一个消息之后只有它被确认发布,后续的消息才能继续发布,waitForConfirmsOrDie(long)这个方法只有在消息被确认的时候才返回,如果在指定时间范围内这个消息没有被确认那么它将抛出异常。
这种确认方式有一个最大的缺点就是:发布速度特别的慢,因为如果没有确认发布的消息就会阻塞所有后续消息的发布,这种方式最多提供每秒不超过数百条发布消息的吞吐量。
当然对于某些应用程序来说这可能已经足够了。


//单个确认
public static void publishMessageIndividually() throws Exception{
Channel channel = RabbitMqUtils.getChannel();
//队列的声明
String queueName = UUID.randomUUID().toString();
channel.queueDeclare(queueName,true,false,false,null);
//开启发布确认
channel.confirmSelect();
//开始时间
long begin = System.currentTimeMillis();//大量的发消息
for (int i = 0; i < MESSAGE_COUNT; i++) {String message = i + "";channel.basicPublish("",queueName,null,message.getBytes());//单个消息就马上进行发布确认boolean flag = channel.waitForConfirms();if(flag){System.out.println("消息发送成功");}
}
//结束时间
long end = System.currentTimeMillis();
System.out.println("发布"+MESSAGE_COUNT+"个单独确认消息,耗时"+(end-begin+"ms"));
}

运行结果:
在这里插入图片描述

批量消息确认发布

上面我们讲到的单个确认发布消息特别慢,与其相比,先发布一批消息然后一起确认可以极大的提高吞吐量,当然这种方式的缺点就是:当发生故障导致发布出现问题时,不知道是哪个消息出现问题了,我们必须将整个批处理保存在内存中,以记录重要的信息而后重新发布消息。当然这种方案仍然是同步的,也一样阻塞消息的发布。

//批量发布确认
public static void publishMessageBatch() throws Exception{
Channel channel = RabbitMqUtils.getChannel();
//队列的声明
String queueName = UUID.randomUUID().toString();
channel.queueDeclare(queueName,true,false,false,null);
//开启发布确认
channel.confirmSelect();
//开始时间
long begin = System.currentTimeMillis();//批量确认消息大小
int batchSize = 100;//批量发送消息 批量发布确认
for (int i = 0; i < MESSAGE_COUNT; i++) {String message = i + "";channel.basicPublish("",queueName,null,message.getBytes());//判断达到100条消息的时候,批量确认一次if(i%batchSize==0){//发布确认channel.waitForConfirms();}
}//结束时间
long end = System.currentTimeMillis();
System.out.println("发布"+MESSAGE_COUNT+"个批量确认消息,耗时"+(end-begin+"ms"));
}

异步确认发布

异步确认虽然编程逻辑比上两个要复杂,但是性价比最高,无论是可靠性还是效率都没得说,他是利用回调函数来达到消息可靠性传递的,这个中问件也是通过函数回调来保证是否投递成功,下面就让我们来详细讲解异步确认是怎么实现的。
在这里插入图片描述

//异步发布确认
public static void publishMessageAsync() throws Exception{
Channel channel = RabbitMqUtils.getChannel();
//队列的声明
String queueName = UUID.randomUUID().toString();
channel.queueDeclare(queueName,true,false,false,null);
//开启发布确认
channel.confirmSelect();
//开始时间
long begin = System.currentTimeMillis();//消息确认成功,回调函数
ConfirmCallback ackCallback = (deliveryTag,multiple)->{System.out.println("确认的消息:"+deliveryTag);
};
//消息确认失败,回调函数
ConfirmCallback nackCallback = (deliveryTag,multiple)->{System.out.println("未确认的消息:"+deliveryTag);
};
//准备消息监听器 监听哪些消息成功了 哪些消息失败了
channel.addConfirmListener(ackCallback,nackCallback);//批量发送消息
for (int i = 0; i < MESSAGE_COUNT; i++) {String message = "消息"+i;channel.basicPublish("",queueName,null,message.getBytes());
}//结束时间
long end = System.currentTimeMillis();
System.out.println("发布"+MESSAGE_COUNT+"个确认消息,耗时"+(end-begin+"ms"));
}

如何处理异步未确认消息

最好的解决的解决方案就是把未确认的消息放到一个基于内存的能被发布线程访问的队列,比如说用ConcurrentLinkedQueue这个队列在confirm callbacks与发布线程之间进行消息的传递。

//异步发布确认
public static void publishMessageAsync() throws Exception{
Channel channel = RabbitMqUtils.getChannel();
//队列的声明
String queueName = UUID.randomUUID().toString();
channel.queueDeclare(queueName,true,false,false,null);
//开启发布确认
channel.confirmSelect();
//线程安全有序的一个哈希表适用于高并发的情况下
/** 1.轻松的将序号与消息进行关联* 2.轻忪批量删除条目只要给到序号3.支持高并发(多线程)*/
ConcurrentSkipListMap<Long,String> outstandingConfirms =
new ConcurrentSkipListMap<>();
//开始时间
long begin = System.currentTimeMillis();//消息确认成功,回调函数
ConfirmCallback ackCallback = (deliveryTag,multiple)->{if(multiple){ConcurrentNavigableMap<Long,String> confirmed =outstandingConfirms.headMap(deliveryTag);confirmed.clear();}else {outstandingConfirms.remove(deliveryTag);}//2.删除到已经确认的消息 剩下的就是未确认的消息System.out.println("确认的消息:"+deliveryTag);
};
//消息确认失败,回调函数
ConfirmCallback nackCallback = (deliveryTag,multiple)->{//打印一下未确认的消息都有哪些String message = outstandingConfirms.get(deliveryTag);System.out.println("未确认的消息:"+message+"::::"+deliveryTag);
};
//准备消息监听器 监听哪些消息成功了 哪些消息失败了
channel.addConfirmListener(ackCallback,nackCallback);//批量发送消息
for (int i = 0; i < MESSAGE_COUNT; i++) {String message = "消息"+i;channel.basicPublish("",queueName,null,message.getBytes());//1.此处记录下所有要发送的消息 消息的总和outstandingConfirms.put(channel.getNextPublishSeqNo(),message);
}//结束时间
long end = System.currentTimeMillis();
System.out.println("发布"+MESSAGE_COUNT+"个确认消息,耗时"+(end-begin+"ms"));
}

交换机

在这里插入图片描述
在上一节中,我们创建了一个工作队列。我们假设的是工作队列背后,每个任务都恰好交付给一个消费者(工作进程)。在这一部分中,我们将做一些完全不同的事情-我们将消息传达给多个消费者。这种模式称为发布/订阅"。
概念
RabbitMQ消息传递模型的核心思想是:生产者生产的消息从不会直接发送到队列。实际上,通常生产者甚至都不知道这些消息传递传递到了哪些队列中。
相反,生产者只能将消息发送到交换机(exchange),交换机工作的内容非常简单,一方面它接收来自生产者的消息,另一方面将它们推入队列。交换机必须确切知道如何处理收到的消息。是应该把这些消息放到特定队列还是说把他们到许多队列中还是说应该丢弃它们。这就的由交换机的类型来决定。

交换机类型
直接(direct),主题(topic),标题(headers),扇出(fanout)
临时队列
每当我们连接到Rabbit时,我们都需要一个全新的空队列,为此我们可以创建一个具有随机名称的队列,或者能让服务器为我们选择一个随机队列名称那就更好了。其次一旦我们断开了消费者的连接,队列将被自动删除。
在这里插入图片描述

绑定

什么是 bingding呢,binding其实是exchange和queue之间的桥梁,它告诉我们exchange和那个队列进行了绑定关系。比如说下面这张图告诉我们的就是×与Q1和Q2进行了绑定
在这里插入图片描述

Fanout交换机

Fanout这种类型非常简单。正如从名称中猜到的那样,它是将接收到的所有消息广播到它知道的所有队列中。系统中默认有些exchange类型

Fannout交换机(消费者)

public class ReceiveLogs02 {//交换机的名称public static final String EXCHANGE_NAME="logs";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();//产生一个交换机channel.exchangeDeclare(EXCHANGE_NAME,"fanout");//声明一个队列 临时队列//当消费者斯开与队列的连接的时候队列就自动鹏除String queueName = channel.queueDeclare().getQueue();//绑定交换机与channel.queueBind(queueName,EXCHANGE_NAME,"");System.out.println("ReceiveLogs01等待接收消息,把接收到消息打印在屏幕上......");//接收消息DeliverCallback deliverCallback = (consumerTag,message)->{System.out.println("控制台打印接收到的消息:"+new String(message.getBody(),"UTF-8"));};//消费者取消消息时回调接口channel.basicConsume(queueName,true,deliverCallback,consumerTag->{});}
}
public class ReceiveLogs02 {//交换机的名称public static final String EXCHANGE_NAME="logs";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();//产生一个交换机channel.exchangeDeclare(EXCHANGE_NAME,"fanout");//声明一个队列 临时队列//当消费者斯开与队列的连接的时候队列就自动鹏除String queueName = channel.queueDeclare().getQueue();//绑定交换机与channel.queueBind(queueName,EXCHANGE_NAME,"");System.out.println("ReceiveLogs02等待接收消息,把接收到消息打印在屏幕上......");//接收消息DeliverCallback deliverCallback = (consumerTag,message)->{System.out.println("控制台打印接收到的消息:"+new String(message.getBody(),"UTF-8"));};//消费者取消消息时回调接口channel.basicConsume(queueName,true,deliverCallback,consumerTag->{});}
}

Fannout交换机(生产者)

//发消息  交换机
public class EmitLog {//交换机的名称public static final String EXCHANGE_NAME = "logs";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();//channel.exchangeDeclare(EXCHANGE_NAME,"fauout");Scanner scanner = new Scanner(System.in);while (scanner.hasNext()){String message = scanner.next();channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes("UTF-8"));System.out.println("生产者发出消息:"+message);}}
}

运行结果:

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

Direct exchage(直接交换机)

上一节中的我们的日志系统将所有消息广播给所有消费者,对此我们想做一些改变,例如我们希望将日志消息写入磁盘的程序仅接收严重错误(erros),而不存储哪些警告(warning)或信息(info)日志消息避免浪费磁盘空间。Fanout这种交换类型并不能给我们带来很大的灵活性-它只能进行无意识的广播,在这里我们将使用direct这种类型来进行替换,这种类型的工作方式是,消息只去到它绑定的routingKey队列中去。
在这里插入图片描述

生产者

public class DirectLogs {//交换机的名称public static final String EXCHANGE_NAME = "direct_logs";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();//channel.exchangeDeclare(EXCHANGE_NAME,"fauout");Scanner scanner = new Scanner(System.in);while (scanner.hasNext()){String message = scanner.next();channel.basicPublish(EXCHANGE_NAME,"info",null,message.getBytes("UTF-8"));System.out.println("生产者发出消息:"+message);}}
}

消费者


public class ReceiveLogDirect01 {public static final String EXCHANGE_NAME = "direct_logs";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();//声明一个交换机channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);//声明一个队列channel.queueDeclare("console", false,false,false,null);channel.queueBind("console",EXCHANGE_NAME,"info");//接收消息DeliverCallback deliverCallback = (consumerTag, message)->{System.out.println("ReceiveLogDirect01控制台打印接收到的消息:"+new String(message.getBody(),"UTF-8"));};//消费者取消消息时回调接口channel.basicConsume("console",true,deliverCallback,consumerTag->{});}
}
public class ReceiveLogDirect02 {public static final String EXCHANGE_NAME = "direct_logs";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();//声明一个交换机channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);//声明一个队列channel.queueDeclare("disk", false,false,false,null);channel.queueBind("disk",EXCHANGE_NAME,"error");//接收消息DeliverCallback deliverCallback = (consumerTag, message)->{System.out.println("ReceiveLogDirect02控制台打印接收到的消息:"+new String(message.getBody(),"UTF-8"));};//消费者取消消息时回调接口channel.basicConsume("disk",true,deliverCallback,consumerTag->{});}
}

运行结果:

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

Topic交换机

在这里插入图片描述
在上面这张图中,我们可以看到X绑定了两个队列,绑定类型是direct。队列Q1绑定键为orange,队列Q2绑定键有两个:一个绑定键为black,另一个绑定键为green.
在这种绑定情况下,生产者发布消息到exchange上,绑定键为orange的消息会被发布到队列Q1。绑定键为 blackgreen.和的消息会被发布到队列Q2,其他消息类型的消息将被丢弃。

Topic要求

发送到类型是topic交换机的消息的routing_key不能随意写,必须满足一定的要求,它必须是一个单词列表,以点号分隔开。这些单词可以是任意单词,比如说:“stock.usd.nyse” ,“nyse.xvmw”,
“quick.orange.rabbit”.这种类型的。当然这个单词列表最多不能超过255个字节。
在这个规则列表中,其中有两个替换符是大家需要注意的
*(星号)可以代替一个单词
#|(井号)可以替代零个或多个单词

Topic交换机(消费者)

public class ReceiveLogsTopic01 {//交换机的名称public static final String EXCHANGE_NAME = "topic_logs";//接收消息public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();//声明交换机channel.exchangeDeclare(EXCHANGE_NAME,"topic");//声明队列String queueName = "Q1";channel.queueDeclare(queueName,false,false,false,null);channel.queueBind(queueName,EXCHANGE_NAME,"*.orange.*");System.out.println("接收消息");//接收消息DeliverCallback deliverCallback = (consumerTag, message)->{System.out.println("控制台打印接收到的消息:"+new String(message.getBody(),"UTF-8"));};channel.basicConsume(queueName,true,deliverCallback,consumeTag->{});}
}
public class ReceiveLogsTopic02 {//交换机的名称public static final String EXCHANGE_NAME = "topic_logs";//接收消息public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();//声明交换机channel.exchangeDeclare(EXCHANGE_NAME,"topic");//声明队列String queueName = "Q2";channel.queueDeclare(queueName,false,false,false,null);channel.queueBind(queueName,EXCHANGE_NAME,"*.*.rabbit");channel.queueBind(queueName,EXCHANGE_NAME,"lazy.#");System.out.println("接收消息");//接收消息DeliverCallback deliverCallback = (consumerTag, message)->{System.out.println("控制台打印接收到的消息:"+new String(message.getBody(),"UTF-8"));};channel.basicConsume(queueName,true,deliverCallback,consumeTag->{});}
}

Topic交换机(生产者)

public class EmitLogTopic {//交换机的名称public static final String EXCHANGE_NAME = "topic_logs";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();Map<String,String> bindingKeyMap = new HashMap<>();bindingKeyMap.put("quick.orange.rabbit","1");bindingKeyMap.put("lazy.orange.elephant","2");bindingKeyMap.put("quick.orange.fox","3");bindingKeyMap.put("lazy.brown.fox","4");bindingKeyMap.put("lazy.pink.rabbit","5");bindingKeyMap.put("quick.brown.fox","6");bindingKeyMap.put("quick.orange.male.rabbit","7");bindingKeyMap.put("lazy.orange.male.rabbit","8");for (Map.Entry<String, String> bindingKeyEntry : bindingKeyMap.entrySet()) {String routingKey = bindingKeyEntry.getKey();String message = bindingKeyEntry.getValue();channel.basicPublish(EXCHANGE_NAME,routingKey,null,message.getBytes());System.out.println("生产者发出消息"+message);}}
}

运行结果:

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

死信队列

某些时候由于特定的原因导致queue中的某些信息无法被消费,但这样的消息如果没有后续的处理,就变成死信,有死信自然就有了死信队列
应用场景:为了保证订单业务的消息数据不丢失,需要使用到RabbitMQ的死信队列机制,当消息消费发生异常时,将消息投入死信队列中.还有比如说:用户在商城下单成功并点击去支付后在指定时间未支付时自动失效

死信的来源

消息TTL过期
队列达到最大长度(队列满了,无法再添加数据到mq.中)
消息被拒绝(basic.reject或 basic.nack)并且requeue=false

死信实战

死信实战(消费者1)

public class Consumer01 {//普通交换机的名称public static final String NORMAL_EXCHANGE = "normal_exchange";//死信交换机的名称public static final String DEAD_EXCHANGE = "dead_exchange";//普通队列的名称public static final String NORMAL_QUEUE = "normal_queue";//死信队列的名称public static final String DEAD_QUEUE = "dead_queue";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();//声明死信和普通交换机,类型为directchannel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);//声明死信队列和普通队列Map<String ,Object> arguments = new HashMap<>();//过期时间//正常队列设置死信交换机//过期时间 10s-10000ms//arguments.put("x-message-ttl",1000000);arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);//设置死信RoutineKeyarguments.put("x-dead-letter-routing-key","lisi");channel.queueDeclare(NORMAL_QUEUE,false,false,false,null);channel.queueDeclare(DEAD_QUEUE,false,false,false,null);//绑定普通的交换机与队列channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");//绑定死信的交换机与队列channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");DeliverCallback deliverCallback = (consumerTag,message)->{System.out.println("Consumer01接收的消息是"+new String(message.getBody(),"UTF-8"));};channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,consumerTag->{});}
}

死信实战(生产者)

public class Producer {//普通交换机的名称public static final String NORMAL_EXCHANGE = "normal_exchange";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();//死信消息,设置TTL时间AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();for(int i=1;i<11;i++){String message = "info"+i;channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null,message.getBytes());}}
}

死信实战(消费者2)

public class Consumer02 {//死信队列的名称public static final String DEAD_QUEUE = "dead_queue";public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();DeliverCallback deliverCallback = (consumerTag,message)->{System.out.println("Consumer02接收的消息是"+new String(message.getBody(),"UTF-8"));};channel.basicConsume(DEAD_QUEUE,true,deliverCallback,consumerTag->{});}
}

延迟队列

延时队列,队列内部是有序的,最重要的特性就体现在它的延时属性上,延时队列中的元素是希望在指定时间到了以后或之前取出和处理,简单来说,延时队列就是用来存放需要在指定时间被处理的元素的队列。

延迟队列使用场景

1.订单在十分钟之内未支付则自动取消
⒉.新创建的店铺,如果在十天内都没有上传过商品,则自动发送消息提醒
3.用户注册成功后,如果三天内没有登{陆则进行短信提醒
4.用户发起退款,如果三天内没有得到处理则通知相关运营人员
5.预定会议后,需要在预定的时间点前十分钟通知各个与会人员参加会议

整合SpringBoot

第一步:导入依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.7.10</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>com.yc.springbootrabbitmq</groupId><artifactId>demo</artifactId><version>0.0.1-SNAPSHOT</version><name>demo</name><description>Demo project for Spring Boot</description><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><!--RabbitMQ依赖--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.47</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><!--swagger--><dependency><groupId>io.springfox</groupId><artifactId>springfox-swagger2</artifactId><version>2.9.2</version></dependency><dependency><groupId>io.springfox</groupId><artifactId>springfox-swagger-ui</artifactId><version>2.9.2</version></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

第二步:导入Swagger配置类

@Configuration
@EnableSwagger2
public class SwaggerConfig {@Beanpublic Docket webApiConfig(){return new Docket(DocumentationType.SWAGGER_2).groupName("webApi").apiInfo(webApiInfo()).select().build();}private ApiInfo webApiInfo(){return new ApiInfoBuilder().title("rabbitmq接口文档").description("文档描述了rabbitmq微服务接口定义").version("1.0").contact(new Contact("enjoy6288","http://atguigu.com","2439317465@qq.com")).build();}
}

队列TTL

第一步:配置文件类代码

//Ttl队列 配置文件类代码
@Configuration
public class TtlQueueConfig {//普通交换机的名称public static final String X_EXCHANGE = "X";//死信交换机的名称public static final String Y_DEAD_LETTER_EXCHANGE = "Y";//普通队列的名称public static final String QUEUE_A="QA";public static final String QUEUE_B="QB";//死信队列的名称public static final String DEAD_LETTER_QUEUE="QD";//普通队列的名称public static final String QUEUE_C="QC";@Bean("queueC")public Queue queueC(){Map<String,Object> arguments = new HashMap<>(3);//设置死信交换机arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);//设置死信RoutineKeyarguments.put("x-dead-letter-routing-key","YD");return QueueBuilder.durable(QUEUE_C).withArguments(arguments).build();}//声明xExchange   别名@Bean("xExchange")public DirectExchange xExchange(){return new DirectExchange(X_EXCHANGE);}//声明yExchange   别名@Bean("yExchange")public DirectExchange yExchange(){return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);}//声明普通队列 TTL 为 10s@Bean("queueA")public Queue queueA(){Map<String,Object> arguments = new HashMap<>(3);//设置死信交换机arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);//设置死信RoutingKeyarguments.put("x-dead-letter-routing-key","YD");//设置TTLarguments.put("x-message-ttl",10000);return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();}//声明普通队列 TTL 为 10s@Bean("queueB")public Queue queueB(){Map<String,Object> arguments = new HashMap<>(3);//设置死信交换机arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);//设置死信RoutingKeyarguments.put("x-dead-letter-routing-key","YD");//设置TTLarguments.put("x-message-ttl",40000);return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();}//死信队列@Bean("queueD")public Queue queueD(){return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();}//绑定@Beanpublic Binding queueABingdingX(@Qualifier("queueA") Queue queueA,@Qualifier("xExchange") DirectExchange xExchange){return BindingBuilder.bind(queueA).to(xExchange).with("XA");}//绑定@Beanpublic Binding queueBBingdingY(@Qualifier("queueB") Queue queueB,@Qualifier("xExchange") DirectExchange xExchange){return BindingBuilder.bind(queueB).to(xExchange).with("XB");}//绑定@Beanpublic Binding queueDBingdingY(@Qualifier("queueD") Queue queueD,@Qualifier("yExchange") DirectExchange yExchange){return BindingBuilder.bind(queueD).to(yExchange).with("YD");}@Beanpublic Binding queueCBindingX(@Qualifier("queueC") Queue queueC,@Qualifier("xExchange") DirectExchange xExchange){return BindingBuilder.bind(queueC).to(xExchange).with("XC");}
}

第二步:配置生产者

@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {@Autowiredprivate RabbitTemplate rabbitTemplate;//开始发消息@GetMapping("/sendMsg/{message}")public void sendMsg(@PathVariable String message){log.info("当前时间:{},发送一条信息给两个TTL队列:{}",new Date().toString(),message);rabbitTemplate.convertAndSend("X","XA","消息来自ttl为10s的队列:"+message);rabbitTemplate.convertAndSend("X","XB","消息来自ttl为40s的队列:"+message);}//开始发消息 消息  TTL@GetMapping("/sendExpirationMsg/{message}/{ttlTime}")public void sendMsg(@PathVariable String message,@PathVariable String ttlTime){log.info("当前时间:{},发送一条时长{}毫秒TTL信息给队列QC:{}",new Date().toString(),ttlTime,message);rabbitTemplate.convertAndSend("X","XC",message,msg->{//发送消息的时候   延迟时长msg.getMessageProperties().setExpiration(ttlTime);return msg;});}
}

第三步:配置消费者

@Slf4j
@Component
public class DeadLetterQueueConsumer {//接收消息@RabbitListener(queues = "QD")public void receiveD(Message message, Channel channel) throws Exception{String msg = new String(message.getBody());log.info("当前时间:{},收到死信队列的消息:{}",new Date().toString(),msg);}
}

基于死信存在的问题

一旦发出两条以上消息,看起来似乎没什么问题,但是在最开始的时候,就介绍过如果使用在消息属性上设置TTL的方式,消息可能并不会按时“死亡“,因为RabbitMQ只会检查第一个消息是否过期,如果过期则丢到死信队列,如果第一个消息的延时时长很长,而第二个消息的延时时长很短,第二个消息并不会优先得到执行。

Rabbitmq插件实现延迟队列

安装延时队列插件
在官网上下载https://www.rabbitmg.com/community-plugins.html,下载
rabbitmq_delayed_message_exchange插件,然后解压放置到 RabbitMQ的插件目录。进入RabbitMQ.的安装目录下的plgins.目录,执行下面命令让该插件生效,然后重启RabbitMQ
/usr/lib/rabbitmq/lib/rabbitmq _server-3.8.8/plugins
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
之前:

在这里插入图片描述
现在:
在这里插入图片描述
示例代码:

第一步:创建配置类

@Configuration
public class DelayedQueueConfig {//交换机public static final String DELAYED_EXCHANGE_NAME="delayed.exchange";//队列public static final String DELAYED_QUEUE_NAME="delayed.queue";//routingKeypublic static final String DELAYED_ROUTING_KEY="delayed.routingkey";@Beanpublic Queue delayedQueue(){return new Queue(DELAYED_QUEUE_NAME);}//声明交换机@Beanpublic CustomExchange delayedExchange(){Map<String,Object> arguments = new HashMap<>();arguments.put("x-delayed-type","direct");return new CustomExchange(DELAYED_EXCHANGE_NAME,"x-delayed-message",true,false);}//绑定public Binding delayedQueueBindingDelayedExchange(@Qualifier("delayedQueue") Queue delayedQueue,@Qualifier("delayedExchange") CustomExchange delayedExchange){return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();}
}

第二步:创建生产者

//开始发消息 基于插件的   消息  及   延迟的时间
@GetMapping("/sendDelayMsg/{message}/{delayTime}")
public void sendMsg(@PathVariable String message, @PathVariable Integer delayTime) {log.info("当前时间:{},发送一条时长{}毫秒TTL信息给队列QC:{}", new Date().toString(), delayTime, message);rabbitTemplate.convertAndSend(DelayedQueueConfig.DELAYED_EXCHANGE_NAME, DelayedQueueConfig.DELAYED_ROUTING_KEY, message, msg -> {//发送消息的时候   延迟时长msg.getMessageProperties().setDelay(delayTime);return msg;});
}

第三步:创建消费者

//基于插件的延迟消息
@Slf4j
@Component
public class DelayQueueConsumer {//监听消息@RabbitListener(queues = DelayedQueueConfig.DELAYED_QUEUE_NAME)public void receiveDelayQueue(Message message){String msg = new String(message.getBody());log.info("当前时间:{},收到延迟队列的信息:{}",new Date().toString(),msg);}
}

运行结果:

在这里插入图片描述


http://www.ppmy.cn/news/47598.html

相关文章

在ROS2中使用奥比中光(ORBBEC)的AstraPro深度相机

0.效果演示 1.下载SDK 到官网下载OpenNI2_SDK 记得是下载这个OpenNI2_SDK,而不是下载那个Orbbec_SDK. 2.拷贝至自定义目录 拷贝到你的ubuntu的一个文件夹中&#xff0c;并解压得到 ros2_astra_camera 文件夹 然后新建一个ros2_ws文件夹&#xff0c;再在ros2_ws文件夹中新建…

Oracle函数记录

一、各个函数介绍 1.OVER(PARTITION BY… ORDER BY…)--开窗函数 1.开窗函数用于为行定义一个窗口&#xff08;这里的窗口是指运算将要操作的行的集合&#xff09;&#xff0c;它对一组 值进行操作&#xff0c;不需要使用GROUP BY子句对数据进行分组&#xff0c;能够在同一…

用机器学习sklearn+opencv-python过计算型验证码

目录 生成计算型验证码图片 用opencv-python处理图片 制作训练数据集 训练模型 识别验证码 总结与提高 源码下载 在本节我们将使用sklearn和opencv-python这两个库过掉计算型验证码&#xff0c;图片示例如下。 生成计算型验证码图片 要识别验证码&#xff0c;我们就需要…

The 2021 China Collegiate Programming Contest (Harbin) D. Math master

题目链接 题解 2 63 2^{63} 263大概是 1 0 19 10^{19} 1019那么一共有19位需要讨论, 每一个位数各有保留和删除两种状态, 全部状态就是 2 18 2^{18} 218种 因为每一位数都有两种状态, 使用二进制数表示每个状态, 正好能全部表示, 在二进制位数下1表示保留, 0表示删除(反过来也…

GFD233A 3BHE022294R0103

GFD233A 3BHE022294R0103 ABB KUC321AE PLC模块 HIEE300698R0001 KU C321 AE01 ABB KUC711 3BHB004661R0001 高压变频模块 KUC711AE ABB KUC755AE105 3BHB005243R0105 驱动控制系统模块 KUC755 ABB KUC755AE106 3BH005243R006 控制系统模块 KU C755 AE 106 ABB LDGRB-01 3BSE01…

自动插入匹配与标题相关的百度图片或者搜狗图片软件-批量插入txt文档-Chatgpt批量写文章配图神器

1、我们用《Chatgpt 3.5-turbo软件》批量生成txt文档&#xff0c;但是这样txt文档里不带图片&#xff0c;直接发布到网站上&#xff0c;光有文字没有图片&#xff0c;效果也不是很理想&#xff0c;就需要一款配图软件。 2、该软件根据txt标题自动匹配百度图片或者搜狗图片里的…

236. 二叉树的最近公共祖先 ——【Leetcode每日一题】

236. 二叉树的最近公共祖先 给定一个二叉树, 找到该树中两个指定节点的最近公共祖先。 百度百科中最近公共祖先的定义为&#xff1a;“对于有根树 T 的两个节点 p、q&#xff0c;最近公共祖先表示为一个节点 x&#xff0c;满足 x 是 p、q 的祖先且 x 的深度尽可能大&#xff…

[网络工程师]-网络规划与设计-逻辑网络设计(四)

7、网络管理设计 为了成功地管理一个网络,要考虑如何选择合适的网络管理系统并集成到管理策略中。 7.1网络管理策略 制定网络管理策略的工作是十分重要的,因为网络管理策略详细描述了应从每台设备上采集哪些信息以及如何分析这些信息。在策略制定过程中,可以从前面介绍的协…

追梦之旅【数据结构篇】——详解小白如何使用C语言实现堆数据结构

详解小白如何使用C语言实现堆数据结构 “痛”撕堆排序~&#x1f60e; 前言&#x1f64c;什么是堆&#xff1f;堆的概念及结构 堆的性质&#xff1a;堆的实现堆向下调整算法画图分析&#xff1a;堆向下调整算法源代码分享&#xff1a;向下调整建小堆向下调整建大堆 堆向上调整算…

Redis(02)Hash--附有示例

文章目录 redis-HashHDELHEXISTSHGETHGETALLHINCRBYHINCRBYFLOATHKEYSHLENHMGETHMSETHRANDFIELDHSCANHSETHSETNXHSTRLENHVALS redis-Hash Redis中哈希表是一种非常实用的数据结构&#xff0c;它能够存储和管理具有结构化数据的业务数据&#xff0c;同时也可以方便地获取哈希表…

哪个牌子手持洗拖一机好?热门洗地机盘点

在家居清洁中&#xff0c;越来越多的家庭选择了通过智能清洁家电来完成地面的清洁工作&#xff0c;其中洗地机时最受大家青睐的清洁工具&#xff0c;它不仅可以提高我们的清洁效率&#xff0c;还可以减轻清洁时的劳动强度。不过&#xff0c;不同品牌之间的产品的差距也是大有不…

使用Vite虚拟模块功能重写多语言和多皮肤插件

背景 为了处理在Vite和Vue3场景下&#xff0c;打包部署后实现多语言和皮肤的更换和修改的功能&#xff0c;我开发了两个vite插件。插件在构建结束前&#xff0c;把资源文件转换和复制到dist中&#xff0c;再使用HTTP请求读取资源&#xff0c;解决了多语言包和多皮肤包扩展的问…

基于Python长时间序列遥感数据处理及在全球变化、物候提取、植被变绿与固碳分析、生物量估算与趋势分析

植被是陆地生态系统中最重要的组分之一&#xff0c;也是对气候变化最敏感的组分&#xff0c;其在全球变化过程中起着重要作用&#xff0c;能够指示自然环境中的大气、水、土壤等成分的变化&#xff0c;其年际和季节性变化可以作为地球气候变化的重要指标。此外&#xff0c;由于…

5G入海, 智慧海洋从此“联通”!

湛江&#xff0c;位于中国大陆的最南端&#xff0c;是一座向海而生的城市。近年来&#xff0c;中国联通在湛江打造智慧渔船管理平台&#xff0c;通过专用系统监控平台、岸基子系统、船载子系统三大平台实现九大核心功能&#xff0c;守卫1200多公里的大陆海岸线&#xff0c;赋能…

JAVA WEB 开发资源汇总

1. 狂神说的JAVA路线&#xff0c;简洁明了易上手&#xff0c;这是个大的路线&#xff0c;需要坚持住天天学。但有时候前面讲的是基础&#xff0c;后面有造好的轮子可以用&#xff0c;看时间吧&#xff0c;来得及把轮子怎么造的也好好看看。、 狂神用了一个不错的数据图在线网站…

《C++模板》(初阶)零基础讲解

本文主要介绍C的模板&#xff0c;包括函数模板和类模板 文章目录 为什么要有模板1、函数模板1.1 函数模板概念1.1 函数模板格式1.3 函数模板的原理1.4 函数模板的实例化1.5 模板参数的匹配原则 2、类模板2.1 类模板的定义格式2.2 类模板的实例化 为什么要有模板 就拿我们写的交…

动物养殖虚拟仿真之生猪屠宰VR教学系统

生猪屠宰是一个复杂而危险的工作&#xff0c;需要有严格的操作规程和丰富的经验。但是传统的生猪屠宰培训存在一些问题&#xff0c;例如成本高、难以模拟真实场景等。 为了解决这些问题&#xff0c;VR技术被应用到生猪屠宰培训中&#xff0c;广州华锐互动由此开发了生猪屠宰VR…

5.1、阻塞/非阻塞、同步/异步(网络IO)

5.1、阻塞/非阻塞、同步/异步&#xff08;网络IO&#xff09; 1.阻塞/非阻塞、同步/异步(网络IO)①典型的一次IO的两个阶段是什么&#xff1f; 2.日志系统①基础知识②整体概述③本文内容④单例模式1.经典的线程安全懒汉模式2.局部静态变量之线程安全懒汉模式 ⑤饿汉模式⑥条件…

YOLOv8 更换主干网络之 ShuffleNetv2

《ShuffleNet V2: Practical Guidelines for Efficient CNN Architecture Design》 目前,神经网络架构设计多以计算复杂度的间接度量——FLOPs为指导。然而,直接的度量,如速度,也取决于其他因素,如内存访问成本和平台特性。因此,这项工作建议评估目标平台上的直接度量,而…

打印流,Properties类

打印流只有输出流&#xff0c;没有输入流 package com.hspedu.printstream;import java.io.IOException; import java.io.PrintStream;/*** author 韩顺平* version 1.0* 演示PrintStream &#xff08;字节打印流/输出流&#xff09;*/ public class PrintStream_ {public stat…