发送消息类
package com.rabbitmq10;import lombok.extern.slf4j.Slf4j;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.PathVariable;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;/*** @author 天真热* @create 2022-02-10 15:03* @desc**/@Slf4j@RestController@RequestMapping("/delayed")public class SendDelayMessageController {@Autowiredprivate RabbitTemplate rabbitTemplate;@GetMapping("/sendExpireMsg/{message}/{ttlTime}")public void sendExpireMsg(@PathVariable String message, @PathVariable Integer ttlTime) {log.info("发送定时消息");rabbitTemplate.convertAndSend(DelayConfig.DELAY_EXCHANGE, DelayConfig.DELAY_ROUNTING_KEY, "消息来自定时消息：" + message, msg -> {//发消息的时候 ， 延迟时长msg.getMessageProperties().setDelay(ttlTime);return msg;});}}
测试：依次执行以下两个地址 ， 可以发现延迟时间短的会先执行 ， 不需要排队
:8090///10000/10000
:8090///500/500
十五、代码实现-发布确认高级
原理图
配置文件
spring:rabbitmq:host: 192.168.248.10port: 5672username: adminpassword: admin#交换机确认接口publisher-confirms: true#新版本：spring.rabbitmq.publisher-confirm-type=correlated#路由回退消息给生产者publisher-returns: true
发送消息类
package com.rabbitmq11;import com.rabbitmq10.DelayConfig;import lombok.extern.slf4j.Slf4j;import org.springframework.amqp.rabbit.connection.CorrelationData;import org.springframework.amqp.rabbit.core.CorrelationDataPostProcessor;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.PathVariable;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;/*** @author 天真热* @create 2022-02-10 15:03* @desc**/@Slf4j@RestController@RequestMapping("/confirm")public class SendConfirmMessageController {@Autowiredprivate RabbitTemplate rabbitTemplate;@GetMapping("/sendMsgToBadExchange/{message}")public void sendMsgToBadExchange(@PathVariable String message) {//类 ， 可以在队列接收消息的时候接受CorrelationData correlationData = http://www.kingceram.com/post/new CorrelationData("1");//发送消息rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE + "bad", ConfirmConfig.CONFIRM_ROUNTING_KEY, "消息来自定时消息：" + message, correlationData);log.info("发送消息");}@GetMapping("/sendMsgToBadRounting/{message}")public void sendMsgToBadRounting(@PathVariable String message) {//类 ， 可以在队列接收消息的时候接受CorrelationData correlationData = http://www.kingceram.com/post/new CorrelationData("1");//发送消息rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE, ConfirmConfig.CONFIRM_ROUNTING_KEY + "bad", "消息来自定时消息：" + message, correlationData);log.info("发送消息");}@GetMapping("/sendMsg/{message}")public void sendMsg(@PathVariable String message) {//类 ， 可以在队列接收消息的时候接受CorrelationData correlationData = http://www.kingceram.com/post/new CorrelationData("1");//发送消息rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE, ConfirmConfig.CONFIRM_ROUNTING_KEY , "消息来自定时消息：" + message, correlationData);log.info("发送消息");}}
配置类
package com.rabbitmq11;import org.springframework.amqp.core.*;import org.springframework.beans.factory.annotation.Qualifier;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import java.util.HashMap;import java.util.Map;/*** 发布确认** @author 天真热* @create 2022-02-10 10:09* @desc**/@Configurationpublic class ConfirmConfig {//队列public static final String CONFIRM_QUEUE = "CONFIRM_QUEUE";//交换机public static final String CONFIRM_EXCHANGE = "CONFIRM_EXCHANGE";//routingKeypublic static final String CONFIRM_ROUNTING_KEY = "CONFIRM_ROUNTING_KEY";//备份交换机public static final String BACKUP_EXCHANGE = "BACKUP_EXCHANGE";//备份队列public static final String BACKUP_QUEUE = "BACKUP_QUEUE";//告警队列public static final String WARNING_QUEUE = "WARNING_QUEUE";//声明交换机//这里因为用的是之前的交换机 ， 所以需要删除原先的交换机才能生效@Beanpublic DirectExchange confirmEchange() {return (DirectExchange) ExchangeBuilder.directExchange(CONFIRM_EXCHANGE).durable(true).withArgument("alternate-exchange", BACKUP_EXCHANGE).build();}//声明备份交换机@Beanpublic FanoutExchange backupEchange() {return new FanoutExchange(BACKUP_EXCHANGE);}//声明队列@Beanpublic Queue confirmQueue() {//创建队列return new Queue(CONFIRM_QUEUE);}//声明备份队列@Beanpublic Queue backupQueue() {//创建队列return new Queue(BACKUP_QUEUE);}//声明报警队列@Beanpublic Queue warningQueue() {//创建队列return new Queue(WARNING_QUEUE);}//绑定队列@Beanpublic Binding confirmBindingQueue(@Qualifier("confirmQueue") Queue confirmQueue, @Qualifier("confirmEchange") DirectExchange confirmEchange) {return BindingBuilder.bind(confirmQueue).to(confirmEchange).with(CONFIRM_ROUNTING_KEY);}//绑定备份队列@Beanpublic Binding backupBindingQueue(@Qualifier("backupQueue") Queue backupQueue, @Qualifier("backupEchange") FanoutExchange backupEchange) {return BindingBuilder.bind(backupQueue).to(backupEchange);}//绑定报警队列@Beanpublic Binding warningBindingQueue(@Qualifier("warningQueue") Queue warningQueue, @Qualifier("backupEchange") FanoutExchange backupEchange) {return BindingBuilder.bind(warningQueue).to(backupEchange);}}