当前位置： 首页 > news >正文

网站建设实训个人总结3000,用jsp做的网站代码,杭州网站建设公司官网,视频网站应该怎么做一、RocketMQ入门一、RocketMQ 介绍1.1 RocketMQ 是什么#xff1f;1.2 RocketMQ 应用场景01、应用解耦02、流量削峰03、数据分发1.3 RocketMQ 核心组成01、NameServer02、Broker03、Producer04、Consumer1.6 运转流程1.5 RocketMQ 架构01、NameServer 集群02、Broker 集群03、… 一、RocketMQ入门一、RocketMQ 介绍1.1 RocketMQ 是什么1.2 RocketMQ 应用场景01、应用解耦02、流量削峰03、数据分发1.3 RocketMQ 核心组成01、NameServer02、Broker03、Producer04、Consumer1.6 运转流程1.5 RocketMQ 架构01、NameServer 集群02、Broker 集群03、Producer 集群04、Consumer 集群07、集群工作流程06、集群间的交互方式1.6 RocketMQ 优缺点01、优点02、缺点1.7 各种 MQ 比较二、RocketMQ 安装Linux 版本2.1 环境要求2.2 安装步骤01、上传安装包02、解压安装包03、参数配置2.3 目录介绍2.4 启动 RocketMQ2.5 测试 RocketMQ2.6 关闭 RocketMQ三、rocketmq-console 集群监控平台搭建3.1 简介3.2 搭建集群监控平台01、下载02、上传解压03、修改配置参数04、打包05、启动和访问06、问题点四、RocketMQ 发送消息基本样例4.1 普通消息发送4.2 普通消息消费一、RocketMQ 介绍 1.1 RocketMQ 是什么 RocketMQ 是一款纯 java、分布式、队列模型的开源消息中间件支持事务消息、顺序消息、批量消息、定时消息、消息回溯等。 1.2 RocketMQ 应用场景 消息队列是一种先进先出的数据结构其应用场景主要包含以下三个方面 01、应用解耦 系统的耦合性越高容错性就越低。 以电商应用为例用户创建订单后如果耦合调用库存系统、物流系统、支付系统任何一个子系统出了故障或者因为升级等原因暂时不可用都会造成下单操作异常影响用户使用体验。 使用消息队列解耦系统的耦合性就会提高了。比如如果物流系统发生故障需要几分钟才能修复好在这段时间内物流系统要处理的数据被缓存到消息队列中用户的下单操作正常完成。当物流系统恢复后补充处理存在消息队列中的订单消息即可终端系统感知不到物流系统发生过几分钟故障。
02、流量削峰 应用系统如果遇到系统请求流量的瞬间猛增有可能会将系统压垮。有了消息队列后可以将大量请求缓存起来分散到很长一段时间处理这样可以大大提升系统的稳定性和用户体验。 一般情况下为了保证系统的稳定性如果系统负载超过阈值就会阻止用户请求这会影响用户体验。而如果使用消息队列将请求缓存起来等待系统处理完毕后通知用户下单完毕这样的体验应该要好很多。 处于经济考量的目的 业务系统正常时段的 QPS 如果是 1000流量最高峰是 10000为了应对流量高峰配置高性能的服务器显然不划算这时就可以考虑使用消息队列对峰值流量削峰。 03、数据分发 通过消息队列可以让数据在多个系统之间进行流通。数据的产生方不需要关心谁来使用数据只需要将数据发送到消息队列数据使用方直接在消息队列中直接获取数据即可。
1.3 RocketMQ 核心组成 RocketMQ 主要有四大核心组成部分NameServer、Broker、Producer、Consumer。 Topic区分消息的种类。一个发送者可以发送消息给一个或多个 Topic一个消息的接收者可以订阅一个或多个 Topic 消息。 Message Queue相当于是 Topic 的分区用于并行发送和接收消息。 01、NameServer NameServer 是一个几乎无状态节点可集群部署节点之间没有任何信息同步。所以 RocketMQ 需要先启动 NameServer 再启动 Broker。 作用 NameServer 是整个 RocketMQ 的 大脑它相当于是服务注册中心的角色用来管理 Broker。举例各个邮局的管理机构。 每个 NameServer 中都保存着 Broker 集群的整个路由信息和用于客户端查询的队列信息生产者和消费者通过 NameServer 去获取整个 Broker 集群的路由信息从而进行消息的投递和消费。 每个 Broker 在启动的时候会到 NameServer 中注册Producer 在发送消息前会根据 Topic 到 NameServer 获取到 Broker 的路由信息进而和 Broker 取得连接。Consumer 也会定时获取 Topic 的路由信息。所以从功能上看应该是和 ZooKeeper 差不多的但是据说 RocketMQ 的早期版本确实是使用的ZooKeeper 后来改为了自己实现 NameServer。 与 ZooKeeper 的区别 NameServer 和 ZooKeeper 的作用大致是相同的。从宏观上来看NameServer 做的东西很少就是保存一些运行数据NameServer 之间不互相连这就需要 Broker 端连接所有的 NameServer运行数据的改动要发送到每一个 NameServer 从而来保证运行数据的一致性这个一致性确实有点弱这样就变成了 NameServer很轻量级但是 Broker 端就要做更多的东西了。 但是在 ZooKeeper 中Broker 只需要连接其中的一台机器运行数据分发、一致性都交给了 ZooKeeper 来完成。 高可用保障 Broker 在启动时向所有 NameServer 注册主要是服务器地址等 生产者在发送消息之前先从NameServer 获取 Broker 服务器地址列表消费者一样然后根据负载均衡算法从列表中选择一台服务器进行消息发送。 NameServer 与每台 Broker 服务保持长连接并间隔 30S 检查 Broker 是否存活如果检测到 Broker 宕机则从路由注册表中将其移除这样就可以实现 RocketMQ 的高可用。
02、Broker Broker 是消息存储中心主要作用是接收来自 Producer 的消息并存储Consumer 从这里取得消息。举例邮局。 它还存储与消息相关的元数据包括用户组、消费进度偏移量、队列信息等。从部署结构图中可以看出 Broker 有 Master 和 Slave 两种类型Master 既可以写又可以读Slave 只可以读不可以写。 从物理结构上看 Broker 的集群部署方式有四种单 Master 、多 Master 、多 Master 多 Slave同步刷盘、多 Master 多 Slave异步刷盘。 单 Master 这种方式风险较大一旦 Broker 重启或者宕机时会导致整个服务不可用。不建议线上环境使用可以用于本地测试。 多 Master 所有消息都是 Master没有 Slave。这种方式优缺点 优点配置简单单个 Master 宕机或重启维护对应用无影响。在磁盘配置为 RAID10 时即使机器宕机不可恢复的情况下由于RAID10磁盘非常可靠消息也不会丢异步刷盘丢失少量消息同步刷盘一条不丢性能最高缺点单台机器宕机期间这台机器上未被消费的消息在机器恢复之前不可订阅消息实时性会受到影响。 多 Master 多 Slave异步复制 异步先响应后再存入磁盘。 每个 Master 配置一个 Slave有多对 Master-SlaveHA 采用异步复制方式主备之间有毫秒级消息延迟。这种方式优缺点 优点即使磁盘损坏消息丢失的非常少且消息实时性不会受到影响。同时 Master 宕机后消费者仍然可以从 Salve 消费而且此过程对应用透明不需要人工干预性能同多Master模式几乎一样缺点Master宕机磁盘损坏情况下会丢失少量消息。 多 Master 多 Slave同步双写 同步立刻存入磁盘后响应。 每个 Master 配置一个 Slave有多对 Master-SlaveHA 采用同步双写方式即只有主备都写成功才向应用返回成功。这种模式的优缺点 优点数据与服务都无单点故障Master 宕机情况下消息无延迟服务可用性与数据可用性都非常高缺点性能比异步复制模式略低大约低10%左右发送单个消息的 RT 会略高且目前版本在主节点宕机后备机不能自动切换为主机。 高可用保障 每个 Broker 与 NameServer 集群中的所有节点建立长连接定时每隔30s注册 Topic 信息到所有 NameServer。NameServer定时每隔10s扫描所有存活 Broker 的连接如果 NameServer 超过2分钟没有收到心跳则 NameServer 断开与 Broker 的连接。
03、Producer Producer 也称为消息发布者负责生产并发送消息至 Topic。举例发信者。生产者向 brokers 发送由业务应用程序系统生成的消息。RocketMQ 提供了发送同步、异步和单向one-way的多种范例。 同步发送 同步发送消息指消息发送方发出数据后会在收到接收方发回响应之后才发下一个数据包。一般用于重要通知消息比如重要通知邮件、营销短信等。 异步发送 异步发送指发送方发出数据后不等接收方发回响应接着发送下个数据包一般用于可能链路耗时较长而对响应时间敏感的业务场景例如用户视频上传后通知启动转码服务。假如过一段时间检测到某个信息发送失败可以选择重新发送。 单向发送 单向发送是指只负责发送消息而不等待服务器回应且没有回调函数触发适用于某些耗时非常短但对可靠性要求并不高的场景例如日志收集。 生产者组 生产者组Producer Group是一类 Producer 的集合这类 Producer 通常发送一类消息并且发送逻辑一致所以将这些 Producer 分组在一起。从部署结构上看生产者通过 Producer Group 的名字来标记自己是一个集群。 高可用保障 Producer 与 NameServer 集群中的其中一个节点随机选择建立长连接定期从 NameServer 取 Topic 路由信息并向提供 Topic 服务的 Master 建立长连接且定时向 Master 发送心跳。Producer 完全无状态可集群部署。 Producer 每隔30s由 ClientConfig 的 pollNameServerInterval 从 Nameserver 获取所有 topic 队列的最新情况这意味着如果 Broker 不可用Producer 最多30s能够感知在此期间内发往 Broker 的所有消息都会失败。 Producer 每隔30s由 ClientConfig 中 heartbeatBrokerInterval 决定向所有关联的 broker 发送心跳Broker 每隔10s扫描所有存活的连接如果 Broker 在2分钟内没有收到心跳数据则关闭与 Producer 的连接。
04、Consumer Consumer 也称为消息订阅者负责从 Topic 接收并消费消息。举例收信者。消费者从 brokers 里拉取信息并将其输入应用程序中。 消费者组 消费者组Consumer Group是一类 Consumer 的集合名称这类 Consumer 通常消费同一类消息并且消费逻辑一致所以将这些 Consumer 分组在一起。消费者组与生产者组类似都是将相同角色的分组在一起并命名。 RocketMQ 中的消息有个特点同一条消息只能被某一消费组其中的一台机器消费但是可以同时被不同的消费组消费。 高可用保障 Consumer 与 NameServer 集群中的其中一个节点随机选择建立长连接定期从 NameServer 取 Topic 路由信息并向提供 Topic 服务的 Master、Slave 建立长连接且定时向 Master、Slave 发送心跳。Consumer 既可以从 Master 订阅消息也可以从 Slave 订阅消息订阅规则由 Broker 配置决定。 Consumer 每隔30s从 Nameserver 获取 topic 的最新队列情况这意味着 Broker 不可用时Consumer 最多最需要30s才能感知。 Consumer 每隔30s由 ClientConfig 中 heartbeatBrokerInterval 决定向所有关联的 broker 发送心跳Broker 每隔10s扫描所有存活的连接若某个连接2分钟内没有发送心跳数据则关闭连接并向该 Consumer Group 的所有 Consumer 发出通知Group 内的 Consumer 重新分配队列然后继续消费。 当 Consumer 得到 master 宕机通知后转向 slave 消费slave 不能保证 master 的消息100%都同步过来了因此会有少量的消息丢失。但是一旦 master 恢复未同步过去的消息会被最终消费掉。
1.6 运转流程 NameServer 先启动Broker 启动时向 NameServer 注册生产者在发送某个主题的消息之前先从 NamerServer 获取 Broker 服务器地址列表有可能是集群然后根据负载均衡算法从列表中选择一台 Broker 进行消息发送NameServer 与每台 Broker 服务器保持长连接并间隔 30S 检测 Broker 是否存活如果检测到 Broker 宕机使用心跳机制 如果检测超120S则从路由注册表中将其移除消费者在订阅某个主题的消息之前从 NamerServer 获取 Broker 服务器地址列表有可能是集群但是消费者选择从 Broker 中 订阅消息订阅规则由 Broker 配置决定。 1.5 RocketMQ 架构 RocketMQ 架构图中展示了四个集群 01、NameServer 集群 提供轻量级的服务发现及路由每个 NameServer 记录完整的路由信息提供相应的读写服务支持快速存储扩展。 NameServer 是一个功能齐全的服务器主要包含两个功能 Broker 管理接收来自 Broker 集群的注册请求提供心跳机制检测 Broker 是否存活路由管理每个 NameServer 持有全部有关 Broker 集群和客户端请求队列的路由信息。 02、Broker 集群 通过提供轻量级的 Topic 和 Queue 机制处理消息存储。同时支持推Push和拉Pull两种模型包含容错机制。提供强大的峰值填充和以原始时间顺序累积数千亿条消息的能力。此外还提供灾难恢复丰富的指标统计数据和警报机制这些都是传统的消息系统缺乏的。 Broker 有几个重要的子模块 远程处理模块Broker 入口处理来自客户端的请求客户端管理包括消息生产者和消费者维护消费者的主题订阅存储服务提供在物理硬盘上存储和查询消息的简单 APIHA 服务提供主从 Broker 间数据同步索引服务通过指定键为消息建立索引并提供快速消息查询。 03、Producer 集群 消息生产者支持分布式部署分布式生产者通过多种负载均衡模式向 Broker 集群发送消息。 04、Consumer 集群 消息消费者也支持 Push 和 Pull 模型的分布式部署还支持集群消费和消息广播。提供了实时的消息订阅机制可以满足大多数消费者的需求。 DefultMQPullConsumer consumer 定时向 broker 发送请求获取内存数据避免给 broker 造成巨大的压力。一般会在本地使用定时任务实现。 DefultMQPushConsumer consumer 向 broker 发送请求两者保持长链接的状态。broker 会定时每 5 秒去查询 consumer 中是否有要订阅的数据有就将消息推送给 consumer。 无论是 pull 还是 push其实本质上都是拉取消息。 07、集群工作流程 启动 NameServerNameServer 启动后监听端口等待 Broker、Producer、Consumer 连接相当于一个路由控制中心。Broker 启动后跟所有的 NameServer 保持长连接定时发送心跳包。心跳包中包含当前 Broker 信息IP 端口等以及存储所有 Topic 信息。注册成功后NameServer 集群中就有 Topic 与 Broker 的映射关系。收发消息前先创建 Topic创建 Topic 时需要指定该 Topic 要存储在哪些 Broker 上也可以在发送消息时自动创建 Topic。Producer 发送消息启动时先跟 NameServer 集群中的其中一台建立长连接并从 NameServer 中获取当前发送的 Topic 存在哪些 Broker 上轮询从队列列表中选择一个队列然后与队列所在的 Broker 建立长连接从而向 Broker 发消息。Consumer 跟 Producer 类似跟其中一台 NameServer 建立长连接获取当前订阅 Topic 存在哪些 Broker 上然后直接跟 Broker 建立连接通道开始消费消息。 06、集群间的交互方式 Broker Master 和 Broker Slave 是主从结构会执行数据同步 Data Sync 每个 Broker 与 NameServer 集群中所有节点建立长连接定时注册 Topic 信息到所有 NameServerProducer 与 NameServer 集群中的其中一个节点随机建立长连接定期从 NameServer 获取 Topic 路由信息并与提供 Topic 服务的 Broker Master 建立长连接定时向 Broker 发送心跳Producer 只能将消息发送到 Broker Master但是 Consumer 同时和 Broker Master 和 Broker Slave 建立长连接既可以从 Master 订阅消息也可以从 Slave 订阅消息。 1.6 RocketMQ 优缺点 01、优点 单机吞吐量十万级可用性非常高分布式架构消息可靠性经过参数优化配置消息可以做到 0 丢失功能支持MQ 功能较为完善还是分布式的扩展性好支持 10 亿级别的消息堆积不会因为堆积导致性能下降源码是 Java方便结合公司自己的业务进行二次开发天生为金融互联网领域而生对于可靠性要求很高的场景尤其是电商里面的订单扣款以及业务削峰在大量交易涌入时后端可能无法及时处理的情况RoketMQ 在稳定性上可能更值得信赖这些业务场景在阿里双11已经经历了多次考验。 02、缺点 支持的客户端语言不多目前仅支持 Java 及 C而且 C 还不成熟没有在 MQ 核心中去实现 JMS 等接口有些系统要迁移需要修改大量代码 1.7 各种 MQ 比较 二、RocketMQ 安装Linux 版本 安装 RocketMQ 版本4.5.0我这里使用的是阿里云服务器也可以在虚拟机上操作。 2.1 环境要求 Linux 64 位操作系统JDK 1.8Maven 3.9.0maven 版本不固定但是最好不要使用最高版本 2.2 安装步骤 01、上传安装包 将下载好的 RocketMQ 安装包上传到服务器上
02、解压安装包

解压

unzip rocketmq-all-4.5.0-bin-release.zip# 将解压包移动到指定路径下 mv rocketmq-all-4.5.0-bin-release ../software03、参数配置 这里需要配置三个文件 /conf/broker.conf 指定 broker 的命名空间地址和当前 broker 监听的 IP: 默认情况下namesrvAddr 127.0.0.1:9876brokerIP1 127.0.0.1。 /bin/runserver.sh RocketMQ 默认的虚拟机内存比较大启动 Broker 如果因为内存不足失败就需要编辑这两个配置文件修改 JVM 内存大小 JAVA_OPT\({JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize128m -XX:MaxMetaspaceSize320m/bin/runbroker.sh 同上修改 JVM 内存大小 JAVA_OPT\){JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m2.3 目录介绍 bin启动脚本包括 shell 脚本和 cmd 脚本conf实例配置文件包括 broker 配置文件、logback 配置文件等lib依赖 jar 包包括 Netty、commons-lang、FastJSON 等 2.4 启动 RocketMQ 启动前先查看进程 启动命令在 bin 目录下执行

启动 nameserver

nohup sh mqnamesrv -n 公网IP:9876 # 启动 broker nohup sh mqbroker -n 公网IP:9876 -c conf/broker.conf autoCreateTopicEnabletrue 启动后查看进程有这两个进程即启动成功
2.5 测试 RocketMQ 模拟生产者发送消息 # 生产者 sh tools.sh org.apache.rocketmq.example.quickstart.Producer输入这个命令后控制台会输出很多的信息不报错就说明发送成功了 模拟消费者接收消息 # 消费者 sh tools.sh org.apache.rocketmq.example.quickstart.Consumer2.6 关闭 RocketMQ

关闭服务

sh mqshutdown namesrv sh mqshutdown broker三、rocketmq-console 集群监控平台搭建 3.1 简介 RocketMQ 有个可视化的管理界面通过可视化界面我们可以方便地监控 RocketMQ 集群并实现很多操作。比如创建管理 Topic查看和发送 essage等等。 3.2 搭建集群监控平台 01、下载 RocketMQ提供了UI管理工具名为rocketmq-console项目地址https://github.com/apache/rocketmq-externals/tree/master/rocketmq-console。 这个是 rocketmq 的扩展里面不仅包含控制台的扩展也包含对大数据 flume、hbase 等组件的对接和扩展。 02、上传解压 上传 解压# 解压 unzip rocketmq-console.zip# 移动到 software 目录下 mv rocketmq-console ../software03、修改配置参数 修改 rocketmq-console\src\main\resources\application.properties 配置文件
04、打包 进入 rocketmq-console 目录下执行命令

打包

mvn clean package ‐Dmaven.test.skiptrue打包完成后会在 /software/rocketMQ/rocketmq-console 目录下生成一个 target 文件夹 05、启动和访问 启动 进入 /rocketmq-console/target 目录下执行 # 指定端口号和命名空间地址 java -jar rocketmq-console-ng-1.0.1.jar –server.port8088 –rocketmq.config.namesrvAddr公网IP:9876访问 http://106.15.0.30:8088
06、问题点 防火墙 防火墙需要开放访问 RocketMQ 的一系列端口号 # 查看防火墙状态 systemctl status firewalld# 关闭防火墙 systemctl stop firewalld# 启动防火墙 systemctl start firewalld# 永久开放指定端口号【把用到的端口号都开放】 firewall-cmd –zonepublic –add-port10909/tcp –permanent firewall-cmd –zonepublic –add-port10911/tcp –permanent firewall-cmd –zonepublic –add-port9876/tcp –permanent firewall-cmd –zonepublic –add-port9877/tcp –permanent# 重新加载防火墙 firewall-cmd –reload# 或者重启防火墙 systemctl restart firewalld.service# 查看防火墙信息列表 firewall-cmd –list-all# 只查看防火墙开放端口号列表 firewall-cmd –list-ports除了这一层防火墙之外阿里云服务器自己还有一层防火墙 iptables是默认配置的我们也需要关闭这层防火墙或者开放对应的端口号 # 查看防火墙状态出现的问题 service iptables status# 关闭防火墙 service iptables stop安全组 防火墙端口号开放之后同时也需要在 ECS 服务器安全组中添加端口规则
四、RocketMQ 发送消息基本样例 引入 jar 包 dependencygroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-client/artifactIdversion4.5.0/version /dependency4.1 普通消息发送 /*** 同步发送消息** author qiaohaojie* date 2023/2/20 17:00/ public class SyncProducer {public static void main(String[] args) throws Exception {// 1. 实例化消息生产者 producerDefaultMQProducer producer new DefaultMQProducer(group_producer_demo1);// 2. 设置 nameServer 的地址producer.setNamesrvAddr(公网IP:9876);// 关闭 VIP 通道 // producer.setVipChannelEnabled(false);producer.setSendMsgTimeout(3000);// 3. 启动 Producer 实例producer.start();for (int i 0; i 10; i) {// 4. 创建消息 messageMessage message new Message(Topic_Demo, Tags, Hello RocketMQ i, hello.getBytes(RemotingHelper.DEFAULT_CHARSET));// 5. 发送消息SendResult result producer.send(message);System.out.println(result);}// 6. 关闭 producerproducer.shutdown();} }这里的 message 实例中有几个参数
topic代表消息的主题tags主要用于消息过滤keys消息的唯一值body消息体代表消息的内容
4.2 普通消息消费 /** 同步发送消息-消费者** author qiaohaojie* date 2023/2/20 21:39*/ public class Consumer {public static void main(String[] args) throws Exception {// 1. 创建DefaultMQPushConsumer实例DefaultMQPushConsumer consumer new DefaultMQPushConsumer(group_consumer_demo1);// 2. 设置nameServer地址consumer.setNamesrvAddr(公网IP:9876);// 设置消息拉取最大数consumer.setConsumeMessageBatchMaxSize(2);// 3. 设置subscribe这里是要读取的主题信息consumer.subscribe(Topic_Demo, *);// 4. 设置消息监听consumer.setMessageListener(new MessageListenerConcurrently() {Overridepublic ConsumeConcurrentlyStatus consumeMessage(ListMessageExt msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {// 5. 获取消息信息// 迭代消息信息for (MessageExt msg : msgs) {try {// 获取主题String topic msg.getTopic();// 获取标签String tags msg.getTags();// 获取信息String result new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);System.out.println(Consumer消费消息–topic: topic ,targs: tags ,result: result);} catch (UnsupportedEncodingException e) {e.printStackTrace();// 消息重试return ConsumeConcurrentlyStatus.RECONSUME_LATER;}}// 6. 返回消息读取状态// 消息消费完成return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 开启Consumerconsumer.start();} }消费消息时有两个参数
topic指定要消费的主题subExpression消息过滤规则