最佳实践:如何基于消息服务MNS实现严格有序队列

本文涉及的产品
对象存储 OSS,20GB 3个月
对象存储 OSS,恶意文件检测 1000次 1年
对象存储 OSS,内容安全 1000次 1年
简介:
问题背景
阿里 消息 服务 提供的队列(queue)主要特点是高可靠、高可用、高并发。每个队列的 数据 都会被持久化三份到阿里云的飞天分布式平台;其中每个队列至少有2台 服务器 向外提供服务;同时每台服务器都支持高并发访问。这些分布式特性,也导致了消息服务队列无法像传统单机队列那样保证严格的消息FIFO特点,只能做到基本有序。


我们的队列如果同时有多个消息发送者(sender),由于并发和网络延迟不一等问题,消息的严格顺序本身就是失去了意义,因为在这种情况下,我们根本无法获知消息在多个sender上的实际发送顺序和消息到达服务器端的真实顺序。同样的,当有多个接收者并发接收消息时,其真正的处理顺序也是不可获知的。

因此,我们认为只有一个发送者(一个进程,可以是多个线程), 一个接收者时,消息顺序才有意义,也只有在这种情况下我们能够感知和记录消息的真实发送顺序和消息的真实接收顺序。

解决方案
基于上述假设,同时为了满足部分用户对于消息消费顺序性的要求,我们设计了下面的方案,确保消息按照用户发送顺序被接收和消费。
主要步骤:
1.消息在发送端进行染色,加上SeqId(例如:#num#)
2.消息在接收端进行还原,并根据SeqId 排序后返回给上层,同时对于已经receive的消息会有后台线程保证消息不会被重复消费。
3.为了避免因为发送者fail,或者接收者fail,导致seqid 丢失。seqid 会被持久存储到本地磁盘文件。
   当然也可以存储到其他存储或 数据库 :例如OSS,OTS, RDS
240_1365541074971606_dcf0b3cd382d9d5.png
程序说明
附件提供了python版的方案实现(依赖MNS Python SDK)。其中,主要提供了OrderedQueueWrapper 类(oredered_queue.py文件),可以将普通的mns queue包装成有序queue。
OrderedQueueWrapper  提供两个方法:SendMessageInOrder()和ReceiveMessageInOrder()。send中对消息进行染色,receive还原消息,并且按顺序返回给接收者。

另外,send_message_in_order.py和receive_message_in_order.py提供了发送者和接收者使用OrderedQueueWrapper的程序示例。
send_message_in_order.py:
    #init orderedQueue
    seqIdConfig = {"localFileName":"/tmp/mns_send_message_seq_id"}   # 指定持久化发送SeqId的磁盘文件。
    seqIdPS = LocalDiskStorage(seqIdConfig)
    orderedQueue = OrderedQueueWrapper(myQueue, sendSeqIdPersistStorage = seqIdPS)
    orderedQueue.SendMessageInOrder(message)

receive_message_in_order.py:
    #init orderedQueue
    seqIdConfig = {"localFileName":"/tmp/mns_receive_message_seq_id"} #指定持久化接收SeqId的磁盘文件
    seqIdPS = LocalDiskStorage(seqIdConfig)
    orderedQueue = OrderedQueueWrapper(myQueue, receiveSeqIdPersistStorage = seqIdPS)
    recv_msg = orderedQueue.ReceiveMessageInOrder(wait_seconds)


运行方法:
1.配置send_message_in_order.py 和receive_message_in_order.py 中下列 配置
g_endpoint,g_accessKeyId,g_accessKeySecret,g_testQueueName 
2.运行send_message_in_order.py
3.运行receive_message_in_order.py (可以不用等步骤2程序运行完成)
发送程序会发送20条消息,接收程序会按顺序消费20条消息
240_1365541074971606_9530b6f93c549f4.png


也可以运行oredered_queue.py (需配置endpoint 和AK)的测试case对比普通mns queue的区别:
运行命令:$python oredered_queue.py
非严格有序:(整体有序,部分相邻消息无序,同时侧面证明MNS 的单个queue同时有多个服务器在提供服务)
240_1365541074971606_0a406a4c540529a.png



严格有序:
240_1365541074971606_4971ecb1f2f1a91.png

注意事项:
1.本帖主要目的是展示顺序消息的解决方案,本帖中的代码未经过严格测试,不建议不加测试直接用于生产环境。同时程序仓促完成,难免由瑕疵,欢迎回帖指正。
2.正常情况下,发送端和接收端的seqid应该和queue中的消息(染色)匹配,当出现删除queue重新创建等 操作 时,请注意磁盘文件中的seqid 是否和queue中的真实情况相符,同时建议不要往染色的消息队列里发送非染色消息。
3.队列的消息有效期设置过短或者每条消息的实际处理结果都有可能会对消息有序性造成影响,在您的程序中需要对这些情况所导致的的乱序现象进行处理。


示例程序下载:
zip.gif ordered_queue_wrapper.tar.gz (4 K) 
相关实践学习
RocketMQ一站式入门使用
从源码编译、部署broker、部署namesrv,使用java客户端首发消息等一站式入门RocketMQ。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
13天前
|
消息中间件 存储 NoSQL
rocketmq实现延迟队列思路探讨
本文介绍了两种实现RocketMQ延迟消息的方法。非任意时间延迟可通过在服务器端配置`messageDelayLevel`实现,但需重启服务。任意时间延迟则分为两种策略:一是结合原生逻辑和时间轮,利用RocketMQ的默认延迟等级组合支持任意延迟,但可能丢失1分钟内的数据;二是使用存储介质(如Redis)加时间轮,消息存储和定时发送结合,能处理数据不一致和丢失问题,但涉及更多组件。推荐项目[civism-rocket](https://github.com/civism/civism-rocket)作为参考。
43 1
|
2月前
|
消息中间件 存储 Kafka
几种 MQ 顺序消息的实现方式
几种 MQ 顺序消息的实现方式
|
8月前
|
消息中间件 存储 安全
RocketMQ-消息消费模式 顺序消费
RocketMQ-消息消费模式 顺序消费
156 0
|
9月前
|
消息中间件 运维 Java
rabbitMQ消息中间件的延时队列以及死信队列的使用和应用场景
rabbitMQ消息中间件的延时队列以及死信队列的使用和应用场景
|
消息中间件 存储 Apache
解析 RocketMQ 业务消息--“顺序消息”
本篇将继续业务消息集成的场景,从功能原理、应用案例、最佳实践以及实战等角度介绍 RocketMQ 的顺序消息功能。
234 0
解析 RocketMQ  业务消息--“顺序消息”
|
消息中间件 安全 Serverless
消息队列 MNS 特性|学习笔记
快速学习消息队列 MNS 特性
198 0
|
消息中间件 NoSQL 关系型数据库
实战:如何防止mq消费方消息重复消费、rocketmq理论概述、rocketmq组成、普通消息的发送
实战:如何防止mq消费方消息重复消费 如果因为网络延迟等原因,mq无法及时接收到消费方的应答,导致mq重试。(计算机网络)。在重试过程中造成重复消费的问题
2492 1
实战:如何防止mq消费方消息重复消费、rocketmq理论概述、rocketmq组成、普通消息的发送
|
消息中间件 RocketMQ
RocketMQ怎么保证队列顺序消费
RocketMQ怎么保证队列顺序消费
257 0
RocketMQ怎么保证队列顺序消费
|
消息中间件 负载均衡 RocketMQ
RocketMQ-顺序消息
RocketMQ-顺序消息
RocketMQ-顺序消息
|
消息中间件 存储 Apache
解析 RocketMQ 业务消息--“顺序消息”
本篇将继续业务消息集成的场景,从功能原理、应用案例、最佳实践以及实战等角度介绍 RocketMQ 的顺序消息功能。
解析 RocketMQ 业务消息--“顺序消息”