RabbitMQ学习笔记

RabbitMQ学习第一章基本介绍 1.1 什么是MQ？

MQ是利用高效可靠的消息传递机制进程异步的数据传输，并基于数据通信进行分布式系统的集成。通过提高消息队列模型和消息传递记账，可以在分布式扩展进程间的通信。

1.2 MQ的应用场景

跨系统数据传递
高并发的流量削峰
数据的分发和异步处理
大数据分析和传递
分布式事务

1.3 MQ之间的对比

ActiveMQ：是Apache出品，是最流行，能力强劲的开源消息总线。它是一个完全支持JMS规范的消息中间件。有丰富的API，多种集群架构模式让ActiveMQ称为老牌成熟的消息中间件，中小企业使用广泛
Kafka：是LinkedIn开源的分布式发布-订阅消息系统，目前属于Apache顶级项目。主要特点是基于Pull的模式处理消息消费，追求吞吐量。一开始的目的就是用于日志收集和传输，0.8版本之后开始支持复制，不支持事务，对消息的重复、丢失、错误没有严格要求，适合产生大量数据的互联网服务的数据收集业务
RocketMQ：是阿里开源的消息中间件，纯java开发，具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。RocketMQ起源于Kafka，对消息的可靠传输及事务性做了优化，在阿里被广泛用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景
RabbitMQ：使用Erlang语言开发的消息队列系统。基于AMQP协议来实现。主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。AMQP协议更多用在企业系统内对数据一致性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求在其次

第二章 RabbitMQ核心功能 2.1 AMQP协议 2.1.1 基本介绍

AMQP：一种高级消息队列协议，是应用层协议的一个开发标准，为面向消息的中间件设计

2.1.2 AMQP协议生产者的流转过程

2.1.3 AMQP协议消费者的流转过程

2.2 RabbitMQ的核心组成部分

以上是RabbitMQ的运行原理图,下边简单介绍下各个组件的名称和作用

server

又称broker,接受客户端的连接，实现AMQP实体服务
connection

tcp连接，应用程序与broker的网络连接TCP/IP三次握手和四次挥手
channel

网络信道，几乎所有的操作都是在channel中进行，channel是进行消息读写操作的通道，客户端可以建立多个channel，每个channel代表着一个会话任务。
message

消息：服务与应用程序之间传送的数据，由Properties和body组成，Properties可以对消息进行修饰，比如消息的优先级，延迟等高级特性，body就是消息体的内容+
virtual host

虚拟地址，用于进行逻辑隔离，最上层的消息路由，一个虚拟主机可以由若干个exchange和quenue，同一个虚拟主机不能有相同名字的exchange
exchange

交换机，接收消息，根据路由键发送消息到绑定的队列（不具备消息存储的能力），每个模式都需要交换机，发送消息时如果没有指定，则使用默认的交换机
bindings

exchange和queue之间的虚拟连接，bindings中可以保存多个routing key，帮队列和交换机绑定起来
routing key

一个路由规则，虚拟机可以通过它来确定一个如何路由一个特定消息，把消息指定给谁消费（做一个消费控制）
queue

队列，保存消息并转发给消费者

2.3 RabbitMQ支持的模式

RabbotMQ一共支持六种模式分别为简单模式(直拉)、工作队列模式、发布订阅模式、路由模式、主题模式、远程调用模式,由于远程调用模式不常用，这里就不介绍了。

代码演示之前需要做好环境搭建

1.准备依赖

    
        
            com.rabbitmq
            amqp-client
            5.3.0
        
        
            com.google.code.gson
            gson
            2.8.5

2.通用代码抽离

public class RabbitUtils {

    private static ConnectionFactory connectionFactory = new ConnectionFactory();
    static {
        //连接信息
        connectionFactory.setHost("192.168.26.129");
        connectionFactory.setPort(5672);//5672是RabbitMQ的默认端口号
        connectionFactory.setUsername("qianyue");
        connectionFactory.setPassword("qianyue");
        connectionFactory.setVirtualHost("/qianyue");
    }
    public static Connection getConnection(){
        Connection conn = null;
        try {
            conn = connectionFactory.newConnection();
            return conn;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    //关闭连接 必须先关闭信道再关闭连接
    public static void close(Channel channel,Connection connection){
        if(channel != null && channel.isOpen()){
                try {
                    channel.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
            //关闭连接
            if(connection != null && connection.isOpen()){
                try {
                    connection.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
    }
}

2.3.1 简单模式

简单模式,顾名思义,生产者发布消息的时候不指定交换机,消费者只需要通过队列名称就可以拉取消息。

1)生产者代码

//生产者代码
public class Producer {

    public static void main(String[] args) {
        //和服务端建立一个TCP连接
        Connection connection = RabbitUtils.getConnection();
        //以防创建管道出错,先赋一个空值
        Channel channel = null;
        try {
            //创建一个管道连接
            channel = connection.createChannel();
            //第一个参数为队列名称
            //第二个参数 表示是否需要持久化
            //第三个参数 表示是否私有化,false则代表所有消费者都可以访问，true代表只有第一次拥有它的消费者才能一直使用，其他消费者不让访问
            //第四个参数 代表队列所有消息消费完毕后是否自动删除该队列
            //第五个是额外的参数
            channel.queueDeclare(RabbitConstant.QUEUE_HELLOWORLD, false, false, false, null);//声明一个队列
            //消息内容
            String message = "simple-mode:=>helloWorld";

            //发布消息
            //参数说明  参数1:交换机名称  参数2:队列名称(简单模式是队列名称) 参数三:额外的参数 参数四:消息的字节数组
            System.out.println("开始发送消息");
            channel.basicPublish("", RabbitConstant.QUEUE_HELLOWORLD, null, message.getBytes());


        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            //关闭信道和连接
            RabbitUtils.close(channel, connection);


        }
    }

}

2)消费者代码

public class Consumer {

    public static void main(String[] args) {
        //和服务端建立一个TCP连接
        Connection connection = RabbitUtils.getConnection();
        //以防创建管道出错,先赋一个空值
        Channel channel = null;
        try {
            //创建一个管道连接
            channel =  connection.createChannel();
            //第一个参数为队列名称
            //第二个参数 表示是否需要持久化
            //第三个参数 表示是否私有化,false则代表所有消费者都可以访问，true代表只有第一次拥有它的消费者才能一直使用，其他消费者不让访问
            //第四个参数 代表队列所有消息消费完毕后是否自动删除该队列
            //第五个是额外的参数
            channel.queueDeclare(RabbitConstant.QUEUE_HELLOWORLD,false,false,false,null);//声明一个队列
            //消息内容
            channel.basicConsume(RabbitConstant.QUEUE_HELLOWORLD, false, new DeliverCallback() {
                public void handle(String s, Delivery delivery) throws IOException {
                    System.out.println("处理消息:" +new String(delivery.getBody(),"utf-8"));
                }
            }, new CancelCallback() {
                public void handle(String s) throws IOException {
                    System.out.println("消息处理失败");
                }
            });


        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            //消费者需要不停读取消息,这里不能关闭
            //RabbitUtils.close(channel,connection); 
        }

    }
}

3)运行结果

生产者

消费者

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-uGJgKkLn-1651841021480)(C:UserswcyAppDataRoamingTyporatypora-user-imagesimage-20220504193123353.png)]

4)说明

简单模式不需要声明交换机,只需要指明队列名称就可以。

2.3.2 工作队列模式

工作队列模式,相较于简单模式来说,消费者可以有多个,当我们的MQ中消息比较多,一个消费者可能处理不完,这时候可以多提供一些消费者来处理消息,因为队列中的消息是被消费后就会消息,所以多个消费者共同消费的时候不会拿到同一个消息。

1)生产者代码

//生产者代码
public class Producer {

    public static void main(String[] args) {
        //和服务端建立一个TCP连接
        Connection connection = RabbitUtils.getConnection();
        //以防创建管道出错,先赋一个空值
        Channel channel = null;
        try {
            //创建一个管道连接
            channel =  connection.createChannel();
            //第一个参数为队列名称
            //第二个参数 表示是否需要持久化
            //第三个参数 表示是否私有化,false则代表所有消费者都可以访问，true代表只有第一次拥有它的消费者才能一直使用，其他消费者不让访问
            //第四个参数 代表队列所有消息消费完毕后是否自动删除该队列
            //第五个是额外的参数

            channel.queueDeclare(RabbitConstant.QUEUE_SMS,false,false,false,null);//声明一个队列
            //投递100条消息模拟订票
            for(int i=1;i<100;i++){
                    SMS sms = new SMS("乘客"+i,"156498623"+i,"您的车票预定成功");
                    String message = new Gson().toJson(sms);
                    channel.basicPublish("",RabbitConstant.QUEUE_SMS,null,message.getBytes());
            }

        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            //关闭信道和连接
            RabbitUtils.close(channel,connection);
        }


    }

}

2)消费者1代码

public class Consumer {

    public static void main(String[] args) {
        //和服务端建立一个TCP连接
        Connection connection = RabbitUtils.getConnection();
        //以防创建管道出错,先赋一个空值
         Channel channel = null;
        try {
            //创建一个管道连接
            channel =  connection.createChannel();
            //第一个参数为队列名称
            //第二个参数 表示是否需要持久化
            //第三个参数 表示是否私有化,false则代表所有消费者都可以访问，true代表只有第一次拥有它的消费者才能一直使用，其他消费者不让访问
            //第四个参数 代表队列所有消息消费完毕后是否自动删除该队列
            //第五个是额外的参数
            channel.queueDeclare(RabbitConstant.QUEUE_SMS,false,false,false,null);//声明一个队列
            //如果不写basicQos（1），则MQ会将所有请求平均发送给所有消费者
            channel.basicQos(5);

            final Channel finalChannel = channel;
            channel.basicConsume(RabbitConstant.QUEUE_SMS , false , new DefaultConsumer(finalChannel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String jsonSMS = new String(body);
                    System.out.println("Consumer-短信发送成功:" + jsonSMS);

                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    finalChannel.basicAck(envelope.getDeliveryTag() , false);
                }
            });


        } catch (IOException e) {
            e.printStackTrace();
        }finally {
           //关闭管道
            RabbitUtils.close(channel,connection);
        }


    }
}

3)消费者2代码

public class Consumer1 {

    public static void main(String[] args) {
        //和服务端建立一个TCP连接
        Connection connection = RabbitUtils.getConnection();
        //以防创建管道出错,先赋一个空值
         Channel channel = null;
        try {
            //创建一个管道连接
            channel =  connection.createChannel();
            //第一个参数为队列名称
            //第二个参数 表示是否需要持久化
            //第三个参数 表示是否私有化,false则代表所有消费者都可以访问，true代表只有第一次拥有它的消费者才能一直使用，其他消费者不让访问
            //第四个参数 代表队列所有消息消费完毕后是否自动删除该队列
            //第五个是额外的参数
            channel.queueDeclare(RabbitConstant.QUEUE_SMS,false,false,false,null);//声明一个队列
            //如果不写basicQos（1），则MQ会将所有请求平均发送给所有消费者
            channel.basicQos(1);

            final Channel finalChannel = channel;
            channel.basicConsume(RabbitConstant.QUEUE_SMS , false , new DefaultConsumer(finalChannel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String jsonSMS = new String(body);
                    System.out.println("Consumer1-短信发送成功:" + jsonSMS);

                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    finalChannel.basicAck(envelope.getDeliveryTag() , false);
                }
            });



        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            //关闭信道和连接
            RabbitUtils.close(channel,connection);

        }
    }
}

4)消费者3代码

public class Consumer2 {

    public static void main(String[] args) {
        //和服务端建立一个TCP连接
        Connection connection = RabbitUtils.getConnection();
        //以防创建管道出错,先赋一个空值
         Channel channel = null;
        try {
            //创建一个管道连接
            channel =  connection.createChannel();
            //第一个参数为队列名称
            //第二个参数 表示是否需要持久化
            //第三个参数 表示是否私有化,false则代表所有消费者都可以访问，true代表只有第一次拥有它的消费者才能一直使用，其他消费者不让访问
            //第四个参数 代表队列所有消息消费完毕后是否自动删除该队列
            //第五个是额外的参数
            channel.queueDeclare(RabbitConstant.QUEUE_SMS,false,false,false,null);//声明一个队列
            //如果不写basicQos（1），则MQ会将所有请求平均发送给所有消费者
            channel.basicQos(1);

            final Channel finalChannel = channel;
            channel.basicConsume(RabbitConstant.QUEUE_SMS , false , new DefaultConsumer(finalChannel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String jsonSMS = new String(body);
                    System.out.println("Consumer2-短信发送成功:" + jsonSMS);

                    try {
                        Thread.sleep(50);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    finalChannel.basicAck(envelope.getDeliveryTag() , false);
                }
            });



        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            //关闭信道和连接
            RabbitUtils.close(channel,connection);
        }


    }
}

5)运行结果

消费者1:

消费者2:

消费者3：

6)说明

工作模式因为有多个消费者共同消费一个队列中的消息,所以有两种策略,一种是轮询策略另一种是公平分发

轮询策略:一个消费者一个,不会因为谁处理消息快就多分配一些消息,大家都是排队一人一条消息，这里是自动应答

公平分发:处理消息快的则多拿一些消息处理。但是需要开启手动应答。

公平分发三个点需要注意

(1)关闭自动应答

channel.basicConsume(RabbitConstant.QUEUE_SMS , false , new DefaultConsumer(finalChannel) 第二个参数为false

(2)设置QPS

channel.basicQos(1); 值越大拿的消息越多

(3)手动应答

finalChannel.basicAck(envelope.getDeliveryTag() , false); 第一个参数是消息ID,第二个是应答所有未应答的消息还是仅仅应答当前消息,false为仅仅应答当前消息。

2.3.3 发布订阅模式

生产者通过交换机把消息发送到指定的队列,消费者可以从绑定了该交换机的队列中拿到消息

1)生产者代码

//推送各个地市的天气数据
public class WeatherBureau {

    public static void main(String[] args) {
        //和服务端建立一个TCP连接
        Connection connection = RabbitUtils.getConnection();
        //以防创建管道出错,先赋一个空值
        Channel channel = null;
        try {
            //创建一个管道连接
            channel =  connection.createChannel();

            String message = "2022年五月四日,南宁市,晴,32°C";

            channel.basicPublish(RabbitConstant.EXCHANGE_WEATHER,"",null,message.getBytes());


        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            //关闭信道和连接
            RabbitUtils.close(channel,connection);
        }
    }
}

2)消费者代码

//新浪订阅天气消息
public class Sina {


    public static void main(String[] args) {
        //和服务端建立一个TCP连接
        Connection connection = RabbitUtils.getConnection();
        //以防创建管道出错,先赋一个空值
        Channel channel = null;
        try {
            //创建一个管道连接
            channel =  connection.createChannel();
            //第一个参数为队列名称
            //第二个参数 表示是否需要持久化
            //第三个参数 表示是否私有化,false则代表所有消费者都可以访问，true代表只有第一次拥有它的消费者才能一直使用，其他消费者不让访问
            //第四个参数 代表队列所有消息消费完毕后是否自动删除该队列
            //第五个是额外的参数
            channel.queueDeclare(RabbitConstant.QUEUE_SINA,false,false,false,null);//声明一个队列

            channel.queueBind(RabbitConstant.QUEUE_SINA,RabbitConstant.EXCHANGE_WEATHER,"");

            final Channel finalChannel = channel;
            channel.basicConsume(RabbitConstant.QUEUE_SINA,false,new DefaultConsumer(finalChannel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("新浪天气收到气象信息：" + new String(body));
                    finalChannel.basicAck(envelope.getDeliveryTag() , false);
                }
            });


        } catch (IOException e) {
            e.printStackTrace();
        }finally {
                //关闭连接
                RabbitUtils.close(channel,connection);    
    
            }   
        
    }

}

//百度接收天气消息
public class BaiDu {


    public static void main(String[] args) {
        //和服务端建立一个TCP连接
        Connection connection = RabbitUtils.getConnection();
        //以防创建管道出错,先赋一个空值
        Channel channel = null;
        try {
            //创建一个管道连接
            channel =  connection.createChannel();
            //第一个参数为队列名称
            //第二个参数 表示是否需要持久化
            //第三个参数 表示是否私有化,false则代表所有消费者都可以访问，true代表只有第一次拥有它的消费者才能一直使用，其他消费者不让访问
            //第四个参数 代表队列所有消息消费完毕后是否自动删除该队列
            //第五个是额外的参数
            channel.queueDeclare(RabbitConstant.QUEUE_BAIDU,false,false,false,null);//声明一个队列


            //把队列和交换机绑定 参数1表示队列名称 参数2代表队列名 参数3代表路由key
            channel.queueBind(RabbitConstant.QUEUE_BAIDU,RabbitConstant.EXCHANGE_WEATHER,"");

            final Channel finalChannel = channel;
            channel.basicConsume(RabbitConstant.QUEUE_BAIDU,false,new DefaultConsumer(finalChannel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("新浪天气收到气象信息：" + new String(body));
                    finalChannel.basicAck(envelope.getDeliveryTag() , false);
                }
            });


        } catch (IOException e) {
            e.printStackTrace();
        }finally {
             RabbitUtils.close(channel,connection);
        }
        

    }
}

3)运行结果

生产者

新浪消费者

百度消费者

4)说明

工作队列模式,交换机类型为faount,实现广播，只要和该交换机绑定的队列都可以收到该条消息,然后消费者通过订阅对应的队列就可以收到这条广播。

实现消费复用,每个消费者拿到同一份数据处理自己的逻辑。

2.3.4 路由模式

faount的交换机，只要是绑定了该交换机的队列就可以收到消息，有时候我们想要筛选，满足某一条件的消息去投递到某一个指定的队列，这时候就需要我们的路由模式,通过direct模式实现路由功能。

1)生产者代码

//推送各个地市的天气数据
public class WeatherBureau {

    public static void main(String[] args) {


        Map area = new LinkedHashMap();
        area.put("china.hunan.changsha.20201127", "中国湖南长沙20201127天气数据");
        area.put("china.hubei.wuhan.20201127", "中国湖北武汉20201127天气数据");
        area.put("china.hunan.zhuzhou.20201127", "中国湖南株洲20201127天气数据");
        area.put("us.cal.lsj.20201127", "美国加州洛杉矶20201127天气数据");

        area.put("china.hebei.shijiazhuang.20201128", "中国河北石家庄20201128天气数据");
        area.put("china.hubei.wuhan.20201128", "中国湖北武汉20201128天气数据");
        area.put("china.henan.zhengzhou.20201128", "中国河南郑州20201128天气数据");
        area.put("us.cal.lsj.20201128", "美国加州洛杉矶20201128天气数据");


        //和服务端建立一个TCP连接
        Connection connection = RabbitUtils.getConnection();
        //以防创建管道出错,先赋一个空值
        Channel channel = null;
        try {
            //创建一个管道连接
            channel =  connection.createChannel();

            Iterator> itr = area.entrySet().iterator();
            while (itr.hasNext()){
                Map.Entry me = itr.next();
                //第一个参数交换机名字   第二个参数作为 消息的routing key
                channel.basicPublish(RabbitConstant.EXCHANGE_WEATHER_ROUTING,me.getKey() , null , me.getValue().getBytes());
            }


        } catch (IOException e) {
            e.printStackTrace();
        }finally {

            RabbitUtils.close(channel,connection);


        }

    }
}

2)消费者代码

新浪

//新浪订阅天气消息
public class Sina {


    public static void main(String[] args) {
        //和服务端建立一个TCP连接
        Connection connection = RabbitUtils.getConnection();
        //以防创建管道出错,先赋一个空值
        Channel channel = null;
        try {
            //创建一个管道连接
            channel =  connection.createChannel();
            //第一个参数为队列名称
            //第二个参数 表示是否需要持久化
            //第三个参数 表示是否私有化,false则代表所有消费者都可以访问，true代表只有第一次拥有它的消费者才能一直使用，其他消费者不让访问
            //第四个参数 代表队列所有消息消费完毕后是否自动删除该队列
            //第五个是额外的参数
            channel.queueDeclare(RabbitConstant.QUEUE_SINA,false,false,false,null);//声明一个队列

            channel.queueBind(RabbitConstant.QUEUE_SINA,RabbitConstant.EXCHANGE_WEATHER_ROUTING,"china.hunan.changsha.20201127");
            channel.queueBind(RabbitConstant.QUEUE_SINA,RabbitConstant.EXCHANGE_WEATHER_ROUTING,"us.cal.lsj.20201127");



            final Channel finalChannel = channel;
            channel.basicConsume(RabbitConstant.QUEUE_SINA,false,new DefaultConsumer(finalChannel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("新浪天气收到气象信息：" + new String(body));
                    finalChannel.basicAck(envelope.getDeliveryTag() , false);
                }
            });


        } catch (IOException e) {
            e.printStackTrace();
        }finally {

           // RabbitUtils.close(channel,connection);
        }




    }


}

百度

//百度接收天气消息
public class BaiDu {


    public static void main(String[] args) {
        //和服务端建立一个TCP连接
        Connection connection = RabbitUtils.getConnection();
        //以防创建管道出错,先赋一个空值
        Channel channel = null;
        try {
            //创建一个管道连接
            channel =  connection.createChannel();
            //第一个参数为队列名称
            //第二个参数 表示是否需要持久化
            //第三个参数 表示是否私有化,false则代表所有消费者都可以访问，true代表只有第一次拥有它的消费者才能一直使用，其他消费者不让访问
            //第四个参数 代表队列所有消息消费完毕后是否自动删除该队列
            //第五个是额外的参数
            channel.queueDeclare(RabbitConstant.QUEUE_BAIDU,false,false,false,null);//声明一个队列


            //把队列和交换机绑定 参数1表示队列名称 参数2代表队列名 参数3代表路由key
            channel.queueBind(RabbitConstant.QUEUE_BAIDU,RabbitConstant.EXCHANGE_WEATHER_ROUTING,"china.hebei.shijiazhuang.20201128");
            channel.queueBind(RabbitConstant.QUEUE_BAIDU,RabbitConstant.EXCHANGE_WEATHER_ROUTING,"china.hubei.wuhan.20201128");

            final Channel finalChannel = channel;
            channel.basicConsume(RabbitConstant.QUEUE_BAIDU,false,new DefaultConsumer(finalChannel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("百度天气收到气象信息：" + new String(body));
                    finalChannel.basicAck(envelope.getDeliveryTag() , false);
                }
            });


        } catch (IOException e) {
            e.printStackTrace();
        }finally {

            // RabbitUtils.close(channel,connection);
        }




    }
}

3)运行结果

新浪

百度

4)说明

路由模式下,我们可以指定哪些服务消费哪些消息,而不是所有消息都接收.

2.3.5主题模式

主题模式相当于路由模式的增强,只不过路由模式是一个精确匹配，主题模式是一个模糊匹配,可以同时向多个满足条件的队列投递消息。

1)生产者代码

//推送各个地市的天气数据
public class WeatherBureau {

    public static void main(String[] args) {


        Map area = new LinkedHashMap();
        area.put("china.hunan.changsha.20201127", "中国湖南长沙20201127天气数据");
        area.put("china.hubei.wuhan.20201127", "中国湖北武汉20201127天气数据");
        area.put("china.hunan.zhuzhou.20201127", "中国湖南株洲20201127天气数据");
        area.put("us.cal.lsj.20201127", "美国加州洛杉矶20201127天气数据");

        area.put("china.hebei.shijiazhuang.20201128", "中国河北石家庄20201128天气数据");
        area.put("china.hubei.wuhan.20201128", "中国湖北武汉20201128天气数据");
        area.put("china.henan.zhengzhou.20201128", "中国河南郑州20201128天气数据");
        area.put("us.cal.lsj.20201128", "美国加州洛杉矶20201128天气数据");


        //和服务端建立一个TCP连接
        Connection connection = RabbitUtils.getConnection();
        //以防创建管道出错,先赋一个空值
        Channel channel = null;
        try {
            //创建一个管道连接
            channel =  connection.createChannel();
            System.out.println("生产者开始投递消息");
            Iterator> itr = area.entrySet().iterator();
            while (itr.hasNext()){
                Map.Entry me = itr.next();
                //第一个参数交换机名字   第二个参数作为 消息的routing key
                channel.basicPublish(RabbitConstant.EXCHANGE_WEATHER_TOPIC,me.getKey() , null , me.getValue().getBytes());
            }
            System.out.println("生产者结束投递消息");


        } catch (IOException e) {
            e.printStackTrace();
        }finally {
           RabbitUtils.close(channel,connection);

        }
    }
}

2)消费者代码

新浪

//新浪订阅天气消息
public class Sina {


    public static void main(String[] args) {
        //和服务端建立一个TCP连接
        Connection connection = RabbitUtils.getConnection();
        //以防创建管道出错,先赋一个空值
        Channel channel = null;
        try {
            //创建一个管道连接
            channel =  connection.createChannel();
            //第一个参数为队列名称
            //第二个参数 表示是否需要持久化
            //第三个参数 表示是否私有化,false则代表所有消费者都可以访问，true代表只有第一次拥有它的消费者才能一直使用，其他消费者不让访问
            //第四个参数 代表队列所有消息消费完毕后是否自动删除该队列
            //第五个是额外的参数
            channel.queueDeclare(RabbitConstant.QUEUE_SINA,false,false,false,null);//声明一个队列

            channel.queueBind(RabbitConstant.QUEUE_SINA,RabbitConstant.EXCHANGE_WEATHER_TOPIC,"us.#");




            final Channel finalChannel = channel;
            channel.basicConsume(RabbitConstant.QUEUE_SINA,false,new DefaultConsumer(finalChannel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("新浪天气收到气象信息：" + new String(body));
                    finalChannel.basicAck(envelope.getDeliveryTag() , false);
                }
            });


        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            //         RabbitUtils.close(channel,connection);
        }




    }


}

百度

//百度接收天气消息
public class BaiDu {


    public static void main(String[] args) {
        //和服务端建立一个TCP连接
        Connection connection = RabbitUtils.getConnection();
        //以防创建管道出错,先赋一个空值
        Channel channel = null;
        try {
            //创建一个管道连接
            channel =  connection.createChannel();
            //第一个参数为队列名称
            //第二个参数 表示是否需要持久化
            //第三个参数 表示是否私有化,false则代表所有消费者都可以访问，true代表只有第一次拥有它的消费者才能一直使用，其他消费者不让访问
            //第四个参数 代表队列所有消息消费完毕后是否自动删除该队列
            //第五个是额外的参数
            channel.queueDeclare(RabbitConstant.QUEUE_BAIDU,false,false,false,null);//声明一个队列


            //把队列和交换机绑定 参数1表示队列名称 参数2代表队列名 参数3代表路由key
            channel.queueBind(RabbitConstant.QUEUE_BAIDU,RabbitConstant.EXCHANGE_WEATHER_TOPIC,"*.*.*.20201127");


            final Channel finalChannel = channel;
            channel.basicConsume(RabbitConstant.QUEUE_BAIDU,false,new DefaultConsumer(finalChannel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("百度天气收到气象信息：" + new String(body));
                    finalChannel.basicAck(envelope.getDeliveryTag() , false);
                }
            });


        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            //         RabbitUtils.close(channel,connection);
        }

    }
}

3)运行结果

生产者

新浪

百度

4)说明

主题模式是路由模式的一个增强,使用的交换机是topic类型,可以通过模糊匹配像多个队列存放消息。

第三章 Spring集成RabbitMQ

本章主要介绍Spring是如何集成RabbitMQ的,有助于接下来Springboot用配置类加注解的方式配置我们的RabbitMQ

演示前代码准备

(1)引入依赖

    
        
            org.springframework
            spring-context
            5.1.7.RELEASE
        

        
            org.springframework.amqp
            spring-rabbit
            2.1.8.RELEASE
        

        
            junit
            junit
            4.12
        

        
            org.springframework
            spring-test
            5.1.7.RELEASE

(2)编写存放连接新的properties

rabbitmq.host=192.168.26.129
rabbitmq.port=5672
rabbitmq.username=qianyue
rabbitmq.password=qianyue
rabbitmq.virtual-host=/qianyue

(3)编写配置文件

3.1 简单模式 1)生产者配置

(1)xml文件

(2)发送消息

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class ProducerTest {

    @Autowired
    private RabbitTemplate  rabbitTemplate;


    @Test
    public void testHelloWorld(){

        //简单模式
        System.out.println("简单模式=>生产者发送消息");
        rabbitTemplate.convertAndSend("spring_queue","hello world spring....");
    }

}

2)消费者配置

(1)监听类

public class SpringQueueListener implements MessageListener {
    public void onMessage(Message message) {
        System.out.println("收到消息=>"+message);
    }
}

(2)xml

3.加载配置文件

public class ConsumerTest {

    public static void main(String[] args) {
        ApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:spring-rabbitmq-consumer.xml");
    }
}

3)运行结果

生产者

消费者

3.2 发布订阅模式 1)生产者代码

1.xml配置

2.发送消息

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class ProducerTest {

    @Autowired
    private RabbitTemplate  rabbitTemplate;


    @Test
    public void testHelloWorld(){

        //广播模式
        System.out.println("广播模式=>生产者发送消息");
        rabbitTemplate.convertAndSend("spring_fanout_queue_1","hello,fanout1");
        rabbitTemplate.convertAndSend("spring_fanout_queue_2","hello,fanout2");
    }


}

2)消费者代码

1.监听器

FanoutListener

public class FanoutListener implements MessageListener {
    public void onMessage(Message message) {
        System.out.println("收到消息=>"+message);
    }
}

FanoutListener2

public class FanoutListener2 implements MessageListener {
    public void onMessage(Message message) {
        System.out.println("收到消息=>"+message);
    }
}

2.xml配置

3.运行结果

生产者

消费者

3.3 路由模式 1)生产者代码

1.xml

2.发送消息

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class ProducerTest {

    @Autowired
    private RabbitTemplate  rabbitTemplate;


    @Test
    public void testHelloWorld(){

        //路由模式模式
        System.out.println("路由模式=>生产者发送消息");
        rabbitTemplate.convertAndSend("spring_direct_exchange","info","direct");

    }


}

2)消费者代码

1.监听器

public class DirectListener implements MessageListener {

    public void onMessage(Message message) {
        System.out.println("收到消息=>"+message);
    }
}

2.xml配置

3.运行结果

生产者

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-aabToiPr-1651841021488)(C:UserswcyAppDataRoamingTyporatypora-user-imagesimage-20220504223351625.png)]

消费者

3.4 主题模式 1）生产者代码

1.xml配置

2.发送消息

    @Test
    public void testHelloWorld(){

        //路由模式模式
        System.out.println("主题模式=>生产者发送消息");
        rabbitTemplate.convertAndSend("spring_topic_exchange","baidu.com","百度");
        rabbitTemplate.convertAndSend("spring_topic_exchange","alibaba.com","阿里巴巴");
        rabbitTemplate.convertAndSend("spring_topic_exchange","tencent.com","腾讯");

    }

2）消费者代码

1.监听器

腾讯监听器

public class TopicTencetListener implements MessageListener {
    public void onMessage(Message message) {
        System.out.println("收到消息=>"+message);
    }
}

阿里监听器

public class TopicAliBaBaListener implements MessageListener {
    public void onMessage(Message message) {
        System.out.println("收到消息=>"+message);
    }
}

百度监听器

public class TopicBaiDuListener implements MessageListener {
    public void onMessage(Message message) {
        System.out.println("收到消息=>"+message);
    }
}

2.xml配置

3.运行结果

生产者

消费者

第四章 Springboot集成RabbitMQ

环境搭建

1.引入依赖

   

        
            org.springframework.boot
            spring-boot-starter-amqp
        
        
            org.springframework.boot
            spring-boot-starter-web
        

        
            org.projectlombok
            lombok
            true
        
        
            org.springframework.boot
            spring-boot-starter-test
            test
        

        
            io.springfox
            springfox-swagger2
            2.6.0
        
        
            io.springfox
            springfox-swagger-ui
            2.6.0
        
        
            org.junit.jupiter
            junit-jupiter
            RELEASE
            test

2.配置文件

server.port=8080
#这几个是默认配置。 不配也行。
spring.rabbitmq.host=192.168.26.129
spring.rabbitmq.port=5672
spring.rabbitmq.username=qianyue
spring.rabbitmq.password=qianyue
spring.rabbitmq.virtual-host=/qianyue

4.1 简单模式

1.配置类

//简单模式
@Configuration
public class SimpleConfig {

    @Bean
    public Queue simpleQueue(){
        return new Queue("simpleQueue"); 
    }

}

2.生产者

	@Autowired
	private RabbitTemplate rabbitTemplate;

	//helloWorld 直连模式
	@ApiOperation(value="helloWorld发送接口",notes="直接发送到队列")
	@GetMapping(value="/helloWorldSend")
	public Object helloWorldSend(String message) throws AmqpException, UnsupportedEncodingException {
		//设置部分请求参数
		MessageProperties messageProperties = new MessageProperties();
		messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);

		//发消息
		rabbitTemplate.send("simpleQueue",new Message(message.getBytes("UTF-8"),messageProperties));
		return "message send: "+message;
	}

3.消费者

@Component
public class Consumer {

    @RabbitListener(queues = "simpleQueue")
    public void helloWorldListener(Message message){
        System.out.println("简单模式接收消息=>"+message);
    }

}

4.运行结果

4.2 工作队列模式

1.配置类

@Configuration
public class WorkQueueConfig {


    @Bean
    public Queue workQueue(){
        return new Queue("workQueue");
    }


}

2.生产者

	//工作队列模式
	@ApiOperation(value="workqueue发送接口",notes="发送到所有监听该队列的消费")
	@GetMapping(value="/workqueueSend")
	public Object workqueueSend(String message) throws AmqpException, UnsupportedEncodingException {
		MessageProperties messageProperties = new MessageProperties();
		messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);
		//制造多个消息进行发送操作
		for (int i = 0; i <10 ; i++) {
			rabbitTemplate.send("workQueue",  new Message(message.getBytes("UTF-8"),messageProperties));
		}
		return "message send : "+message;
	}

3.消费者

    @RabbitListener(queues ="workQueue")
    public void workQueueListener1(Message message){
        System.out.println("工作队列模式消费者1收到消息"+message);
    }

    @RabbitListener(queues ="workQueue")
    public void workQueueListener2(Message message){
        System.out.println("工作队列模式消费者2收到消息"+message);
    }

4.运行结果

4.3 发布订阅模式

1.配置类

@Configuration
public class PubSubConfig {


    //声明一个队列1
    @Bean
    public Queue pubSubQueue1(){
        return  new Queue("pub_sub_queue1");
    }

    //声明一个队列2
    @Bean
    public Queue pubSubQueue2(){
        return  new Queue("pub_sub_queue2");
    }

    //声明一个广播交换机
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("fanoutExchange");
    }
    //队列1绑定交换机
    @Bean
    public Binding bind1(){
        return BindingBuilder.bind(pubSubQueue1()).to(fanoutExchange());
    }
    //队列2绑定交换机
    @Bean
    public Binding bind2(){
        return BindingBuilder.bind(pubSubQueue2()).to(fanoutExchange());
    }


}

2.生产者代码

	// pub/sub 发布订阅模式   交换机类型 fanout
	@ApiOperation(value="fanout发送接口",notes="发送到fanoutExchange。消息将往该exchange下的所有queue转发")
	@GetMapping(value="/fanoutSend")
	public Object fanoutSend(String message) throws AmqpException, UnsupportedEncodingException {
		MessageProperties messageProperties = new MessageProperties();
		messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);
		//fanout模式只往exchange里发送消息。分发到exchange下的所有queue
		rabbitTemplate.send("fanoutExchange", "", new Message(message.getBytes("UTF-8"),messageProperties));
		return "message send : "+message;
	}

3.消费者代码

    @RabbitListener(queues = "pub_sub_queue1")
    public void pubSubListener1(Message message){
        System.out.println("广播模式消费者1监听队列1=>"+message);
    }
    @RabbitListener(queues = "pub_sub_queue2")
    public void pubSubListener2(Message message){
        System.out.println("广播模式消费者2监听队列2=>"+message);
    }

4.运行结果

4.4 路由模式

一个交换机只是通过不同的路由key去发送到绑定的队列中

1.配置文件

@Configuration
public class DirectConfig {

    //声明队列1
    @Bean
    public Queue directQueue1(){
        return new Queue("directQueue1");
    }

    //声明队列2
    @Bean
    public Queue directQueue2(){
        return new Queue("directQueue2");
    }
    //声明队列3
    @Bean
    public Queue directQueue3(){
        return new Queue("directQueue3");
    }
    //声明交换机
    @Bean
    public DirectExchange directExchange(){
        return new DirectExchange("directExchange");
    }
    //把队列2和交换机通过路由key"baidu"连接起来
    @Bean
    public Binding directBind1(){
        return BindingBuilder.bind(directQueue1()).to(directExchange()).with("baidu");
    }

    //把队列2和交换机通过路由key"alibaba"连接起来
    @Bean
    public Binding directBind2(){
        return BindingBuilder.bind(directQueue2()).to(directExchange()).with("alibaba");
    }

    //把队列3和交换机通过路由key"tencent"连接起来
    @Bean
    public Binding directBind3(){
        return BindingBuilder.bind(directQueue3()).to(directExchange()).with("tencent");
    }

}

2.生产者

	//routing路由工作模式  交换机类型 direct
	@ApiOperation(value="direct发送接口",notes="发送到directExchange。exchange转发消息时，会往routingKey匹配的queue发送")
	@GetMapping(value="/directSend")
	public Object routingSend(String routingKey,String message) throws AmqpException, UnsupportedEncodingException {

		if(null == routingKey) {
			routingKey="baidu";
		}
		MessageProperties messageProperties = new MessageProperties();
		messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);
		//fanout模式只往exchange里发送消息。分发到exchange下的所有queue
		rabbitTemplate.send("directExchange", routingKey, new Message(message.getBytes("UTF-8"),messageProperties));
		return "message send : routingKey >"+routingKey+";message > "+message;
	}

3.消费者

    @RabbitListener(queues = "directQueue1")
    public void directListener1(Message message){
        System.out.println("路由模式消费者1监听队列1=>"+message);
    }

    @RabbitListener(queues = "directQueue2")
    public void directListener2(Message message){
        System.out.println("路由模式消费者2监听队列2=>"+message);
    }

    @RabbitListener(queues = "directQueue3")
    public void directListener3(Message message){
        System.out.println("路由模式消费者3监听队列3=>"+message);
    }

4.运行结果

4.5 主题模式

1.配置类

@Configuration
public class TopicConfig {
    //声明队列1
    @Bean
    public Queue topicQueue1(){
        return new Queue("topicQueue1");
    }

    //声明队列2
    @Bean
    public Queue topicQueue2(){
        return new Queue("topicQueue2");
    }
    //声明队列3
    @Bean
    public Queue topicQueue3(){
        return new Queue("topicQueue3");
    }

    //声明交换机
    @Bean
    public TopicExchange topicExchange(){
        return  new TopicExchange("topicExchange");
    }

    @Bean
    public Binding topicBind1(){
        return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("*.baidu.*");
    }

    @Bean
    public Binding topicBind2(){
        return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("#.baidu");
    }

    @Bean
    public Binding topicBind3(){
        return BindingBuilder.bind(topicQueue3()).to(topicExchange()).with("*.baidu");
    }

}

2.生产者

	//topic 工作模式   交换机类型 topic
	@ApiOperation(value="topic发送接口",notes="发送到topicExchange。exchange转发消息时，会往routingKey匹配的queue发送，*代表一个单词，#代表0个或多个单词。")
	@GetMapping(value="/topicSend")
	public Object topicSend(String routingKey,String message) throws AmqpException, UnsupportedEncodingException {

		if(null == routingKey) {
			routingKey="www.baidu.com";
		}
		MessageProperties messageProperties = new MessageProperties();
		messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);
		//fanout模式只往exchange里发送消息。分发到exchange下的所有queue
		rabbitTemplate.send("topicExchange", routingKey, new Message(message.getBytes("UTF-8"),messageProperties));
		return "message send : routingKey >"+routingKey+";message > "+message;
	}

3.消费者

    @RabbitListener(queues = "topicQueue1")
    public void topicListener1(Message message){
        System.out.println("主题模式消费者1监听队列1=>"+message);
    }


    @RabbitListener(queues = "topicQueue2")
    public void topicListener2(Message message){
        System.out.println("主题模式消费者2监听队列2=>"+message);
    }


    @RabbitListener(queues = "topicQueue3")
    public void topicListener3(Message message){
        System.out.println("主题模式消费者3监听队列3=>"+message);
    }

4.运行结果

第五章高级特性 5.1 消息的可靠投递

在使用RabbitMQ的时候,作为消息发送方希望杜绝任何消息丢失或者投递失败的场景。RabbitMQ为我们提供了两种方式用来控制消息的投递可靠性模式。

confirm 确认模式
return 退回模式

先了解一下我们消息的流转过程

producer->rabbitMQ broker->exchange->queue->consumer

消息从producer到exchange会返回一个confirmCallback。
消息从exchange->queue投递失败则会返回一个returnCallback。可以利用这两个模式来控制消息的可靠投递。

接下来就是代码演示

1)xml配置

特别注意 publisher-confirms="true"和publisher-returns=“true” 代码不可省略

2)confirm代码

        //测试可靠消费
        @Test
        public  void TestConfirm(){

            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause ) {
                    if(ack){
                        System.out.println("消息已到rabbitmq broker=>"+cause);
                    }else {
                        System.out.println("消息失败到达rabbitmq broker=>"+cause);
                        //这里可以做个兜底 让生产者重新投递这条信息。
                    }

                }
            });

            //进行消息发送
            for (int i = 0; i < 5; i++) {
                rabbitTemplate.convertAndSend("test_exchange_confirm","confirm","message Confirm...");

            }

        }

3)return模式

       //设置交换机处理失败消息的模式   为true的时候，消息达到不了 队列时，会将消息重新返回给生产者
        @Test
        public void testReturn(){
            rabbitTemplate.setMandatory(true);
            //定义回调
            
            rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
                @Override
                public void returnedMessage(Message message, int relyCode, String replyText, String exchange, String routingKey) {
                    System.out.println("退回模式回调方法执行=>");
                    System.out.println("message=>"+message);
                    System.out.println("relyCode=>"+relyCode);
                    System.out.println("replyText=>"+replyText);
                    System.out.println("exchange=>"+exchange);
                    System.out.println("routingKey=>"+routingKey);
                }
            });

            //进行消息发送
            rabbitTemplate.convertAndSend("test_exchange_confirm","confirm1","message return...");
        }

4)小结

设置ConnectionFactory的publisher-confirms=“true” 开启确认模式。
使用rabbitTemplate.setConfirmCallback设置回调函数。当消息发送到exchange后回调confirm方法。在方法中判断ack，如果为true，则发送成功，如果为false，则发送失败，需要处理。

设置ConnectionFactory的publisher-returns=“true” 开启退回模式。
使用rabbitTemplate.setReturnCallback设置退回函数，当消息从exchange路由到queue失败后，如果设置了rabbitTemplate.setMandatory(true)参数，则会将消息退回给producer。并执行回调函数returnedMessage。

5.2 消息的可靠消费

ack指的是Acknowledge,表示消费者收到消息的确认机制

有三种确认机制:

自动确认：acknowledge=“none”
手动确认：acknowledge=“manual”
根据异常情况确认：acknowledge=“auto”

其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

接下来就是代码演示

1)编写监听器

@Component
public class AckListener implements ChannelAwareMessageListener {


    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        //1.获取消息id
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try {
            //2.得到消息
            System.out.println("message=>"+message.getBody());
            //3.处理业务
            System.out.println("开始处理业务=>");

            int i = 1/0;
            //4.成功消费消息
            System.out.println("成功消费=>");

            //5.确认消息被消费
            channel.basicAck(deliveryTag,false);
        }catch (Exception e){
            //拒绝签收
             
            channel.basicNack(deliveryTag,false,true);
        }
    }
}

2)xml配置

3)运行结果

消费者一直在重试

4)小结

在rabbit:listener-container标签中设置acknowledge属性，设置ack方式 none：自动确认，manual：手动确认

如果在消费端没有出现异常，则调用channel.basicAck(deliveryTag,false);方法确认签收消息

如果出现异常，则在catch中调用 basicNack或 basicReject，拒绝消息，让MQ重新发送消息。

除了要考虑消息的可靠投递和可靠消费,还要保证broker的高可用的消息的持久化,所以要实现消息的高可靠性还要满足以下要求

1.持久化(交换机、队列、消息持久化)

2.生产者确认机制

3.消费者ack

4.Broker高可用

5.3 限流

限流是在消费者这一段进行限流操作,比如说队列中有5000消息,本来每秒钟可以拿2000,我们可以限流每秒钟拿1000消息进行处理

下边由代码演示

1)编写监听器

@Component
public class QosListener implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        System.out.println("处理消息"+new String(message.getBody()));

        //Thread.sleep(1000);

        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }
}

2)修改配置文件

3)运行结果

这里做个说明吧，如果需要限流必须把开启手动应答，prefetch可以根据具体情况设置，表示每次处理消息的数量,如果把 channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);忘记，则消费者拿到prefetch设置的值的消息然后就进入阻塞状态，因为broker是根据返回的ack进行消息分发的。

4)小结

配置 prefetch属性设置消费端一次拉取多少消息

消费端的确认模式一定为手动确认。acknowledge=“manual”

5.4 TTL

TTL 全称 Time To Live（存活时间/过期时间）。
当消息到达存活时间后，还没有被消费，会被自动清除。
RabbitMQ可以对消息设置过期时间，也可以对整个队列（Queue）设置过期时间。

接下来是代码演示

1)xml配置

2)发送消息

        //测试过期时间
        @Test
        public void testTTL(){
            for (int i = 0; i < 10; i++) {
                // 发送消息
                rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.test", "message ttl....");
            }
        }

3)运行结果

刚发送消息

10s过后

5.5 死信队列

死信队列，英文缩写：DLX 。Dead Letter Exchange（死信交换机），当消息成为Dead message后，可以被重新发送到另一个交换机，这个交换机就是DLX。

消息进入死信队列有三种情况:

(1)队列消息长度到达限制；

(2)消费者拒接消费消息，basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false

(3)原队列存在消息过期设置，消息到达超时时间未被消费；

接下来代码演示：

1)xml配置

2)发送消息

        @Test
        public void testDlx(){
            //测试过期时间
            rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.dead","死信队列测试1");
        }

3)效果展示

向正常队列发送一条消息

10s后私信队列多了一条记录

4)小结

1.死信交换机和死信队列和普通的没有区别
2.当消息成为死信后，如果该队列绑定了死信交换机，则消息会被死信交换机重新路由到死信队列

3.消息成为死信的三种情况：
(1). 队列消息长度到达限制；
(2). 消费者拒接消费消息，并且不重回队列；
(3). 原队列存在消息过期设置，消息到达超时时间未被消费；

5.6 延迟队列

延迟队列，即消息进入队列后不会立即被消费，只有到达指定时间后，才会被消费。

代码实现

1)xml配置

2)模拟下单

        //测试延迟队列
        @Test
        public void testDelay() throws InterruptedException {
            //1.发送订单消息。 将来是在订单系统中，下单成功后，发送消息
            rabbitTemplate.convertAndSend("order_exchange","order.msg","订单信息：id=1,time=2022年5月...");
            //2.打印倒计时10秒
            for (int i = 10; i > 0 ; i--) {
                System.out.println(i+"...");
                Thread.sleep(1000);
            }

        }

3)消费者监听死信队列

@Component
public class DelayListener  implements ChannelAwareMessageListener {


    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try {
            System.out.println(new String(message.getBody()));
            System.out.println("开始处理业务");
            System.out.println("判断订单支付状态");
            System.out.println("如果没有支付则取消订单回滚库存");

            channel.basicAck(deliveryTag,true);

        }catch (Exception e){
            System.out.println("出现异常,不重回队列");
            channel.basicNack(deliveryTag,true,false);
        }

    }
}

4)消费者配置

5)运行结果

1.生产者

2.消费者

5.7 消息幂等性

等性指一次和多次请求某一个资源，对于资源本身应该具有同样的结果。也就是说，其任意多次执行对资源本身所产生的影响均与一次执行的影响相同。在MQ中指，消费多条相同的消息，得到与消费该消息一次相同的结果。

这一点可以通过乐观锁机制来保证消息的幂等性问题。

5.8 消息积压

-消费者宕机积压
-消费者消费能力不足积压
-发送者发流量太大
解决方案:上线更多的消费者,进行正常消费上线专门的队列消费服务,将消息先批量取出来,记录数据库,再慢慢处理

RabbitMQ学习笔记

云平台相关栏目本月热门文章