前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >消息被拒MQ

消息被拒MQ

作者头像
一个风轻云淡
发布2022-11-13 11:18:25
3620
发布2022-11-13 11:18:25
举报
文章被收录于专栏:java学习java

生产者

代码语言:javascript
复制
/**
 * 消息被拒的情况
 */
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);
        }
    }
}

消费者:

代码语言:javascript
复制
/**
 * 消息被拒的情况
 */
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 -> {});
        }
}

结果:

本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2022-07-26,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档