外贸网站批量收录,阿里云怎么上传网站,品牌策略的7种类型,简述网络营销策略目录
一、RocketMQ简介
???1.1、核心概念
???1.2、应用场景
???1.3、架构设计
2、RocketMQ Server安装
3、RocketMQ可视化控制台安装与使用
4、SpringBoot整合RocketMQ实现消息发送和接收?
? ? ? ? 4.1、添加maven依赖
???4.2、yaml配置
???4.3、…目录
一、RocketMQ简介
???1.1、核心概念
???1.2、应用场景
???1.3、架构设计
2、RocketMQ Server安装
3、RocketMQ可视化控制台安装与使用
4、SpringBoot整合RocketMQ实现消息发送和接收?
? ? ? ? 4.1、添加maven依赖
???4.2、yaml配置
???4.3、生产者
???4.4、消费者
???4.5、接口
???4.6、接口测试 一、RocketMQ****简介 RocketMQ是由阿里捐赠给Apache的一款低延迟、高并发、高可用、高可靠的分布式消息中间件。经历 了淘宝双十一的洗礼。RocketMQ既可为分布式应用系统提供异步解耦和削峰填谷的能力同时也具备互联网应用所需的海量消息堆积、高吞吐、可靠重试等特性。 官方文档https://rocketmq.apache.org/docs/quick-start/ github中文主页https://github.com/apache/rocketmq/tree/master/docs/cn 1.1、核心概念
Topic消息主题一级消息类型生产者向其发送消息。Message生产者向Topic发送并最终传送给消费者的数据消息的载体。消息属性生产者可以为消息定义的属性包含Message Key和Tag。Message Key消息的业务标识由消息生产者Producer设置唯一标识某个业务逻辑。Message ID消息的全局唯一标识由消息队列RocketMQ系统自动生成唯一标识某条消息。Tag消息标签二级消息类型用来进一步区分某个Topic下的消息分类Producer也称为消息发布者负责生产并发送消息至Topic。Consumer也称为消息订阅者负责从Topic接收并消费消息。分区即Topic Partition物理上的概念。每个Topic包含一个或多个分区。消费位点每个Topic会有多个分区每个分区会统计当前消息的总条数这个称为最大位点MaxOffset分区的起始位置对应的位置叫做起始位点MinOffset。Group一类生产者或消费者这类生产者或消费者通常生产或消费同一类消息且消息发布或订 阅的逻辑一致。Group IDGroup的标识。队列单个Topic下会由一到多个队列来存储消息。Exactly-Once****投递语义Exactly-Once投递语义是指发送到消息系统的消息只能被Consumer处理且仅处理一次即使Producer重试消息发送导致某消息重复投递该消息在Consumer也只被消费 一次。集群消费一个Group ID所标识的所有Consumer平均分摊消费消息。例如某个Topic有9条消息一个Group ID有3个Consumer实例那么在集群消费模式下每个实例平均分摊只消费其中的3条 消息。广播消费一个Group ID所标识的所有Consumer都会各自消费某条消息一次。例如某个Topic有9条消息一个Group ID有3个Consumer实例那么在广播消费模式下每个实例都会各自消费9条消息。定时消息Producer将消息发送到消息队列RocketMQ服务端但并不期望这条消息立马投递而是推迟到在当前时间点之后的某一个时间投递到Consumer进行消费该消息即定时消息。延时消息Producer将消息发送到消息队列RocketMQ服务端但并不期望这条消息立马投递而是延迟一定时间后才投递到Consumer进行消费该消息即延时消息。事务消息RocketMQ提供类似X/Open XA的分布事务功能通过消息队列RocketMQ的事务消息能达到分布式事务的最终一致。顺序消息RocketMQ提供的一种按照顺序进行发布和消费的消息类型分为全局顺序消息和分区顺序消息。全局顺序消息对于指定的一个Topic所有消息按照严格的先入先出FIFO的顺序进行发布和消费。分区顺序消息对于指定的一个Topic所有消息根据Sharding Key进行区块分区。同一个分区内的消息按照严格的FIFO顺序进行发布和消费。Sharding Key是顺序消息中用来区分不同分区的关键字段和普通消息的Message Key是完全不同的概念。消息堆积Producer已经将消息发送到消息队列RocketMQ的服务端但由于Consumer消费能力有限未能在短时间内将所有消息正确消费掉此时在消息队列RocketMQ的服务端保存着未被消费的消息该状态即消息堆积。消息过滤Consumer可以根据消息标签Tag对消息进行过滤确保Consumer最终只接收被过滤后的消息类型。消息过滤在消息队列RocketMQ的服务端完成。消息轨迹在一条消息从Producer发出到Consumer消费处理过程中由各个相关节点的时间、地点等数据汇聚而成的完整链路信息。通过消息轨迹您能清晰定位消息从Producer发出经由消息队列RocketMQ服务端投递给Consumer的完整链路方便定位排查问题。重置消费位点以时间轴为坐标在消息持久化存储的时间范围内默认3天重新设置Consumer对已订阅的Topic的消费进度设置完成后Consumer将接收设定时间点之后由Producer发送到消息队列RocketMQ服务端的消息。死信队列死信队列用于处理无法被正常消费的消息。当一条消息初次消费失败消息队列RocketMQ会自动进行消息重试达到最大重试次数后若消费依然失败则表明Consumer在正常情况下无法正确地消费该消息。此时消息队列RocketMQ不会立刻将消息丢弃而是将这条消息发送到该Consumer对应的特殊队列中。消息队RocketMQ将这种正常情况下无法被消费的消息称为死信消息Dead-Letter Message将存储死信消息的特殊队列称为死信队列Dead-Letter Queue。
1.2、应用场景
削峰填谷诸如秒杀、抢红包、企业开门红等大型活动时皆会带来较高的流量脉冲或因没做相应的保护而导致系统超负荷甚至崩溃或因限制太过导致请求大量失败而影响用户体验消息队列RocketMQ可提供削峰填谷的服务来解决该问题。异步解耦交易系统作为淘宝和天猫主站最核心的系统每笔交易订单数据的产生会引起几百个下游业务系统的关注包括物流、购物车、积分、流计算分析等等整体业务系统庞大而且复杂消息队列RocketMQ可实现异步通信和应用解耦确保主站业务的连续性。顺序收发细数日常中需要保证顺序的应用场景非常多例如证券交易过程时间优先原则交易系统中的订单创建、支付、退款等流程航班中的旅客登机消息处理等等。与先进先出FIFOFirst In First Out原理类似消息队列RocketMQ提供的顺序消息即保证消息FIFO。分布式事务一致性交易系统、支付红包等场景需要确保数据的最终一致性大量引入消息队列RocketMQ的分布式事务既可以实现系统之间的解耦又可以保证最终的数据一致性。大数据分析数据在“流动”中产生价值传统数据分析大多是基于批量计算模型而无法做到实时的数据分析利用阿里云消息队列RocketMQ与流式计算引擎相结合可以很方便的实现业务数据的实时分析。分布式缓存同步天猫双11大促各个分会场琳琅满目的商品需要实时感知价格变化大量并发访问数据库导致会场页面响应时间长集中式缓存因带宽瓶颈限制了商品变更的访问流量通过消息队列RocketMQ构建分布式缓存实时通知商品数据的变化。
1.3、架构设计 RocketMQ架构上主要分为四部分如上图所示:
**Producer**消息发布的角色支持分布式集群方式部署。 Producer 通过 MQ 的负载均衡模块选择相应的Broker集群队列进行消息投递投递的过程支持快速失败并且低延迟。
**Consumer**消息消费的角色支持分布式集群方式部署。支持以 push 推 pull拉两种模式对消息进行消费。同时也支持集群方式和广播方式的消费它提供实时消息订阅机制可以满足大多数用户的需求。
NameServer NameServer 是一个非常简单的 Topic 路由注册中心其角色类似 Dubbo 中的zookeeper支持 Broker 的动态注册与发现。主要包括两个功能 Broker 管理 NameServer 接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制检查Broker是否还存活路由信息管理每个 NameServer 将保存关于 Broker 集群的整个路由信息和用于客户端查询的队列信息。然后Producer 和 Conumser 通过 NameServer 就可以知道整个 Broker 集群的路由信息从而进行消息的投递和消费。NameServer 通常也是集群的方式部署各实例间相互不进行信息通讯。Broker 是向每一台 NameServer 注册自己的路由信息所以每一个NameServer实例上面都保存一份完整的路由信息。当某个 NameServer 因某种原因下线了Broker仍然可以向其它 NameServer 同步其路由信息 Producer,Consumer仍然可以动态感知
Broker的路由的信息。BrokerServer Broker 主要负责消息的存储、投递和查询以及服务高可用保证为了实现这些功能Broker包含了以下几个重要子模块。 **1. Remoting Module**整个Broker 的实体负责处理来自 clients端的请求。 **2. Client Manager**负责管理客户端(Producer/Consumer) 和维护 Consumer 的 Topic订阅信息 **3. Store Service**提供方便简单的API接口处理消息存储到物理硬盘和查询功能。 4. HA Service 高可用服务提供Master Broker 和 Slave Broker之间的数据同步功能。 5. Index Service 根据特定的Message key 对投递到 Broker的消息进行索引服务以提供消息的快速查询。 2、RocketMQ Server****安装 RocketMQ依赖Java环境要求有JDK 1.8以上版本 支持Windows和Linux平台支持源码方式安装和使用已经编译好的安装包安装 我们用windows平台安装RocketMQ Server编译好的安装包来讲解RocketMQ 下载地址https://rocketmq.apache.org/dowloading/releases/ 我们选择4.9.0二进制版本 解压后的目录 benchmark里面是测试Demo
bin可执行脚本conf配置文件
lib依赖的jar包
我们把rocketmq解压包放到D盘soft目录下重命名 rocketmq
第一步系统环境变量加两个配置 ROCKETMQ_HOME“D:soft ocketmq” NAMESRV_ADDR“localhost:9876” 第二步启动 Name Server
进入命令行执行 .inmqnamesrv.cmd 第三步启动 Broker
进入命令行执行 .inmqbroker.cmd -n localhost:9876 autoCreateTopicEnabletrue 第四步发送和接收消息测试
进入命令行消息发送执行 .in ools.cmd org.apache.rocketmq.example.quickstart.Producer 消息发送成功
进入命令行消息接收执行 .in ools.cmd org.apache.rocketmq.example.quickstart.Consumer 消息接收成功
第五步关闭服务
windows 下直接关闭命令行窗口即可
3、RocketMQ****可视化控制台安装与使用
RocketMQ 提供了一些扩展项目支持地址 https://github.com/apache/rocketmq-externals
其中一个 rocketmq - connect - console 项目就是我们需要的可视化控制台 我们把整个项目下载下来打开 rocketmq - console 项目项目是 SpringBoot 开发 打开 application.properties 配置文件我们至少需要修改两个配置项
server.port8080 这个是可视化项目启动端口我们改成 19876
rocketmq.config.namesrvAddr 这个是指定 nameServer 地址和端口我们暂时先搞成localhost:9876等后面搞集群的话要再修改 修改后保存在项目目录下进入命令行执行
D:soft ocketmqClient ocketmq-externals-master ocketmq-console mvn clean package -Dmaven.test.skiptrue 打包执行完后在 target 目录会生成一个可运行 jar rocketmq - console - ng - 2.0.0.jar 我们运行这个 jar 进入命令行执行 java -jar rocketmq-console-ng-2.0.0.jar 启动成功后浏览器输入http://localhost:19876/ 说明一切OK
4、SpringBoot整合RocketMQ****实现消息发送和接收
4.1、添加maven依赖 org.apache.rocketmq rocketmq-spring-boot-starter 2.2.3 4.2、yaml配置 rocketmq: name-server: localhost:9876 producer: # 发送同一类消息的设置为同一个group保证唯一 group: producer-group # 发送消息超时时间默认3000 sendMessageTimeout: 5000 # 发送消息失败重试次数默认2 retryTimesWhenSendFailed: 2 # 异步消息重试此处默认2 retryTimesWhenSendAsyncFailed: 2 # 消息最大长度默认1024 * 1024 * 4(默认4M) maxMessageSize: 4096 # 压缩消息阈值默认4k(1024 * 4) compressMessageBodyThreshold: 4096 # 是否在内部发送失败时重试另一个broker默认false retryNextServer: false consumer: group: consumer-group # 消费者组名 4.3、生产者
Service
public class RocketMQProducer {Autowiredprivate RocketMQTemplate rocketMQTemplate;public void sendMessage(String topic, String message) {rocketMQTemplate.convertAndSend(topic, message);System.out.println(Message sent: message);}
}4.4、消费者
Service
RocketMQMessageListener(topic your-topic, consumerGroup consumer-group)
public class RocketMQConsumer implements RocketMQListenerString {Overridepublic void onMessage(String message) {System.out.println(Received message: message);}
}4.5、接口 GetMapping(/send)public String sendMessage(RequestParam String topic, RequestParam String message) {rocketMQProducer.sendMessage(topic, message);return Message sent: message;}4.6、接口测试 至此测试完成
