RabbitMQ实战教程

MQ引言

什么是MQ

MQ(Message Quene)：翻译为消息队列，通过典型的生产者和消费者模型生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入轻松的实现系统间解耦。别名为消息中间件，通过利用高效可靠的消息传递机制进行平台无关的数据交流，并基于数据通信来进行分布式系统的集成。

MQ有哪些

当今市面上有很多主流的消息中间件，如老牌的ActiveMQ、RabbitMQ，炙手可热的Kafka，阿里巴巴自主开发RocketMQ等。

不同MQ特点

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

RabbitMQ

RabbitMQ引言

基于 AMOP协议，erlang语言开发，是部署最广泛的开源消息中间件是最受欢迎的开源消息中间件之一。

AMQP协议

AMQP（advanced message queuing protocol）在2003年时被提出，最早用于解决金融领不同平台之间的消息传递交互问题。顾名思义，AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。这使得实现了AMQP的
provider天然性就是跨平台的。以下是AMQP协议模型:

QQ截图20220217190825

安装

偷懒，直接使用yum在线安装

yum在线安装

erlang

curl -s https：//packagecloud.io/install/repositories/rabbitmq/erlang/script.rpm.sh | sudo bash

```
yum install erlang -y
```

rabbitmq

curl -s https：//packagecloud.io/install/repositories/rabbitmq/rabbitmq-server/script.rpm.sh | sudo bash

```
yum install rabbitmq-server -y
```

rpm安装

erlang下载地址

rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm

rabbitmq下载地址

yum install -y rabbitmq-server-3.7.18-1.el7.noarch.rpm

开启管理插件

rabbitmq-plugins enable rabbitmq_management

配置

rabbitMq有三个配置文件，分别为主配置文件（rabbitmq.conf），Erlang术语格式配置文件(advanced.config)、环境变量配置文件(rabbitmq-env.conf)。

配置远程访问

在/etc/rabbitmq创建rabbitmq.conf，添加如下
```
loopback_users.guest = false
```

常见命令

服务启动相关命令

systemctl start|restart|stop|status rabbitmq-server

rabbitmq基础命令
```
rabbitmqctl  help
```

插件管理命令

rabbitmq-plugins enable|list|disable|set

WEB界面概览

QQ截图20220218191052

常见概念
- Server(broker)：接受客户端连接，实现AMQP消息队列和路由功能的进程
- Virtual Host：虚拟主机，类似于权限控制组，一个Virtual Host里面可以有若干个Exchange和Queue，但是权限控制的最小粒度是Virtual Host
- Connections：无论生产者还是消费者，都需要与RabbitMQ建立连接后才能完成消息的生产和消费，在这里可以查看连接情况。对于RabbitMQ而言，其实就是一个位于客户端和Broker之间的TCP连接
- Channels：通道，建立连接后会形成通道，消息的传递获取依赖于通道
- Exchanges：交换机，用来实现消息的路由。接收生产者发送的消息，并根据一定规则将消息路由给服务器中的队列
- Message Queues：消息队列，消息存放在队列中，等待消费，消费后被移除队列
- Message：由Header和Body组成。Header是由生产者添加的各种属性的集合，包括Message是否被持久化、由哪个Message Queue接受、优先级是多少等。而Body是真正需要传输的APP数据。
创建用户
- Tags选项用户可选类型
  - Admin：超级管理员，可登录管理控制台，可查看所有信息，并且可以对用户，策略(policy)进行操作\
  - Monitoring：监控者，可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)
  - Policymaker：策略制定者，可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)
  - Management：普通管理者，仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理
  - 其他：无法登陆管理控制台，通常就是普通的生产者和消费者。
创建虚拟主机

为了让各个用户可以互不干扰的工作，RabbitMQ添加了虚拟主机（Virtual Hosts）的概念。其实就是一个独立的访问路径，不同用户使用不同路径，各自有自己的队列、交换机，互相不会影响。
虚拟主机设置权限

点击虚拟主机下的分配权限（默认配置、读、写）
拓展topic permission

Topic Permissions，这是 RabbitMQ3.7 开始的一个新功能，可以针对某一个 topic exchange 设置权限，主要针对 STOMP 或者 MQTT 协议，日常 Java 开发用上这个配置的机会很少。如果用户不设置的话，相应的 topic exchange 也总是有权限的

Hello RabbitMQ

消息模型介绍

QQ截图20220218214841

依赖

              <dependency>
                    <groupId>com.rabbitmq</groupId>
                    <artifactId>amqp-client</artifactId>
                    <version>5.14.2</version>
                </dependency>

直连模型

QQ截图20220220220817

P：生产者，要发送消息的程序

C：消费者，等待接收消息的程序

Queue：消息队列，图中红色部分。实质上是一个大的消息缓冲区。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。遵循先进先出原则

工具类RabbitmqUtils

public class RabbitmqUtils {
    private static final ConnectionFactory connectionFactory;
    static {
        //创建连接mq的连接工厂对象，重量级对象，类加载时创建一次即可
        connectionFactory = new ConnectionFactory();
        //设置连接rabbitmq的主机
        connectionFactory.setHost("47.93.254.136");
        //设置端口号
        connectionFactory.setPort(5672);
        //设置连接的虚拟主机
        connectionFactory.setVirtualHost("/ems");
        //设置访问虚拟主机的用户名和密码
        connectionFactory.setUsername("ems");
        connectionFactory.setPassword("ems");
    }
    //获取连接对象
    public static Connection getConnection(){
        try {
            return connectionFactory.newConnection();
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
        return null;
    }
    //关闭通道和连接
    public static void close(Channel channel, Connection connection){
        if(channel != null){
            try {
                channel.close();
            } catch (IOException | TimeoutException e) {
                e.printStackTrace();
            }
        }
        if(connection != null){
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

生产者Provider

public class Provider {
    @Test
    public void testSendMessage() {
        Connection connection = null;
        Channel channel = null;
        try {
            //获取连接对象
            connection = RabbitmqUtils.getConnection();
            //获取通道
            if (connection != null) {
                channel = connection.createChannel();
                //通道绑定对应消息队列
                /*
                 * 参数1 queue：队列名称(不存在自动创建)
                 * 参数2 durable：用来定义队列特性是否需要持久化(为true该队列将在服务器重启后保留下来，持久化到硬盘中)
                 * 参数3 exclusive：是否独占队列(为true仅限此连接)
                 * 参数4 autoDelete：是否在消费完成后自动删除队列
                 * 参数5 arguments：队列的其他属性（构造参数）
                 * */
                channel.queueDeclare("hello", true, false, false, null);
                //发布消息
                /*
                 * 参数1 exchange：要将消息发布到的交换机
                 * 餐数2 routingKey：路由键，指定队列
                 * 参数3 props：消息的其他属性
                 * 参数4 body：消息具体内容
                 * */
                channel.basicPublish("", "hello", MessageProperties.PERSISTENT_TEXT_PLAIN, "hello rabbitmq".getBytes());
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            System.out.println("生产者发布消息完成......");
            RabbitmqUtils.close(channel, connection);
        }
    }
}

消费者Consumer

public class Consumer {
    public static void main(String[] args) {
        Connection connection;
        Channel channel = null;
        try {
            connection = RabbitmqUtils.getConnection();
            if (connection != null) {
                channel = connection.createChannel();
                channel.queueDeclare("hello", true, false, false, null);
            }
            //消费回调接口
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
                //获取消息并且处理。此方法类似于事件监听，有消息时会被自动调用
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                    System.out.println("message:" + new String(body));  //body即消息体
                }
            };
            //消费消息
            /*
             * 参数1 queue：队列名称
             * 参数2 autoAck：开启消息的自动确认机制
             * 参数3 Consumer callback：消费时的回调接口
             * */
            channel.basicConsume("hello", true, defaultConsumer);
        } catch (
                IOException e) {
            e.printStackTrace();
        } finally {
            System.out.println("消费者启动完成......");
        }
    }
}

细节（持久化设置）

队列持久化

//第二个参数改为true，
channel.queueDeclare("hello",true,false,false,null);

消息持久化

//发布消息时添加消息的属性设置，第三个参数改为MessageProperties.PERSISTENT_TEXT_PLAIN
channel.basicPublish("","hello", MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());

工作队列模型

Work queues，也被称为（Task queues），任务模型。当消息处理比较耗时的时候，可能生产消息的速度会
远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。此时就可以使用work 模型：
让多个消费者绑定到一个队列，共同消费队列中的消息。队列中的消息一旦消费，就会消失，因此任务是不会被重复执行的。

QQ截图20220220220932

P：生产者：任务的发布者
C1：消费者-1，领取任务并且完成任务，假设完成速度较慢
C2：消费者-2：领取任务并完成任务，假设完成速度快

生产者Provider

public class Provider {
    @Test
    public void testSendMessage() {
        Connection connection = null;
        Channel channel = null;
        try {
            connection = RabbitmqUtils.getConnection();
            if (connection != null) {
                channel = connection.createChannel();
                channel.queueDeclare("work", true, false, false, null);
                for (int i = 0; i < 100; i++) {
                    channel.basicPublish("", "work", MessageProperties.PERSISTENT_TEXT_PLAIN, ("work queue the " + i + " message").getBytes());
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            System.out.println("生产者发布消息完成......");
            RabbitmqUtils.close(channel, connection);
        }
    }
}

消费者Consumer

public class Consumer {
    public static void main(String[] args) {
        Connection connection;
        Channel channel = null;
        try {
            connection = RabbitmqUtils.getConnection();
            if (connection != null) {
                channel = connection.createChannel();
                //设置服务器一次请求将传递的最大邮件数
                channel.basicQos(1);
                channel.queueDeclare("work", true, false, false, null);
            }
            Channel finalChannel = channel;
            DefaultConsumer defaultConsumer = new DefaultConsumer(finalChannel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                    System.out.println("consumer message:" + new String(body));
                    try {
                        //设置手动应答
                        /*
                         * 参数1: 手动确认消息标识
                         * 参数2: false 每次确认一个
                         */
                        finalChannel.basicAck(envelope.getDeliveryTag(),false);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            };
            channel.basicConsume("work", false, defaultConsumer);
        } catch (
                IOException e) {
            e.printStackTrace();
        } finally {
            System.out.println("消费者启动完成......");
        }
    }
}

消费者ConsumerTwo

public class ConsumerTwo {
    public static void main(String[] args) {
        Connection connection;
        Channel channel = null;
        try {
            connection = RabbitmqUtils.getConnection();
            if (connection != null) {
                channel = connection.createChannel();
                channel.basicQos(1);
                channel.queueDeclare("work", true, false, false, null);
            }
            Channel finalChannel = channel;
            DefaultConsumer defaultConsumer = new DefaultConsumer(finalChannel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                    System.out.println("consumer two message:" + new String(body));
                    try {
                        TimeUnit.SECONDS.sleep(1L);
                        finalChannel.basicAck(envelope.getDeliveryTag(),false);
                    } catch (InterruptedException | IOException e) {
                        e.printStackTrace();
                    }
                }
            };
            channel.basicConsume("work", false, defaultConsumer);
        } catch (
                IOException e) {
            e.printStackTrace();
        } finally {
            System.out.println("消费者启动完成......");
        }
    }
}

细节（设置能者多劳）

在默认的Work queues中，多个消费者之间是轮询且一次性分配的。为了实现能者多劳，我们要修改原有的消息确认机制，改为手动确认，并且限制消费者每次只能拿到一个消息。

手动确认

//第二个参数由其true改为false，关闭自动确认
channel.basicConsume("work", false, defaultConsumer);
//设置手动确认
finalChannel.basicAck(envelope.getDeliveryTag(),false);

消息限制

//参数一 服务器将提供的最大内容量
//参数二 服务器将传递的最大邮件数
//参数三 全局参数，是否应用到整个频道
channel.basicQos(0,1,false);

发布订阅模型

QQ截图20220221193843

在广播模式下，消息发送流程是这样的：

可以有多个消费者

每个消费者有自己的queue（队列）

每个队列都要绑定到Exchange（交换机）

生产者发送的消息，只能发送到交换机，

交换机来决定要发给哪个队列，生产者无法决定。

交换机把消息发送给绑定过的所有队列

队列的消费者都能拿到消息。实现一条消息被多个消费者消费

生产者Provider

public class Provider {
    @Test
    public void testSendMessage() {
        Connection connection = null;
        Channel channel = null;
        try {
            connection = RabbitmqUtils.getConnection();
            if (connection != null) {
                channel = connection.createChannel();
                //将通道声明指定交换机
                /*
                 * 参数1: 交换机名称
                 * 参数2: 交换机类型  fanout 广播类型
                 */
                channel.exchangeDeclare("publish","fanout");
                channel.basicPublish("publish", "", MessageProperties.PERSISTENT_TEXT_PLAIN, "publish&subscribe type message".getBytes());
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            System.out.println("生产者发布消息完成......");
            RabbitmqUtils.close(channel, connection);
        }
    }
}

消费者Consumer

public class Consumer {
    public static void main(String[] args) {
        Connection connection;
        Channel channel;
        try {
            connection = RabbitmqUtils.getConnection();
            if (connection != null) {
                channel = connection.createChannel();
                // 频道绑定交换机
                // channel.exchangeDeclare("publish", "fanout");
                // 创建临时队列
                String queue = channel.queueDeclare().getQueue();
                // 队列绑定交换机
                /*
                 *  param1：destination，目的地，队列的名字
                 *  param2：source，资源，交换机的名字
                 *  param3：routingKey，路由键（目前没有用到routingKey，填 "" 即可）
                 */
                channel.queueBind(queue, "publish", "");
                DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                        System.out.println("consumer message:" + new String(body));
                    }
                };
                channel.basicConsume(queue, true, defaultConsumer);
            }
        } catch (
                IOException e) {
            e.printStackTrace();
        } finally {
            System.out.println("消费者启动完成......");
        }
    }
}

路由模型

在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

在Direct模型下：队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）消息的发送方在向 Exchange发送消息时，也必须指定消息的 RoutingKey。

Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息

QQ截图20220221223243

P：生产者，向Exchange发送消息，发送消息时，
会指定一个routing key。
X：Exchange（交换机），接收生产者的消息，
然后把消息递交给与routing key完全匹配的队列
C1：消费者，其所在队列指定了需要routing key 为 error 的消息
C2：消费者，其所在队列指定了需要routing key 为 info、
error、warning 的消息

生产者Provider

public class Provider {
    @Test
    public void testSendMessage() {
        Connection connection = null;
        Channel channel = null;
        try {
            connection = RabbitmqUtils.getConnection();
            if (connection != null) {
                channel = connection.createChannel();
                //将通道声明指定交换机
                /*
                 * 参数1: 交换机名称
                 * 参数2: 交换机类型  direct 直连类型
                 */
                channel.exchangeDeclare("routing", "direct");
                String routingKey = "error";
                channel.basicPublish("routing", routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, ("direct type message routingKey=" + routingKey).getBytes());
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            System.out.println("生产者发布消息完成......");
            RabbitmqUtils.close(channel, connection);
        }
    }
}

消费者Consumer

public class Consumer {
    public static void main(String[] args) {
        Connection connection;
        Channel channel;
        try {
            connection = RabbitmqUtils.getConnection();
            if (connection != null) {
                channel = connection.createChannel();
                channel.exchangeDeclare("routing", "direct");
                String queue = channel.queueDeclare().getQueue();
                //基于route key绑定队列和交换机
                channel.queueBind(queue, "routing", "error");
                DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                        System.out.println("consumer message:" + new String(body));
                    }
                };
                channel.basicConsume(queue, true, defaultConsumer);
            }
        } catch (
                IOException e) {
            e.printStackTrace();
        } finally {
            System.out.println("消费者启动完成......");
        }
    }
}

细节
- 来自弹幕
  - 生产者只需要设置交换机和路由key，消费者只设置队列名；交换机，路由key和队列的关系直接在队列的后台配置

话题模型

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key的时候使用通配符！这种模型Routingkey一般都是由一个或多个单词组成，多个单词之间以”.”分割。

QQ截图20220221231242

通配符

* (star) can substitute for exactly one word. 匹配不多不少恰好1个词

# (hash) can substitute for zero or more words. 匹配零个、一个或多个词

生产者Provider

public class Provider {
    @Test
    public void testSendMessage() {
        Connection connection = null;
        Channel channel = null;
        try {
            connection = RabbitmqUtils.getConnection();
            if (connection != null) {
                channel = connection.createChannel();
                //将通道声明指定交换机
                /*
                 * 参数1: 交换机名称
                 * 参数2: 交换机类型  topic 动态路由类型
                 */
                channel.exchangeDeclare("topics", "topic");
                String routingKey = "user.save";
                channel.basicPublish("topics", routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, ("topic type message routingKey=" + routingKey).getBytes());
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            System.out.println("生产者发布消息完成......");
            RabbitmqUtils.close(channel, connection);
        }
    }
}

消费者Consumer

public class Consumer {
    public static void main(String[] args) {
        Connection connection;
        Channel channel;
        try {
            connection = RabbitmqUtils.getConnection();
            if (connection != null) {
                channel = connection.createChannel();
                channel.exchangeDeclare("topics", "topic");
                String queue = channel.queueDeclare().getQueue();
                channel.queueBind(queue, "topics", "user.*");
                DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                        System.out.println("consumer message:" + new String(body));
                    }
                };
                channel.basicConsume(queue, true, defaultConsumer);
            }
        } catch (
                IOException e) {
            e.printStackTrace();
        } finally {
            System.out.println("消费者启动完成......");
        }
    }
}

远程过程调用模型

QQ截图20220221234455

服务端

public class RPCServer {
    private static final String RPC_QUEUE_NAME = "rpc_queue";
    private static int fib(int n) {
        if (n == 0) return 0;
        if (n == 1) return 1;
        return fib(n - 1) + fib(n - 2);
    }
    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
            channel.queuePurge(RPC_QUEUE_NAME);
            channel.basicQos(1);
            System.out.println(" [x] Awaiting RPC requests");
            Object monitor = new Object();
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                AMQP.BasicProperties replyProps = new AMQP.BasicProperties
                        .Builder()
                        .correlationId(delivery.getProperties().getCorrelationId())
                        .build();
                String response = "";
                try {
                    String message = new String(delivery.getBody(), "UTF-8");
                    int n = Integer.parseInt(message);
                    System.out.println(" [.] fib(" + message + ")");
                    response += fib(n);
                } catch (RuntimeException e) {
                    System.out.println(" [.] " + e.toString());
                } finally {
                    channel.basicPublish("", delivery.getProperties().getReplyTo(), replyProps, response.getBytes("UTF-8"));
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    // RabbitMq consumer worker thread notifies the RPC server owner thread
                    synchronized (monitor) {
                        monitor.notify();
                    }
                }
            };
            channel.basicConsume(RPC_QUEUE_NAME, false, deliverCallback, (consumerTag -> { }));
            // Wait and be prepared to consume the message from RPC client.
            while (true) {
                synchronized (monitor) {
                    try {
                        monitor.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
}

客户端

public class RPCClient implements AutoCloseable {
    private Connection connection;
    private Channel channel;
    private String requestQueueName = "rpc_queue";
    public RPCClient() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        connection = factory.newConnection();
        channel = connection.createChannel();
    }
    public static void main(String[] argv) {
        try (RPCClient fibonacciRpc = new RPCClient()) {
            for (int i = 0; i < 32; i++) {
                String i_str = Integer.toString(i);
                System.out.println(" [x] Requesting fib(" + i_str + ")");
                String response = fibonacciRpc.call(i_str);
                System.out.println(" [.] Got '" + response + "'");
            }
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
    }
    public String call(String message) throws IOException, InterruptedException {
        final String corrId = UUID.randomUUID().toString();
        String replyQueueName = channel.queueDeclare().getQueue();
        AMQP.BasicProperties props = new AMQP.BasicProperties
                .Builder()
                .correlationId(corrId)
                .replyTo(replyQueueName)
                .build();
        channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));
        final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
        String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> {
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                response.offer(new String(delivery.getBody(), "UTF-8"));
            }
        }, consumerTag -> {
        });
        String result = response.take();
        channel.basicCancel(ctag);
        return result;
    }
    public void close() throws IOException {
        connection.close();
    }
   }

发布者确认

在频道上启用发布服务器确认

Channel channel = connection.createChannel();
channel.confirmSelect();

单独发布消息

使用 Channel.waitForConfirmsOrDie（long）方法等待其确认。一旦确认消息，该方法将立即返回。如果消息未在超时内得到确认，或者消息是 nack-ed（意味着代理由于某种原因无法处理它），则该方法将引发异常。异常的处理通常包括记录错误消息和/或重试发送消息。
```
while (thereAreMessagesToPublish()) {
    channel.basicPublish(exchange, queue, properties, body);
    channel.waitForConfirmsOrDie(5_000);
}
```
批量发布消息

与等待单个消息的确认相比，等待一批消息得到确认可大大提高吞吐量（使用远程 RabbitMQ 节点时最多可提高 20-30 次）。一个缺点是，如果发生故障，我们不知道到底出了什么问题，因此我们可能必须在内存中保留整个批次以记录有意义的内容或重新发布消息。并且此解决方案仍然是同步的，因此它会阻止消息的发布。
```
int batchSize = 100;
int outstandingMessageCount = 0;
while (thereAreMessagesToPublish()) {
    channel.basicPublish(exchange, queue, properties, body);
    outstandingMessageCount++;
    if (outstandingMessageCount == batchSize) {
        ch.waitForConfirmsOrDie(5_000);
        outstandingMessageCount = 0;
    }
}
if (outstandingMessageCount > 0) {
    ch.waitForConfirmsOrDie(5_000);
}
```
异步处理发布服务器确认
有 2 个回调：一个用于已确认的消息，另一个用于 nack 消息（代理可视为丢失的消息）。每个回调有 2 个参数：
- 序列号：标识已确认或已确认消息的数字。我们很快就会看到如何将其与已发布的消息相关联。
- multiple：这是一个布尔值。如果为 false，则仅确认/nack-ed 一条消息;如果为 true，则确认/nack-ed 所有序列号较低或相等的消息。
```
Channel channel = connection.createChannel();
channel.confirmSelect();
channel.addConfirmListener((sequenceNumber, multiple) -> {
    // code when message is confirmed
}, (sequenceNumber, multiple) -> {
    // code when message is nack-ed
});
```
总结

在某些应用程序中，确保已发布的消息发送到代理可能是必不可少的。发布者确认是 RabbitMQ 功能，有助于满足此要求。发布者确认本质上是异步的，但也可以同步处理它们。没有明确的方法来实现发布者确认，这通常归结为应用程序和整个系统中的约束。典型的技术是：
- 单独发布消息，同步等待确认：简单，但吞吐量非常有限。
- 批量发布消息，同步等待批处理的确认：简单，合理的吞吐量，但很难推断出何时出现问题。
- 异步处理：最佳性能和资源使用，在发生错误时控制良好，但可以涉及正确实现。

Springboot整合

初始环境

RabbitTemplate用来简化操作，使用时候直接在项目中注入即可使用。

pom

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

application.yml

spring:
  application:
    name: springboot_rabbitmq
  rabbitmq:
    host: 10.15.0.9
    port: 5672
    username: ems
    password: 123
    virtual-host: /ems

直连模型

生产端没有指定交换机，只指定routingKey和Object。
消费方产生hello队列，放在默认的交换机(AMQP default)上。
而默认的交换机有一个特点，只要你的routerKey的名字与这个交换机的队列有相同的名字，他就会自动路由上。

生产者

	@Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    void helloWorld() {
        rabbitTemplate.convertAndSend("hello","hello world");
    }

消费者

@Component
@RabbitListener(queuesToDeclare = @Queue(value = "hello"))
public class Consumer {
    @RabbitHandler
    public void receive(String message) {
        System.out.println("message = " + message);
    }
}

工作队列模型

springboot2.6.3，默认能者多劳

生产者

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    void work() {
        for (int i = 0; i < 10; i++) {
            rabbitTemplate.convertAndSend("work", "hello work [" + i + "]");
        }
    }

消费者

@Component
public class TestConsumer {
    @RabbitListener(queuesToDeclare = @Queue(value = "work"))
    public void receive(String message) {
        System.out.println("message = " + message);
    }

    @RabbitListener(queuesToDeclare = @Queue(value = "work"))
    public void receiveTwo(String message) {
        System.out.println("message two = " + message);
    }
}

发布订阅模型

生产者

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    void fanout() {
        rabbitTemplate.convertAndSend("logs","","hello fanout");
    }

消费者

@Component
public class PublishConsumer {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue, // 创建临时队列
            exchange = @Exchange(name = "logs", type = "fanout")
    ))
    public void receive(String message) {
        System.out.println("message = " + message);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue, //创建临时队列
            exchange = @Exchange(name = "logs", type = "fanout")  //绑定交换机类型
    ))
    public void receiveTwo(String message) {
        System.out.println("message two = " + message);
    }
}

路由模型

生产者

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    void routing() {
        rabbitTemplate.convertAndSend("routing", "info", "hello routing");
    }

消费者

@Component
public class RoutingConsumer {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue, // 创建临时队列
            exchange = @Exchange(name = "directs", type = "direct"),
            key = {"info"} // 路由key
    ))
    public void receive(String message) {
        System.out.println("message = " + message);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue, //创建临时队列
            exchange = @Exchange(name = "directs", type = "direct"),  //绑定交换机类型
            key = {"info", "error"}// 路由key
    ))
    public void receiveTwo(String message) {
        System.out.println("message two = " + message);
    }
}

话题模型

生产者

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    void topic() {
        rabbitTemplate.convertAndSend("topics", "user.save.findAll", "user.save.findAll 的消息");
    }

消费者

@Component
public class TopicConsumer {
    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue,
                    key = {"user.*"},
                    exchange = @Exchange(type = "topic", name = "topics")
            )
    })
    public void receive1(String message) {
        System.out.println("message = " + message);
    }

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue,
                    key = {"user.#"},
                    exchange = @Exchange(type = "topic", name = "topics")
            )
    })
    public void receive2(String message) {
        System.out.println("message two = " + message);
    }
}

MQ的应用场景

异步处理

场景说明：用户注册后，需要发注册邮件和注册短信,传统的做法有两种

串行的方式

并行的方式

串行方式: 
将注册信息写入数据库后,发送注册邮件,再发送注册短信,以上三个任务全部完成后才返回给客户端。 这有一个问题是,邮件,
短信并不是必须的,它只是一个通知,而这种做法让客户端等待没有必要等待的东西.

QQ截图20220223184604

并行方式: 
将注册信息写入数据库后,发送邮件的同时,发送短信,以上三个任务完成后,返回给客户端,并行的方式能提高处理的时间。

QQ截图20220223184650

消息队列:
假设三个业务节点分别使用50ms,串行方式使用时间150ms,并行使用时间100ms。虽然并行已经提高的处理时间,但是,前面说过,邮件和短信对我正常的使用网站没有任何影响，客户端没有必要等着其发送完成才显示注册成功,应该是写入数据库后就返回. 引入消息队列后，不是必须的业务逻辑异步处理

QQ截图20220223184918

应用解耦

场景：双11是购物狂节,用户下单后,订单系统需要通知库存系统

传统做法
订单系统调用库存系统的接口. 缺点: 当库存系统出现故障时,订单就会失败。订单系统和库存系统高耦合.

QQ截图20220223191033

引入消息队列
订单系统:用户下单后,订单系统完成持久化处理,将消息写入消息队列,返回用户订单下单成功。
库存系统:订阅下单的消息,获取下单消息,进行库操作。就算库存系统出现故障,消息队列也能保证消息的可靠投递,不会导致消息丢失.

QQ截图20220223191142

流量削峰

场景: 秒杀活动，一般会因为流量过大，导致应用挂掉,为了解决这个问题，一般在应用前端加入消息队列。

作用:
1.可以控制活动人数，超过此一定阀值的订单直接丢弃
2.可以缓解短时间的高流量压垮应用(应用程序按自己的最大处理能力获取订单)

操作
1.用户的请求,服务器收到之后,首先写入消息队列,加入消息队列长度,超过最大值,则直接抛弃用户请求或跳转到错误页面.  
2.秒杀业务根据消息队列中的请求信息，再做后续处理.

QQ截图20220223191446

RabbitMQ集群

主备集群

RabbitMQ代理操作所需的所有数据/状态都会在所有节点上复制。消息队列是一个例外，默认情况下，消息队列位于一个节点上，但所有节点都可以看到和访问它们。

主节点提供读写，从节点不提供读写服务，只是负责提供备份服务,备份节点的主要功能是在主节点宕机时，完成自动切换从-->主，未持久化时，主节点宕机会丢失信息。

QQ截图20220223223535

集群搭建

# 0.集群规划
	node1: 10.15.0.3  mq1  master 主节点
	node2: 10.15.0.4  mq2  repl1  副本节点
	node3: 10.15.0.5  mq3  repl2  副本节点

# 1.克隆三台机器主机名和ip映射
	vim /etc/hosts加入:
		10.15.0.3 mq1
    	10.15.0.4 mq2
    	10.15.0.5 mq3
	node1: vim /etc/hostname 加入:  mq1
	node2: vim /etc/hostname 加入:  mq2
	node3: vim /etc/hostname 加入:  mq3

# 2.三个机器安装rabbitmq,并同步cookie文件,在node1上执行:
	scp /var/lib/rabbitmq/.erlang.cookie root@mq2:/var/lib/rabbitmq/
	scp /var/lib/rabbitmq/.erlang.cookie root@mq3:/var/lib/rabbitmq/

# 3.查看cookie是否一致:
	node1: cat /var/lib/rabbitmq/.erlang.cookie 
	node2: cat /var/lib/rabbitmq/.erlang.cookie 
	node3: cat /var/lib/rabbitmq/.erlang.cookie 

# 4.后台启动rabbitmq所有节点执行如下命令,启动成功访问管理界面:
	rabbitmq-server -detached 

# 5.在node2和node3执行加入集群命令:
	1.关闭        rabbitmqctl stop_app
	2.加入集群    rabbitmqctl join_cluster rabbit@mq1
	3.启动服务    rabbitmqctl start_app

# 6.查看集群状态,任意节点执行:
	rabbitmqctl cluster_status

# 7.如果出现如下显示,集群搭建成功:
	Cluster status of node rabbit@mq3 ...
	[{nodes,[{disc,[rabbit@mq1,rabbit@mq2,rabbit@mq3]}]},
	{running_nodes,[rabbit@mq1,rabbit@mq2,rabbit@mq3]},
	{cluster_name,<<"rabbit@mq1">>},
	{partitions,[]},
	{alarms,[{rabbit@mq1,[]},{rabbit@mq2,[]},{rabbit@mq3,[]}]}]

伪集群搭建

# 1.清除历史记录
	cd /var/lib/rabbitmq/mnesia
	rm -rf *

# 2.启动三个实例(记得开启云服务器的对应端口)
	cd /usr/lib/rabbitmq/bin
    RABBITMQ_NODE_PORT=5672 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15672}]" 
    RABBITMQ_NODENAME=rabbit rabbitmq-server -detached
    RABBITMQ_NODE_PORT=5673 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15673}]" 
    RABBITMQ_NODENAME=rabbit2 rabbitmq-server -detached
    RABBITMQ_NODE_PORT=5674 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15674}]" 
    RABBITMQ_NODENAME=rabbit3 rabbitmq-server -detached
 
# 3.配置主备集群
    rabbitmqctl -n rabbit2 stop_app
    rabbitmqctl -n rabbit2 reset
    rabbitmqctl -n rabbit2 join_cluster rabbit@`hostname -s`
    rabbitmqctl -n rabbit2 start_app

    rabbitmqctl -n rabbit3 stop_app
    rabbitmqctl -n rabbit3 reset
    rabbitmqctl -n rabbit3 join_cluster rabbit@`hostname -s`
    rabbitmqctl -n rabbit3 start_app

# 4.查看集群信息
	rabbitmqctl -n rabbit cluster_status

镜像集群

默认情况下，RabbitMQ集群中队列的内容位于单个节点（声明队列的节点）上。这与交换和绑定形成对比，后者总是被认为位于所有节点上。队列可以选择性地跨多个节点镜像。

QQ截图20220223224100

搭建集群

# 0.策略说明
	rabbitmqctl set_policy [-p <vhost>] [--priority <priority>] [--apply-to <apply-to>] <name> <pattern>  <definition>
    -p Vhost： 可选参数，针对指定vhost下的queue进行设置
    Name:     policy的名称
    Pattern: queue的匹配模式(正则表达式)
    Definition：镜像定义，包括三个部分ha-mode, ha-params, ha-sync-mode
    	ha-mode:指明镜像队列的模式，有效值为 all/exactly/nodes
            all：表示在集群中所有的节点上进行镜像
            exactly：表示在指定个数的节点上进行镜像，节点的个数由ha-params指定
            nodes：表示在指定的节点上进行镜像，节点名称通过ha-params指定
        ha-params：ha-mode模式需要用到的参数
        ha-sync-mode：进行队列中消息的同步方式，有效值为automatic和manual
        priority：可选参数，policy的优先级

# 1.查看当前策略
 	rabbitmqctl list_policies

# 2.添加策略
	rabbitmqctl set_policy ha-all '^hello' '{"ha-mode":"all","ha-sync-mode":"automatic"}' 
	说明:策略正则表达式为 “^” 表示所有匹配所有队列名称  ^hello:匹配hello开头队列

# 3.删除策略
	rabbitmqctl clear_policy ha-all

石一歌的RabbitMQ笔记

RabbitMQ实战教程

MQ引言

什么是MQ

MQ有哪些

不同MQ特点

RabbitMQ

RabbitMQ引言

AMQP协议

安装

配置

常见命令

WEB界面概览

Hello RabbitMQ

消息模型介绍

依赖

直连模型

工作队列模型

发布订阅模型

路由模型

话题模型

远程过程调用模型

发布者确认

Springboot整合

初始环境

直连模型

工作队列模型

发布订阅模型

路由模型

话题模型

MQ的应用场景

异步处理

应用解耦

流量削峰

RabbitMQ集群

主备集群

镜像集群

参考链接

推荐阅读