如何使用RabbitMq来实现死信队列

首先什么是MQ消息中间件

  全称MessageQueue，主要是⽤于程序和程序直接通信，异步+解耦 
       使⽤场景： 
          核⼼应⽤ 
          解耦：订单系统-》物流系统 
          异步：⽤户注册-》发送邮件，初始化信息 
          削峰：秒杀、⽇志处理 
          跨平台 、多语⾔ 
          分布式事务、最终⼀致性 
          RPC调⽤上下游对接，数据源变动->通知下属
  总而言之 可以对各种场景进行异步校验，应用广泛可以保持链路的完整性

交换机类型
Direct Exchange 定向将⼀个队列绑定到交换机上，要求该消息与⼀个特定的路由键完全匹配例⼦：如果⼀个队列绑定到该交换机上要求路由键 “aabb”，则只有被标记为“aabb”的消息才被转发，不会转发aabb.cc，也不会转发gg.aabb，只会转发 aabb 处理路由健
Fanout Exchange ⼴播只需要简单的将队列绑定到交换机上，⼀个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。很像⼦⽹⼴播，每台⼦⽹内的主机都获得了⼀份复制的消息 Fanout交换机转发消息是最快的，⽤于发布订阅，⼴播形式，中⽂是扇形不处理路由健
Topic Exchange 通配符主题交换机是⼀种发布/订阅的模式，结合了直连交换机与扇形交换机的特点将路由键和某模式进⾏匹配。此时队列需要绑定要⼀个模式上符号“#”匹配⼀个或多个词，符号“*”匹配不多不少⼀个词例⼦：因此“abc.#”能够匹配到“abc.def.ghi”，但是 “abc.*” 只会匹配到“abc.def”。
Headers Exchanges（少⽤）根据发送的消息内容中的headers属性进⾏匹配, 在绑定Queue与Exchange时指定⼀组键值对当消息发送到RabbitMQ时会取到该消息的headers 与Exchange绑定时指定的键值对进⾏匹配；如果完全匹配则消息会路由到该队列，否则不会路由到该队列不处理路由键
什么是TTL time to live 消息存活时间如果消息在存活时间内未被消费，则会别清除 RabbitMQ⽀持两种ttl设置单独消息进⾏配置ttl 整个队列进⾏配置ttl（居多）
什么是rabbitmq的死信队列没有被及时消费的消息存放的队列
什么是rabbitmq的死信交换机 Dead Letter Exchange（死信交换机，缩写：DLX）当消息成为死信后，会被重新发送到另⼀个交换机，这个交换机就是DLX死信交换机。

下面就是实现的大概流程：    ![](https://i-blog.csdnimg.cn/direct/03ae7cf2329440178fe78886925715d2.png)

例子：

RabbitMq不自带死信队列，那么我们创建好交换机，延迟队列设置过期时间为15秒，对延迟队列和交换机以及死信队列（普通队列）和交换机进行绑定，同时设置交换机为Topic主题交换机。随后消费者通过RabbitMqListener进行监听，同时使用ACK进行标志的确认。


@Configuration
@Data
public class RabbitMqConfig {
    /**
     * 交换机
     */
    @Value("${mqconfig.coupon_event_exchange}")
    private String eventExchange;
    /**
     * 第⼀个队列延迟队列，
     */
    @Value("${mqconfig.coupon_release_delay_queue}")
    private String couponReleaseDelayQueue;
    /**
     * 第⼀个队列的路由key
     * 进⼊队列的路由key
     */
    @Value("${mqconfig.coupon_release_delay_routing_key}")
    private String couponReleaseDelayRoutingKey;
    /**
     * 第⼆个队列，被监听恢复库存的队列
     */
    @Value("${mqconfig.coupon_release_queue}")
    private String couponReleaseQueue;
    /**
     * 第⼆个队列的路由key
     * <p>
     * 即进⼊死信队列的路由key
     */
    @Value("${mqconfig.coupon_release_routing_key}")
    private String couponReleaseRoutingKey;
    /**
     * 过期时间
     */
    @Value("${mqconfig.ttl}")
    private Integer ttl;
    /**
     * 消息转换器
     *
     * @return
     */
    @Bean
    public MessageConverter messageConvertor() {
        return new Jackson2JsonMessageConverter();
    }
    /**
     * 交换机
     *
     * @return
     */
    @Bean
    public TopicExchange exchange() {
        return new TopicExchange(eventExchange, true, false);
    }
    /**
     * 创建延迟队列
     *
     * @return
     */
    @Bean
    public Queue couponReleaseDelayQueue() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-message-ttl", ttl);
        args.put("x-dead-letter-exchange", eventExchange);
        args.put("x-dead-letter-routing-key", couponReleaseRoutingKey);
        return new Queue(couponReleaseDelayQueue, true, false, false, args);
    }
    /**
     * 创建死信队列   普通队列  用于监听
     *
     * @return
     */
    @Bean
    public Queue couponReleaseQueue() {
        return new Queue(couponReleaseQueue, true, false, false);
    }
    /**
     * 绑定延迟队列和交换机
     *
     * @return
     */
    @Bean
    public Binding couponReleaseDelayBinding() {
        return new Binding(couponReleaseDelayQueue, Binding.DestinationType.QUEUE,
                eventExchange,couponReleaseDelayRoutingKey,null);
    }
    /**
     * 绑定死信队列队列和交换机
     * @return
     */
    @Bean
    public Binding couponReleaseDeadBinding() {
        return new Binding(couponReleaseQueue, Binding.DestinationType.QUEUE,
                eventExchange, couponReleaseRoutingKey, null);
    }
}

新建测试：

@RunWith(SpringRunner.class)
@SpringBootTest(classes = CouponApplication.class)
@Slf4j
public class MQTest {
    @Autowired
    private RabbitTemplate rabbitTemplate1;
    @Test
    public void send(){
        rabbitTemplate1.convertAndSend("coupon.event.exchange","coupon.release.delay.routing.key","5qeqweqw");
    }
}

http://127.0.0.1:15672/#进入RabbitMq后台进行查看

配置文件配置如下：

  #消息队列
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    virtual-host: /
    password: guest
    username: guest
    #开启⼿动确认消息
    listener:
      simple:
        acknowledge-mode: manual
  mqconfig:
   #延迟队列，不能被监听消费
    coupon_release_delay_queue: coupon.release.delay.queue
    #延迟队列的消息过期后转发的队列
    coupon_release_queue: coupon.release.queue
    #交换机
    coupon_event_exchange: coupon.event.exchange
  
    #进⼊延迟队列的路由key
    coupon_release_delay_routing_key: coupon.release.delay.routing.key
    #消息过期，进⼊释放死信队列的key
    coupon_release_routing_key: coupon.release.routing.key
    #消息过期时间,毫秒,测试改为15秒
    ttl: 15000

监听者进行监听，同时使用ACK进行标志的确认。代码如下：

@Slf4j
@Component
@RabbitListener(queues = "${mqconfig.coupon_release_queue}")
public class CouponMQListener {
    @Autowired
    private CouponRecordService couponRecordService;
    @Autowired
    private RedissonClient redissonClient;
    @RabbitHandler
    public void ReleaseCouponRecord(CouponRecordMessage recordMessage, Message message, Channel channel) throws IOException {
    log.info("收到传递的消息：{}",recordMessage);
        long tag = message.getMessageProperties().getDeliveryTag();
        boolean flag =  couponRecordService.ReleaseCouponRecord(recordMessage);
//        RLock lock = redissonClient.getLock("coupon:lock:release:" + recordMessage.getTaskId());
//        lock.lock();
        try{
            if(flag){
                channel.basicAck(tag,false);
                log.info("释放优惠券成功：{}",recordMessage);
            }else {
                log.error("释放优惠券失败：{}",recordMessage);
                channel.basicReject(tag,true);
            }
        } catch (Exception e) {
            log.error("释放优惠券异常：{}",e.getMessage());
            channel.basicReject(tag,true);
        }
//        finally {
//            lock.unlock();
//        }
    }

tag后面的参数是是否重入队列。可以通过下面的网页使用配置的账号和密码进行查看

http://127.0.0.1:15672/#/

标签： rabbitmq 分布式 java

本文转载自: https://blog.csdn.net/qq_62627234/article/details/143979753
版权归原作者 努力的程序猿2 所有，如有侵权，请联系我们删除。

如何使用RabbitMq来实现死信队列

发表评论

“如何使用RabbitMq来实现死信队列”的评论:

关于作者

overfit同步小助手

相关阅读

文章导航