【学习笔记】RabbitMQ

教学视频：B站尚硅谷RabbitMQ教学视频
代码资源：代码资源

文章目录

一、入门

1、基本概念2、基本功能3、MQ的分类4、MQ的选择5、四大核心概念 二、核心部分

1、 名词解释2、安装3、常用命令4、创建Java开发环境6、六大核心/六大模式

1）、简单模式--“HelloWorld”2）、工作模式--“WorkQueues”3）、消息应答模式（自创的，教学视频一团乱麻）4）、发布确认模式--“Publisher/confirm/is”5）、发布订阅模式--“Publish/Subscribe”6）、路由模式--“Routing”7）、主题模式--“Topics” 7、死信队列8、延迟队列 三、高级部分

1、高级发布确认2、幂等性问题3、优先级队列4、惰性队列 四、集群部分

一、入门 1、基本概念

2、基本功能

流量削峰

应用解耦

异步处理

3、MQ的分类

1、ActiveMQ
2、kafka
3、RoketMQ
RabbitMQ
4、MQ的选择

5、四大核心概念

二、核心部分 1、 名词解释

生产者、连接（含信道）、MQ（含交换机和队列）、连接（含信道）、消费者

2、安装

受不了直接临时学了点docker，秒装！！！

开启、运行容器：docker run -d --hostname my-rabbit --name rabbit -p 15672:15672 -p 5672:5672 rabbitmq安装管理界面（进入rabbitmq操作界面）：rabbitmq-plugins enable rabbitmq_management设置用户，添加权限
在阿里云中开启15672端口即可访问登录rabbitmq界面 3、常用命令

开机自启：chkconfig rabbitmq-server on启动服务：/sbin/service/rabbitmq-server start 4、创建Java开发环境

引入依赖

    
        
            org.apache.maven.plugins
            maven-compiler-plugin
            
                8
                8
            
        
    

    
        com.rabbitmq
        amqp-client
        5.8.0
    

    
        commons-io
        commons-io
        2.5
    

6、六大核心/六大模式

1）、简单模式–“HelloWorld”

生产者代码

public class Product {
    //队列名
    public static final String QUEUE_NAME = "hello";
    public static void main(String[] args) throws IOException, TimeoutException {
        //连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置ip、用户名、密码和端口号
        factory.setHost("139.224.46.237");
        factory.setUsername("admin");
        factory.setPassword("123");
        factory.setPort(5672);

        //创建连接
        Connection connection = factory.newConnection();
        //创建信道
        Channel channel = connection.createChannel();
        
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        
        channel.basicPublish("",QUEUE_NAME,null,"helloworld".getBytes());
        System.out.println("success！！");
    }
}

消费者代码

public class Consumer {
    public static final String QUEUENAME = "hello";

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("139.224.46.237");
        factory.setUsername("admin");
        factory.setPassword("123");
        factory.setPort(5672);

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        //声明消息接收成功后的回调
        DeliverCallback deliverCallback = (consumerTag,message)->{
            System.out.println(new String(message.getBody()));
        };

        //声明消息接受未成功的回调
        CancelCallback cancelCallback = ConsumerTag->{
            System.out.println("消息接收被中断");
        };

        
        channel.basicConsume(QUEUENAME,true, deliverCallback, cancelCallback);
    }
}

2）、工作模式–“WorkQueues”

基本概念

一个发布，多个轮循接收代码
略，大致与简单队列一模一样，多个接受者也是自动轮循接收的 3）、消息应答模式（自创的，教学视频一团乱麻）

基本概念

消息应答
自动应答
应答的方式

消息自动重新入队
代码
队列持久化
概念—略
即使RabbitMQ关机重启也不会消失
消息持久化
不公平分发


得到接受者的回应就会发送下一个消息，而没有回应的接受者则让其一直等待预期值

prefetchCount：可以理解为消息堆积上限，接受者提前处理完的可以继续接收，0就代表不管接受者的回应，完全按照轮循分发 4）、发布确认模式–“Publisher/confirm/is”

原理

使用
发布确认的三种方式

单个

批量

异步

//成功确认回调，消息标记，是否为批量确认
        ConfirmCallback ackCallback = (deliveryTag,mutiple)->{
            System.out.println(deliveryTag+"success");
        };
        //确认失败回调
        ConfirmCallback nackCallback = (deliveryTag,mutiple)->{
            System.out.println(deliveryTag+"fail");
        };
        //添加异步发布确认监听器
        channel.add/confirm/iListener(ackCallback,nackCallback);

        //发消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            channel.basicPublish("",queueName,null,String.valueOf(i).getBytes());
        }

处理异步未确认的消息

//支持高并发的hashMap,用于记录未确认的消息：消息标记为long，消息为string
        ConcurrentSkipListMap skipListMap = new ConcurrentSkipListMap<>();

        //成功确认回调，消息标记，是否为批量确认
        ConfirmCallback ackCallback = (deliveryTag,mutiple)->{
            //成功确认的消息就从hashmap中移除
            if (mutiple){
                //批量移除:获取从第一个到deliveryTag的所有k，v，clear清除所有kv
                ConcurrentNavigableMap headMap = skipListMap.headMap(deliveryTag);
                headMap.clear();
            }else{
                //单个移除
                skipListMap.remove(deliveryTag);
            }
            System.out.println(deliveryTag+"success");
        };
        //确认失败回调
        ConfirmCallback nackCallback = (deliveryTag,mutiple)->{
            System.out.println(deliveryTag+"fail");
        };
        //添加异步发布确认监听器
        channel.add/confirm/iListener(ackCallback,nackCallback);

        //发消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
            //先记录消息：序号，消息内容
            skipListMap.put(channel.getNextPublishSeqNo(),message);
        }

5）、发布订阅模式–“Publish/Subscribe”

基本概念

交换机概念

交换机类型

直接类型—路由模式；主题—主题模式；标题类型不常用了；扇出类型----发布模式；
无名类型—默认交换机

临时队列

绑定

可以通过不同的routingkey和bandingkey区分不同“队列组”（下图的routingkey错了）


在不同模式中的bindingkey和routingkey使用场景不同

扇出/发布订阅概念

使用/例子

Channel channel = RabbitMQUtils.getChannel();
        //发消息
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()){
            String message = scanner.nextLine();
            //交换机和routingkey，其他和消息体
            channel.basicPublish(EXCHANGE_NAME,"1",null,message.getBytes());
        }

		//接收消息，创建交换机，在发送端创建也可以
		//信道
        Channel channel = RabbitMQUtils.getChannel();
        //声明一个交换机:交换机名，交换机类型
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        //临时队列
        String tempQueueName = channel.queueDeclare().getQueue();
        //绑定交换机与队列：队列名，交换机名，routingkey,可绑定多个routingkey
        channel.queueBind(tempQueueName,EXCHANGE_NAME,"1");
        //接收消息
        //声明消息接收成功后的回调
        DeliverCallback deliverCallback = (consumerTag, message)->{
            System.out.println(new String(message.getBody()));
        };

        //声明消息接受未成功的回调
        CancelCallback cancelCallback = ConsumerTag->{
            System.out.println("消息接收被中断");
        };

        channel.basicConsume(tempQueueName,true, deliverCallback, cancelCallback);

6）、路由模式–“Routing”

基本概念
需要确认routingkey与bindingkey才会将消息发布到对应的队列
例子
与发布订阅模式一致，只不过会按照routingkey和bindingkey分发了而已 7）、主题模式–“Topics”

基本概念

之前的模式无法解决的问题
routingkey的命名规范

例子
略 7、死信队列

概念
产生
例子
代码（ttl过期、队列达到最大长度、消息被拒）

//生产者
public static final String NORMAL_EXCHANGE = "normal_exchange";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtils.getChannel();
        //设置消息过期时间10s
//        AMQP.BasicProperties properties = new AMQP.BasicProperties()
//                .builder().expiration("10000").build();

        //发消息
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()){
            String message = scanner.nextLine();
            //交换机和routingkey，其他和消息体
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null,message.getBytes());
        }
    }

//正常队列消费者
//交换机名
    public static final String NORMAL_EXCHANGE = "normal_exchange";
    public static final String DEAD_EXCHANGE = "dead_exchange";
    //队列名
    public static final String NORMAL_QUEUE = "normal_queue";
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtils.getChannel();
        //交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);

        //队列:在普通队列绑定死信交换机
        Map arguments = new HashMap<>();
        //设置过期时间:也可以由发送方设置时间
        //arguments.put("x-message-ttl",10000);
        //绑定死信交换机
        arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        //设置死信routingkey
        arguments.put("x-dead-letter-routing-key","lisi");
        //设置队列最大长度:超过最大长度会把最先到达队列的放入死信队列（先到先去死信）
//        arguments.put("x-max-length",6);

        channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);

        //绑定交换机与队列
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");


        //接收消息
        //声明消息接收成功后的回调
        DeliverCallback deliverCallback = (consumerTag, message)->{
            //检验消息
            String msg = new String(message.getBody(), StandardCharsets.UTF_8);
            if (msg.equals("1")){
                System.out.println("reject!!!");
                //拒绝消息消息标签，是否放回队列
                channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
            }else{
                //接受消息:消息标记，是否批量接受
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
                System.out.println(new String(message.getBody()));
            }
        };

        //声明消息接受未成功的回调
        CancelCallback cancelCallback = ConsumerTag->{
            System.out.println("消息接收被中断");
        };
        //开启手动应答：false
        channel.basicConsume(NORMAL_QUEUE,false,deliverCallback,cancelCallback);

    }

//死信队列消费者
public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtils.getChannel();

        //接收消息
        //声明消息接收成功后的回调
        DeliverCallback deliverCallback = (consumerTag, message)->{
            System.out.println(new String(message.getBody()));
        };

        //声明消息接受未成功的回调
        CancelCallback cancelCallback = ConsumerTag->{
            System.out.println("消息接收被中断");
        };

        channel.basicConsume(DEAD_QUEUE,true,deliverCallback,cancelCallback);

    }

8、延迟队列

概念
近似于ttl过期的死信队列
使用场景

-例子(随便整合springboot)

代码结构图
代码
略优化

存在问题

用插件解决p66

基于插件的不再使用死信队列
解决的代码（装不了插件，这里的代码先不敲了）

配置文件




product、controller
三、高级部分 1、高级发布确认

问题
解决方案一：回调反馈

重点：回调接口

simple是同步确认方式，发一个就等待确认，效率很慢

@Component
public class /confirm/iCallBack implements RabbitTemplate./confirm/iCallback {
    //由于/confirm/iCallback是内部类，需要注入，才能让内部调用我们重写的类
    @Autowired
    RabbitTemplate rabbitTemplate;

    //这个是JSR250定义的注解，被注解的方法将在bean创建并且赋值完成，在执行初始化方法之前调用，
    // 原理：后置处理器起的作用
    @PostConstruct
    public void init(){
        //将父类的接口换成我们重写的这个
        rabbitTemplate.set/confirm/iCallback(this);
    }

    
    @Override
    public void /confirm/i(CorrelationData correlationData, boolean b, String s) {
        //获取交换机id
        String exchangeId = correlationData!=null?correlationData.getId():"";
        if (b) {
            System.out.println(exchangeId + "已经收到");
        } else {
            System.out.println(exchangeId+"未收到:"+s);
        }


    }
}

但是这只是交换机的反馈，队列消息的丢失并不能正常反馈

回退消息

路有即是从交换机发往队列的过程

配置文件中开启：spring.rabbitmq.publisher-returns=true重写ReturnCallback，注入（可以与之前的写在一起）


解决方法二：备份交换机

代码架构图

在/confirm/iExchange的交换机声明中绑定备份交换机

2、幂等性问题

概念

解决方法

3、优先级队列

使用场景
使用

//优先级队列
        Map arguments = new HashMap<>();
        //优先级范围0-255，这里为0-10，不要设置过大cpu会有影响
        arguments.put("x-max-priority",10);
        
        channel.queueDeclare(QUEUE_NAME,true,false,false,arguments);

        //优先级队列：发消息
        for (int i = 0; i < 10; i++) {
            if(i==5){
                //增加第五条的优先级，其他消息不变
                AMQP.BasicProperties properties =
                        new AMQP.BasicProperties().builder().priority(10).build();
                channel.basicPublish("",QUEUE_NAME,properties,
                        String.valueOf(i).getBytes());
            }else{
                channel.basicPublish("",QUEUE_NAME,null,
                        String.valueOf(i).getBytes());
            }
        }

4、惰性队列

使用队列
两种模式
代码
略 四、集群部分

概念
略搭建步骤




镜像队列

概念
搭建

docker的搭建方式
docker的rabbitmq搭建集群
勘误：
1、加入节点到集群中进入后台的命令中的名字应该与运行容器的名字一样，去掉my加上mq
2、加入节点的名字也是错的去掉mq本节内容

pattern:队列命名符合此处的正则表达式才会进行备份
ha-mode:备份模式–指令模式
ha-parass：指令----主从在内备份两份
ha-sync-mode：同步模式—自动效果：如果节点一宕机了，则会使用备份节点中的队列，另外再找一个备份节点，已达到备份数量为2的策略，以此类推。因此只要集群中仍有一个节点，就不会丢失信息 利用nginx或haproxy、keepalive进行负载均衡高可用
略

---

联邦交换机（Federation exchange）

概念
个地区有多个服务器，需要让用户按照地区进行访问，因此如要用联邦交换机来同步各个服务器之间的信息
信息只能从上游（即被联邦的一方）同步到下游（联邦别人的一方），如果要实现双向的信息同步就要进行双向的联邦，操作同理使用

联邦队列

使用
Shovel

概念

使用