SpringBoot下RabbitMq实现定时任务
本文实例为大家分享了SpringBoot下RabbitMq实现定时任务,供大家参考,具体内容如下
定时任务场景:订单下单15分钟未付款自动关闭
延迟任务实现原理图如下:
根据上图看出我们需要两个队列(一是死信队列,消息在里面度过TLL时间,二是处理队列,消息度过TLL时间后进入该队列),两个交换机和路由(一是用来将消息送入死信队列,二是将消息从死信队列送到处理队列),但是交换机其实可以用同一个,也就是一个交换机搭配两个路由的方式。
以下为代码实现过程:
//首先rabbitAdmin的配置 @Bean publicRabbitAdminrabbitAdmin(CachingConnectionFactorycachingConnectionFactory){ RabbitAdminrabbitAdmin=newRabbitAdmin(cachingConnectionFactory); rabbitAdmin.setIgnoreDeclarationExceptions(true); returnrabbitAdmin; }
然后是队列交换机以及绑定的声明(Queue,Exchange,Binding)
@Autowired RabbitAdminrabbitAdmin; /** *创建延迟队列,该队列内容不会呗监听,TLL到期自动转发 *@paramqueueName *@paramdelayExchange */ publicvoidcreateDelayQueue(StringqueueName,StringdelayExchange,StringdelayRoutingKey){ Mapmap=newHashMap<>(); /**注意这两个参数必须绑定**/ map.put("x-dead-letter-exchange",delayExchange); map.put("x-dead-letter-routing-key",delayRoutingKey); rabbitAdmin.declareQueue(newQueue(queueName,true,false,false,map)); } /** *创建普通队列 *@paramqueueName */ publicvoidcreateQueue(StringqueueName){ rabbitAdmin.declareQueue(newQueue(queueName,true,false,false,newHashMap<>())); } /** *创建延迟队列得交换机,就是一个普通得交换机 *@paramname */ publicvoidcreateExchange(Stringname){ rabbitAdmin.declareExchange(newDirectExchange(name,true,false)); } /** *绑定队列 *@paramqueueName *@paramdelayExchange *@paramrouteKey */ publicvoidbindQueue(StringqueueName,StringdelayExchange,StringrouteKey){ rabbitAdmin.declareBinding(newBinding(queueName,Binding.DestinationType.QUEUE,delayExchange,routeKey,newHashMap<>())); } /** *发送消息 *@parammessage *@paramexchange *@paramrouteKey *@paramtll时间单位毫秒 */ publicvoidsendMessageWithTll(Stringmessage,Stringexchange,StringrouteKey,Stringtll){ rabbitAdmin.getRabbitTemplate().convertAndSend(exchange,routeKey,message,newMessagePostProcessor(){ @Override publicMessagepostProcessMessage(Messagemessage)throwsAmqpException{ message.getMessageProperties().setExpiration(tll); returnmessage; } }); } @Override publicvoidsendMessage(Stringmessage,Stringexchange,StringrouteKey){ rabbitAdmin.getRabbitTemplate().convertAndSend(exchange,routeKey,message,newMessagePostProcessor(){ @Override publicMessagepostProcessMessage(Messagemessage)throwsAmqpException{ returnmessage; } }); }
进行测试
/** *监听消息 *@parammessage *@return */ @RabbitListener(queues="consumerQueue") publicvoidgetMessage(Stringmessage){ System.out.println("收到消息getMessage:"+message+",time="+MyUtil.FormatDate(newDate())); }
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持毛票票。
声明:本文内容来源于网络,版权归原作者所有,内容由互联网用户自发贡献自行上传,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任。如果您发现有涉嫌版权的内容,欢迎发送邮件至:czq8825#qq.com(发邮件时,请将#更换为@)进行举报,并提供相关证据,一经查实,本站将立刻删除涉嫌侵权内容。