消息被拒MQ

简介: 消息被拒MQ

生产者


/**
 * 消息被拒的情况
 */
public class Produce0001 {
    private  static  final  String NORMAL_EXCHANGE="normal_exchange";
    public static void main(String[] args) throws  Exception{
        Channel channel = untils.getChannel();
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        //该消息用作队列的个数限制
        for(int i=0;i<10;i++)
        {
            String message="info"+i;
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null,message.getBytes(StandardCharsets.UTF_8));
            System.out.println("生产者发送消息"+message);
        }
    }
}


消费者:


/**
 * 消息被拒的情况
 */
public class Consumer0001 {
        //普通交换机
        private  static  final  String NORMAL_EXCHANGE="normal_exchange";
        //死信交换机
        private  static  final  String DEAD_EXCHANGE="dead_exchange";
        public static void main(String[] args) throws  Exception{
            Channel channel = untils.getChannel();
            //声明死信交换机,类型为direct
            channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
            channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);
            //声明死信队列
            String deadQueue="dead_queue";
            channel.queueDeclare(deadQueue,false,false,false,null);
            //死信队列绑定交换和routingKey值
            channel.queueBind(deadQueue,DEAD_EXCHANGE,"lisi");
            //正常队列绑定死信队列
            Map<String,Object> params=new HashMap<>();
            //正常队列设置死信交换机,参数key是固定值
            params.put("x-dead-letter-exchange",DEAD_EXCHANGE);
            //正常队列设置死信routing-key,参数key是固定值
            params.put("x-dead-letter-routing-key", "lisi");
            //正常队列设置的最大限制长度
            params.put("x-max-length",6);
            System.out.println("等待接收消息....");
            String normalQueue="normal_queue";
            channel.queueDeclare(normalQueue,false,false,false,params);
            channel.queueBind(normalQueue,NORMAL_EXCHANGE,"zhangsan");
            DeliverCallback deliverCallback=(consumerTag, message) -> {
                String s = new String(message.getBody(), StandardCharsets.UTF_8);
                if (s.equals("info5"))
                {
                    System.out.println("info5拒接");
                    channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
                }
                else
                {
                    System.out.println("01接收到消息"+s);
                    channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
                }
            };
            channel.basicConsume(normalQueue,false,deliverCallback,consumerTag -> {});
        }
}


结果:


2dab69ad4ab947f9bcea85d85da4d9cd.png

bb3f99a75625411293d8aa03e0edf330.png003d8e5dcb674dbe8a5232ecb23859be.png


相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
6月前
|
小程序 API
培训报名小程序-订阅消息发送
培训报名小程序-订阅消息发送
|
消息中间件 存储 缓存
3分钟白话RocketMQ系列—— 如何消费消息
3分钟白话RocketMQ系列—— 如何消费消息
733 0
|
4月前
|
消息中间件 编解码 网络协议
京东面试 rockmq是推消息还是拉消息?他的消息模型是啥?
RocketMQ采用拉模式结合长轮询模拟推效果,减少延迟并优化资源使用。在长轮询中,服务器在无消息时保持请求开放,待有新消息时立即响应,提升实时性。利用Netty的TCP连接和异步处理,RocketMQ构建高效通信协议,适应不同吞吐量和实时性需求场景,兼顾控制与实时响应。
48 0
京东面试 rockmq是推消息还是拉消息?他的消息模型是啥?
|
4月前
|
消息中间件 API RocketMQ
消息队列 MQ使用问题之消息在没有消费者的情况下丢失,该如何解决
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
6月前
|
消息中间件
【面试问题】如何确保消息正确地发送至 RabbitMQ? 如何确保消息接收方消费了消息?
【1月更文挑战第27天】【面试问题】如何确保消息正确地发送至 RabbitMQ? 如何确保消息接收方消费了消息?
|
5月前
|
消息中间件 网络协议 物联网
消息队列 MQ产品使用合集之如何让消费者不从最开始进行消费,而是从最后一条消息开始消费
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
6月前
|
消息中间件 存储 Apache
MQ产品使用合集之RocketMQ如果配置所有的ip,有些namesrv挂了的话,消息就发送失败了,消费也是失败的如何解决
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
11月前
|
消息中间件 存储 安全
mq 消费者监听经常断会出现丢消息的问题吗
在消息队列(MQ)系统中,消费者监听经常断开可能会导致消息丢失的问题,具体取决于消息队列系统的设计和配置,以及你的应用程序的处理方式。以下是一些可能导致消息丢失问题的情况: 1. **消费者断开连接:** 如果消费者监听过程中发生意外断开,例如网络故障、消费者应用程序崩溃等,那么在断开连接的瞬间,可能存在未被消费的消息。 2. **消息确认机制:** 消息队列通常提供消息确认机制,确保消息在被成功处理后才被从队列中移除。如果你的消费者应用程序在处理消息时没有发送确认,或者确认机制配置不正确,可能导致消息在被处理前被从队列中移除,从而丢失。 3. **持久化设置:** 消息队列通常提供持久
160 1
|
11月前
|
消息中间件 设计模式 Java
原来RocketMQ消息会重复消费是无奈的”Bug“
大家好,我是三友~~ 在众多关于MQ的面试八股文中有这么一道题,“如何保证MQ消息消费的幂等性”。 为什么需要保证幂等性呢?是因为消息会重复消费。 为什么消息会重复消费? 明明已经消费了,为什么消息会被再次被消费呢? 不同的MQ产生的原因可能不一样 本文就以RocketMQ为例,来扒一扒RocketMQ中会导致消息重复消息的原因,最终你会发现,其实消息重复消费算是RocketMQ无奈的“bug”。
原来RocketMQ消息会重复消费是无奈的”Bug“
|
消息中间件 缓存 算法
阿里二面:RocketMQ 消息积压了,增加消费者有用吗?
阿里二面:RocketMQ 消息积压了,增加消费者有用吗?
258 0
阿里二面:RocketMQ 消息积压了,增加消费者有用吗?