SpringBoot下RabbitMq實現定時任務
阿新 • • 發佈:2020-01-07
本文例項為大家分享了SpringBoot下RabbitMq實現定時任務,供大家參考,具體內容如下
定時任務場景:訂單下單15分鐘未付款自動關閉
延遲任務實現原理圖如下:
根據上圖看出我們需要兩個佇列(一是死信佇列,訊息在裡面度過TLL時間,二是處理佇列,訊息度過TLL時間後進入該佇列),兩個交換機和路由(一是用來將訊息送入死信佇列,二是將訊息從死信佇列送到處理佇列),但是交換機其實可以用同一個,也就是一個交換機搭配兩個路由的方式。
以下為程式碼實現過程:
//首先rabbitAdmin的配置 @Bean public RabbitAdmin rabbitAdmin(CachingConnectionFactory cachingConnectionFactory){ RabbitAdmin rabbitAdmin = new RabbitAdmin(cachingConnectionFactory); rabbitAdmin.setIgnoreDeclarationExceptions(true); return rabbitAdmin; }
然後是佇列交換機以及繫結的宣告(Queue,Exchange,Binding)
@Autowired RabbitAdmin rabbitAdmin; /** * 建立延遲佇列,該佇列內容不會唄監聽,TLL到期自動轉發 * @param queueName * @param delayExchange */ public void createDelayQueue(String queueName,String delayExchange,String delayRoutingKey){ Map<String,Object> map = new HashMap<>(); /**注意這兩個引數必須繫結**/ map.put("x-dead-letter-exchange",delayExchange); map.put("x-dead-letter-routing-key",delayRoutingKey); rabbitAdmin.declareQueue(new Queue(queueName,true,false,map)); } /** * 建立普通佇列 * @param queueName */ public void createQueue(String queueName){ rabbitAdmin.declareQueue(new Queue(queueName,new HashMap<>())); } /** * 建立延遲佇列得交換機,就是一個普通得交換機 * @param name */ public void createExchange(String name){ rabbitAdmin.declareExchange(new DirectExchange(name,false)); } /** * 繫結佇列 * @param queueName * @param delayExchange * @param routeKey */ public void bindQueue(String queueName,String routeKey){ rabbitAdmin.declareBinding(new Binding(queueName,Binding.DestinationType.QUEUE,delayExchange,routeKey,new HashMap<>())); } /** * 傳送訊息 * @param message * @param exchange * @param routeKey * @param tll 時間 單位毫秒 */ public void sendMessageWithTll(String message,String exchange,String routeKey,String tll){ rabbitAdmin.getRabbitTemplate().convertAndSend(exchange,message,new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setExpiration(tll); return message; } }); } @Override public void sendMessage(String message,String routeKey) { rabbitAdmin.getRabbitTemplate().convertAndSend(exchange,new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { return message; } }); }
進行測試
/** * 監聽訊息 * @param message * @return */ @RabbitListener(queues = "consumerQueue") public void getMessage(String message){ System.out.println("收到訊息getMessage:"+message +",time="+MyUtil.FormatDate(new Date())); }
以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支援我們。