文章目录前言一、rocketmq是什么二、rocketmq核心概念三、rocketmq核心应用四、go如何使用rocketmq总结前言

当我们的业务达到一定规模很多业务需要解耦以及需要流量削峰的时候我们需要使用MQ来让我们系统能够正常运转。

RocketMQ是由阿里捐赠给Apache的一款低延迟、高并发、高可用、高可靠的分布式消息中间件。

经历了淘宝双十一的洗礼。

RocketMQ既可为分布式应用系统提供异步解耦和削峰填谷的能力同时也具备互联网应用所需的海量消息堆积、高吞吐、可靠重试等特性。

二、rocketmq核心概念

Topic消息主题一级消息类型生产者向其发送消息。

Message生产者向Topic发送并最终传送给消费者的数据消息的载体。

消息属性生产者可以为消息定义的属性包含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

三、rocketmq核心应用

削峰填谷诸如秒杀、抢红包、企业开门红等大型活动时皆会带来较高的流量脉冲或因没做相应的保护而导致系统超负荷甚至崩溃或因限制太过导致请求大量失败而影响用户体验消息队列RocketMQ可提供削峰填谷的服务来解决该问题。

异步解耦交易系统作为淘宝和天猫主站最核心的系统每笔交易订单数据的产生会引起几百个下游业务系统的关注包括物流、购物车、积分、流计算分析等等整体业务系统庞大而且复杂消息队列RocketMQ可实现异步通信和应用解耦确保主站业务的连续性。

顺序收发细数日常中需要保证顺序的应用场景非常多例如证券交易过程时间优先原则交易系统中的订单创建、支付、退款等流程航班中的旅客登机消息处理等等。

与先进先出FIFOFirst

First

Out原理类似消息队列RocketMQ提供的顺序消息即保证消息FIFO。

分布式事务一致性交易系统、支付红包等场景需要确保数据的最终一致性大量引入消息队列RocketMQ的分布式事务既可以实现系统之间的解耦又可以保证最终的数据一致性。

大数据分析数据在“流动”中产生价值传统数据分析大多是基于批量计算模型而无法做到实时的数据分析利用阿里云消息队列RocketMQ与流式计算引擎相结合可以很方便的实现业务数据的实时分析。

分布式缓存同步天猫双11大促各个分会场琳琅满目的商品需要实时感知价格变化大量并发访问数据库导致会场页面响应时间长集中式缓存因带宽瓶颈限制了商品变更的访问流量通过消息队列RocketMQ构建分布式缓存实时通知商品数据的变化。

四、go如何使用rocketmq

github.com/apache/rocketmq-client-go/v2初始化消费者

package

(contextfmtgin/common/alarmgin/common/functionconsumer2

gin/config/extra/consumergin/libgithub.com/apache/rocketmq-client-go/v2github.com/apache/rocketmq-client-go/v2/consumergithub.com/apache/rocketmq-client-go/v2/primitivegithub.com/apache/rocketmq-client-go/v2/producergithub.com/apache/rocketmq-client-go/v2/rloggithub.com/tidwall/gjsonreflectsync

)var

rocketmq.NewProducer(producer.WithNsResolver(primitive.NewPassthroughResolver([]string{127.0.0.1:9876})),producer.WithRetry(2),producer.WithGroupName(gin-hong-api),)err

p.Start()if

p.SendSync(context.Background(),

msg)if

rocketmq2.NewPushConsumer(consumer.WithGroupName(gin-hong-api),consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{127.0.0.1:9876})),)if

err

new(sync.WaitGroup)wg.Add(1)defer

func()

function.ErrorToString(err)trace

function.PrintStackTrace()alarm.DingDing(errMsg

func(ctx

string(message.Body)fmt.Println(msg)}return

nil})err

一个消费者组中的功能应该一致防止遗漏消费一个Topic中的所有队列和一个消费者组中的消费者是多对一的关系防止重复消费一个消费者组对应消费者集群一个消费者对应一个进程。

同组中的消费者可以是一个机器中的多个进程也可以是多个机器中的进程(建立集群防止宕机导致服务不可用)在一个进程中可以并发运行不同消费者组的消费者实例但不能并发运行同组的消费者线程。

Go

SDK内部也没有提供单例线程数的设置。

一个消费者组可以订阅多个Topic所有消费者组由nameserve统一管理

由于rocketmq

包使用pullConsumer还没实现。

所以目前只能使用Subscribe来消费。