108.【RabbitsMQ】（二）

6.消息的分发策略

(1).消息的分发策略

MQ消息 队列有如下几个角色

1. 生产者
2. 存储消息
3. 消费者

那么生产者生成消息以后，MQ进行存储，消费者是如何获取消息的呢？一般获取数据的方式无外乎推（push）或者拉（pull）两种方式，典型的 git就有推拉机制，我们发送的 http (request respon)请求就是一种典型的拉取数据库数据返回的过程。而消息队列 MQ是一种推送的过程，而这些推机制会使用到很多的业务场景也有很多对应推机制策略.

(2).场景分析一

比如我在 APP上下了一个订单，我们的系统和服务很多，我们如何得知这个消息被哪个系统或者哪些服务器或者系统进行消费，那这个时候就需要一个分发的策略。这就需要消费策略。或者称之为消费的方法论

(3).场景分析二

在发送消息的过程中可能会出现异常，或者网络的抖动，故障等等因为造成消息的无法消费，比如用户在下订单，消费 MQ接受，订单系统出现故障，导致用户支付失败，那么这个时候就需要消息中间件就必须支持消息重试机制策略。也就是支持：出现问题和故障的情况下，消息不丢失还可以进行重发

(4).消息分发策略的机制和对比

发布订阅: 定于公众号

轮询分发: 不看网速的问题，完全公平的-》抢票机制

公平分发: 会造成数据的倾斜(会收到网速的影响，能者多劳模式)

重发:假如创建了订单，但没有完成。那么就会留存数据(抢票未付钱,预留20分钟)

7.消息队列高可用和高可靠

(1).什么是高可用机制

所谓高可用：是指产品在规定的条件和规定的时刻或时间内处于可执行规定功能状态的能力

当业务量增加时，请求也过大，一台消息中间件服务器的会触及硬件（CPU，内存，磁盘）的极限，一台消息服务器你已经无法满足业务的需求，所以消息中间件必须支持集群部署，来达到高可用的目的。

(2).集群模式1 - Master-slave主从共享数据的部署方式

(3).集群模式2 - Master-slave主从同步部署方式

解释：这种模式写入消息同样在 Master主节点上，但是主节点会同步数据到 slave节点形成副本，和 zookeeper或者 redis主从机制很雷同。这样可以达到负载均衡的效果，如果消费者有多个这样就可以去不同的节点进行消费，以为消息的拷贝和同步会占用很大的带宽和网络资源。在后去的 rabbitmq中会有使用

(4).集群模式3 - 多主集群同步部署模式

(5).集群模式4 - 多主集群转发部署模式

解释：如果你插入的数据是 broker-1中国，元数据信息会存储数据的相关描述和记录存放的位置（队列）。它会对描述信息也就是元数据信息进行同步，如果消费者在 broker-2中进行消费，发现自己节点没有对应的信息，可以从对应的元数据信息中去查询，然后返回对应的消息信息，场景：比如买火车票或者黄牛买演唱会门票，比如第一个黄牛有顾客说要买的演唱会门票，但是没有但是他回去联系其他的黄牛询问，如果有就返回

(6).集群模式5 Master-slave与 Broker-cluster组合的方案

解释：实现多主多从的热备机制来完成消息的高可用以及数据的热备机制，在生产规模达到一定的阶段的时候，这种使用的频率比较高。

终归三句话:

1. 要么消息共享
2. 要么消息同步
3. 要么元数据共享

(8).什么是高可靠机制

所谓高可靠是指：系统可以无故障持续运行，比如一个系统突然崩溃，报错，异常等等并不影响线上业务的正常运行，出错的几率极低，就称之为：高可靠

在高并发的业务场景中，如果不能保证系统的高可靠，那造成的隐患和损失是非常严重的

如何保证中间件消息的可靠性呢，可以从两个方面考虑：

1. 消息的传输：通过协议来保证系统间数据解析的正确性
2. 消息的存储区可靠：通过持久化来保证消息的可靠性

(二)、RabbitMQ入门及安装

1. RabbitMQ入门及安装

(1). 概述

简单概述：

RabbitMQ是一个开源的遵循 AMQP协议实现的基于 Erlang语言编写，支持多种客户端（语言），用于在分布式系统中存储消息，转发消息，具有高可用，高可扩性，易用性等特征

(2).下载RabbitMQ

1. 下载地址：https://www.rabbitmq.com/download.html
2. 环境准备：CentOS7.x + /Erlang

RabbitMQ是采用 Erlang语言开发的，所以系统环境必须提供 Erlang环境，第一步就是安装 Erlang

(3) 安装Erlang

安装下载

mkdir -p /home/rabbitmq
cd /home/rabbitmq
# 将安装包上传到linux系统中
erlang-22.0.7-1.el7.x86_64.rpm
rabbitmq-server-3.7.18-1.el6.noarch.rpm
# 解压erlang
rpm -Uvh erlang-22.0.7-1.el7.x86_64.rpm
yum install -y erlang
#查看版本号
erl -v

(4). 安装socat

安装下载

yum install -y socat

(5). 安装rabbitmq

安装下载

rpm -Uvh rabbitmq-server-3.7.18-1.el6.noarch.rpm
yum install rabbitmq-server -y

启动服务

# 启动服务
systemctl start rabbitmq-server
# 查看服务状态，如图
systemctl status rabbitmq-server.service
# 开机自启动
systemctl enable rabbitmq-server
# 停止服务
systemctl stop rabbitmq-server

2.RabbitMQ可视化界面

(1).安装操作

默认情况下，是没有安装web端的客户端插件，需要安装才可以生效

rabbitmq-plugins enable rabbitmq_management

利用我们的域名访问即可

说明：rabbitmq有一个默认账号和密码是：guest默认情况只能在 localhost本计下访问，所以需要添加一个远程登录的用户

安装完毕以后，重启服务即可

systemctl restart rabbitmq-server

一定要记住，在对应服务器（阿里云，腾讯云等）的安全组中开放15672端口

(2). 授权账号和密码

新增用户

rabbitmqctl add_user admin admin

设置用户分配操作权限

rabbitmqctl set_user_tags admin administrator

用户级别：

1. administrator：可以登录控制台、查看所有信息、可以对 rabbitmq进行管理
2. monitoring：监控者 登录控制台，查看所有信息
3. policymaker：策略制定者 登录控制台，指定策略
4. managment 普通管理员 登录控制台

为用户添加资源权限

rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"

网页登录成功

修改密码

rabbitmqctl change_password admin xxxx

3. RabbitMQ之Docker安装

(1). Dokcer安装RabbitMQ

(1). 查询rabbitmq
docker search rabbitmq
 (2).拉取
docker pull rabbitmq:management
 (3). 安装容器
 docker run -di --name myrabbit -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15672:15672 -p 55672:55672 -p 5672:5672 -p 21613:21613 -p 1883:1883 rabbitmq:management
(4). 启动容器
docker start 容器id 

4.RabbitMQ的角色分类

层层递进----

(1).RabbitMQ的角色分类

1. none:

• 不能访问management plugin (也就是不能访问rabbitMq 的web应用)

2. mangement: 查看自己相关节点信息

• 列出自己可以通过AMQP登入的虚拟机
• 查看自己的虚拟机节点 virtual hosts的queues,exchaanges和bindings信息
• 查看和关闭自己的channels和connections
• 查看有关自己的虚拟机节点virtual hotos的统计信息，包括用户在这个节点 virtual hostos中的信息活动。

3. Policymaker

• 包含mangement所有权限
• 查看和创建和删除自己的virtual hosts所属的polices和parmeters信息。

4. Monitoring

包含management所有权限

• 罗列出所有的virtual hosts 包括不能登入的virtual hostos
• 查看其他用户的 connections和channels信息
• 查看节点级别的数据 如: clustering和memory使用情况
• 查看所有的 virtual hostos的全局统计信息

5. Adminitrator

• 最高权限
• 可以创建和删除 virtuak hostos
• 可以查看，创建和删除 uers
• 查看创建 permissions
• 关闭所有用户的connections

(三)、RabbitMQ入门案例

1. RabbitMQ入门案例 - Simple 简单模式

实现步骤

1. jdk1.8
2. 构建一个 maven工程
3. 导入 rabbitmq的 maven依赖
4. 启动 rabbitmq-server服务
5. 定义生产者
6. 定义消费者
7. 观察消息的在 rabbitmq-server服务中的进程

(1).构建一个Maven项目

(2).导入依赖

java原生依赖

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

(3).第一种Simple模型

在上图的模型中，有以下概念：

1. 生产者，也就是要发送消息的程序
2. 消费者：消息的接受者，会一直等待消息到来。
3. 消息队列：图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

1. 生产者

1. 假如说我们没有指定交换机，那么系统就会选择默认的交换机。并且在发布信息的时候，我们的第二个参数一定要编写队列名而不是，路由Key。
2. 如果我们有指定交换机的话，那么就必须只当路由Key而不是：我们的队列名。
3. 接受信息的参数: 第一个是队列名、第二个是: 是否自动确认、第三个是接受监听的信息。

Products.java

package com.jsxs.rabbitmq.simple;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 所有的中间件技术都是基于tcp/ip协议基础之上的协议规范，只不过rabbitmq遵循的是 ampq 协议也就是tcp/ip是基础.
        //port ip
        // 1.创建链接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("8.130.48.9");  // ip
        connectionFactory.setPort(5672);  // port
        connectionFactory.setUsername("admin"); //username
        connectionFactory.setPassword("xxxxx"); // password
        connectionFactory.setVirtualHost("/"); // 消息放在哪
        Connection connection = null;
        Channel channel = null;
        try {
            // 2: 创建连接Connection Rabbitmq为什么是基于channel去处理而不是链接? 长连接----信道channel
            connection = connectionFactory.newConnection("生成者");
            // 3: 通过连接获取通道Channel
            channel = connection.createChannel();
            // 4: 通过通创建交换机，声明队列，绑定关系，路由key,发送消息，和接收消息
            String queueName = "queue1";
            channel.queueDeclare(queueName, false, false, false, null);
            // 5: 准备消息内容
            String message = "Hello jsxs!!!";
            // 6: 发送消息给队列queue
            // @params1: 交换机  @params2 队列、路由key @params 消息的状态控制  @params4 消息主题
            // 面试题：可以存在没有交换机的队列吗？不可能，虽然没有指定交换机但是一定会存在一个默认的交换机。
            channel.basicPublish("", queueName, null, message.getBytes());
            System.out.println("消息发送成功!!!");
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            // 7: 关闭通道
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
            // 8: 关闭连接
            if (connection != null && connection.isOpen()) {
                try {
                    connection.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }
    }
}