- sms
package com.jsxs.service.ttl; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Service; /** * @Author Jsxs * @Date 2023/4/2 13:44 * @PackageName:com.jsxs.service.faout * @ClassName: SmsConsumer * @Description: TODO * @Version 1.0 */ @Service @RabbitListener(queues = {"sms.ttl.queue"}) // 这个客户端的队列是哪个? public class SmsConsumer { @RabbitHandler // 接收到的消息放在这 public void receiveMessage(String message){ System.out.println("sms接收到的信息是:->"+message); } }
package com.jsxs.service.ttl; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Service; /** * @Author Jsxs * @Date 2023/4/2 13:44 * @PackageName:com.jsxs.service.faout * @ClassName: WechatConsumer * @Description: TODO * @Version 1.0 */ @Service @RabbitListener(queues = {"wechat.ttl.queue"}) // 这个客户端的队列是哪个? public class WechatConsumer { @RabbitHandler // 接收到的消息放在这 public void receiveMessage(String message){ System.out.println("wechat接收到的信息是:->"+message); } }
我们启动生产者之后,会在消息队列中产生消息..
如果我们消费者没有在五秒内对这个消息进行接收,那么就会自动消失
2.过期时间TTL (消息)
(1).生产者
3.给消息设置过期时间 MessagePostProcessor messagePostProcessor = new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setExpiration("5000"); // 过期时间 message.getMessageProperties().setContentEncoding("UTF-8"); // 字符编码 return message; } };
package com.jsxs.service; import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.stereotype.Service; import javax.annotation.Resource; import java.util.UUID; /** * @Author Jsxs * @Date 2023/4/2 11:24 * @PackageName:com.jsxs.service * @ClassName: OrderService * @Description: TODO : 分别给对应的路由Key发送消息. * @Version 1.0 */ @Service public class OrderService { @Resource // 获取rabbitMQ的服务 private RabbitTemplate rabbitTemplate; /** * * @param userId * @param productID * @param num */ public void makeOrder(String userId,String productID,int num){ //1. 生成订单 String orderID = UUID.randomUUID().toString().replace("-",""); System.out.println("订单号已经生产成功-"+orderID); //2. 设置交换机名字和路由 String exchangeName="ttl_order_producer"; // 3.给消息设置过期时间 MessagePostProcessor messagePostProcessor = new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setExpiration("5000"); // 过期时间 message.getMessageProperties().setContentEncoding("UTF-8"); // 字符编码 return message; } }; //4. 发送消息 // 参数: (交换机、路由key或队列名、消息内容) rabbitTemplate.convertAndSend(exchangeName,"ttl","1",messagePostProcessor); rabbitTemplate.convertAndSend(exchangeName,"ttl","2",messagePostProcessor); rabbitTemplate.convertAndSend(exchangeName,"ttl","3",messagePostProcessor); rabbitTemplate.convertAndSend(exchangeName,"ttl","4",messagePostProcessor); } }
(2).配置文件
我们对前两个队列进行进行队列TTL和消息TTL两个的同时设置；对后两个队列仅仅设置消息TTL。 这四个队列的交换机是一致的。
package com.jsxs.config; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /** * @Author Jsxs * @Date 2023/4/3 16:16 * @PackageName:com.jsxs.config * @ClassName: TTLRabbitMQConfig * @Description: TODO * @Version 1.0 */ import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.Queue; import java.util.HashMap; @Configuration public class TTLRabbitMQConfig { // 1. 声明注册direct模式的交换机 @Bean public DirectExchange directExchange(){ // (交换机的名字、是否持久化。是否自动删除) return new DirectExchange("ttl_order_producer",true,false); } // 2. 声明队列: 以及过期时间 @Bean public Queue SmsQueue(){ HashMap<String, Object> args = new HashMap<>(); args.put("x-message-ttl",5000); return new Queue("sms.ttl.queue",true,false,false,args); } @Bean public Queue MessageQueue(){ HashMap<String, Object> args = new HashMap<>(); args.put("x-message-ttl",5000); return new Queue("message.ttl.queue",true,false,false,args); } // ---------------上面我们同时设置队列过期时间和消息过期时间-----下面我们设置仅消息过期时间 @Bean public Queue EmailQueue(){ return new Queue("email.message.ttl.queue",true,false,false); } @Bean public Queue WeChatQueue(){ return new Queue("wechat.message.ttl.queue",true,false,false); } // 3. 将队列与交换机进行绑定的操作 @Bean public Binding SmsBind(){ return BindingBuilder.bind(SmsQueue()).to(directExchange()).with("ttl"); } @Bean public Binding MessageBind(){ return BindingBuilder.bind(MessageQueue()).to(directExchange()).with("ttl"); } @Bean public Binding EmailBind(){ return BindingBuilder.bind(EmailQueue()).to(directExchange()).with("ttl"); } @Bean public Binding WechatBind(){ return BindingBuilder.bind(WeChatQueue()).to(directExchange()).with("ttl"); } }
(3).消费者
package com.jsxs.service.ttl; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Service; /** * @Author Jsxs * @Date 2023/4/2 13:45 * @PackageName:com.jsxs.service.faout * @ClassName: EmailConsumer * @Description: TODO * @Version 1.0 */ @Service @RabbitListener(queues = {"email.message.ttl.queue"}) // 这个客户端的队列是哪个? public class EmailConsumer { @RabbitHandler // 接收到的消息放在这 public void receiveMessage(String message){ System.out.println("email接收到的信息是:->"+message); } }
- message
package com.jsxs.service.ttl; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Service; /** * @Author Jsxs * @Date 2023/4/2 13:44 * @PackageName:com.jsxs.service.faout * @ClassName: MessageConsumer * @Description: TODO * @Version 1.0 */ @Service @RabbitListener(queues = {"message.ttl.queue"}) // 这个客户端的队列是哪个? public class MessageConsumer { @RabbitHandler // 接收到的消息放在这 public void receiveMessage(String message){ System.out.println("Message接收到的信息是:->"+message); } }
- sms
package com.jsxs.service.ttl; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Service; /** * @Author Jsxs * @Date 2023/4/2 13:44 * @PackageName:com.jsxs.service.faout * @ClassName: SmsConsumer * @Description: TODO * @Version 1.0 */ @Service @RabbitListener(queues = {"sms.ttl.queue"}) // 这个客户端的队列是哪个? public class SmsConsumer { @RabbitHandler // 接收到的消息放在这 public void receiveMessage(String message){ System.out.println("sms接收到的信息是:->"+message); } }
- wechate
package com.jsxs.service.ttl; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Service; /** * @Author Jsxs * @Date 2023/4/2 13:44 * @PackageName:com.jsxs.service.faout * @ClassName: WechatConsumer * @Description: TODO * @Version 1.0 */ @Service @RabbitListener(queues = {"wechat.message.ttl.queue"}) // 这个客户端的队列是哪个? public class WechatConsumer { @RabbitHandler // 接收到的消息放在这 public void receiveMessage(String message){ System.out.println("wechat接收到的信息是:->"+message); } }
