TableStore Timeline:轻松构建千万级IM和Feed流系统

本文涉及的产品
对象存储 OSS,20GB 3个月
日志服务 SLS,月写入数据量 50GB 1个月
对象存储 OSS,内容安全 1000 次 1年
简介: 在文章《现代IM系统中消息推送和存储架构的实现》中介绍了一种适用于IM的消息存储和推送模型Timeline,在本篇文章中,会扩展Timeline模型到IM和Feed流系统中,并且提供成熟的LIB实现。用户基于TableStore-Timeline LIB可轻松实现千万级的IM和Feed流系统。

场景

移动互联网时代,微信和微博已经成为这个时代的两大支柱类社交应用。
这两类应用,其中一个是IM产品,一个是Feed流产品,微信的朋友圈也属于Feed流。
如果再细心去发现,会发现基本所有移动App都有Feed流的功能:消息广场、个人关注、通知、新闻聚合和图片分享等等。各种各样的Feed流产品占据了我们生活的方方面面。

现状

IM和Feed流功能已经基本成为所有App标配,如何开发一个IM或者Feed流功能是很多架构师、工程师要面临的问题。虽然是一个常见功能,但仍然是一个巨大的挑战,要考虑的因素非常多,比如:

  1. 存储系统里面如何选型才能支持大量数据存储,而且价格便宜。
  2. 同步系统使用何种架构、推送模型和系统才能保证高并发的同时延迟稳定?

为了解决上述问题,我们之前推出了三篇文章来阐述:

上述三篇文章推出后,用户反响很好,在各个平台的传播很广,为很多用户提供了设计一款IM和Feed流产品的架构思路,但是从这里到完全实现一个可靠的IM、Feed流系统平台还有很长的路,比如:

  1. 如何才能更加简单的弄懂消息存储和同步模型?
  2. 如何将架构模型映射到存储系统和推送系统中?
  3. 如何才能保证对存储系统和推送系统的使用是最佳方式?

Timeline模型

针对上述三个问题,在《现代IM系统中消息推送和存储架构的实现》中引入了一个逻辑模型概念:Timeline模型。

在《现代IM系统中消息推送和存储架构的实现》中基于IM系统提出了Timeline模型,进一步会发现Timeline模型适用场景可以更广泛:

  • 有一方是消息生产者。
  • 有一方是消息消费者。
  • 生产者产生的一条消息可能会被一个或多个消费者消费。
  • 消费者需要聚合来自多个生产者的消息在一个页面展现。

IM和Feed流产品完全匹配上述四个特征,所以Timeline模型可以完全适用于IM和Feed流场景中。

模型适配性

下面我们来看看如何在各个场景中使用Timeline:

IM

  • 单聊就是三个Timeline:

    • 会话Timeline(存储历史消息)
    • 用户A的收件箱(A的同步库Timeline)
    • 用户B的收件箱(B的同步库Timeline)
  • 群聊就是1 + N个Timeline:

    • 会话Timeline(存储历史消息)
    • 用户A的收件箱(A的同步库Timeline)
    • 。。。。。。
    • 用户N的收件箱(N的同步库Timeline)

每个用户只有一个同步库Timeline,就算用户A在10个群里面,那么这个10个群的同步消息都是发送给用户A的这一个同步库Timeline中。

朋友圈

  • 发一条朋友圈状态就是1 + N个Timeline:

    • 自己历史消息Timeline(自己的存储库Timeline)
    • 朋友A的收件箱(A的同步库Timeline)
    • 。。。。。。
    • 朋友N的收件箱(N的同步库Timeline)

微博

如果是微博,粉丝可能会达到上亿级别,这时候会比朋友圈稍微复杂些:

  • 大V发一条微博就是 1 + M个Timeline(M << N,N是粉丝数)。

    • 自己历史消息Timeline(自己的存储库Timeline)
    • 粉丝A的收件箱(A的同步库Timeline)
    • 。。。。。。
    • 粉丝M的收件箱(M的同步库Timeline)
      剩余N - M的粉丝直接读大V自己的存储库Timeline内容即可,那么怎么设置M和N - M,可以参考《如何打造千万级Feed流系统》这篇文章。

从上面分析可以看出来,不管是IM,还是Feed流产品都可以将底层的存储、同步逻辑抽象成一个对多个Timeline进行读写的模型。

问题

有了Timeline概念模型后,从IM/Feed流应用映射到Timeline就比较容易了,但是从Timeline映射到存储、同步系统仍然很复杂,主要体现在:

  • 如何实现Timeline概念模型?
  • 如何将Timeline模型转换成对存储系统、同步系统的读写接口?
  • 如何设计存储系统、同步系统的表结构?
  • 如何选择存储系统、同步系统的读写方式?
  • 如何评估存储系统、同步系统的最大承载能力?
  • 如何实现才能保证性能最佳?
  • 如何才能吸取大型同类型系统架构设计的经验教训、
  • 如何才能避免一些实现、使用上的隐患?

这些问题涉及的内容光,细节多,深度大,坑较多等,整体上很繁杂,这一部分在耗费了大量人力之后,结果可能并不理想。

Timeline LIB

针对上述问题,只要存储系统和推送系统确定后,剩余的工作都是类似的,可以完全将经验封装起来成为一个LIB,将表结构设计,读写方式,隐患等等都解决好,然后供后来者使用,后来者可以不用再关心Timeline到底层存储系统之间的事情了。

所以,我们基于JAVA语言实现了一个TableStore-Timeline LIB,简称Timeline LIB。

目前已经开源在了GitHub上:Timeline@GitHub

Timeline LIB的结构如下:
timeline1

整个Timeline分为两层,上层的Timeline层和下层的Store层。

Timeline层,提供最终的读写接口,用户操作的也是Timeline的接口。

Store层,负责存储系统的交互,目前Timeline LIB中提供了DistributeTimelineStore,基于Table Store,同时实现了分布式的存储和同步。后续会继续实现GlobalTimelineStore等。如果有用户有其他系统需求,比如MySQL,Redis,可以通过实现IStore接口来新增MySQLStore和RedisStore。

也欢迎大家将自己实现的Store通过GitHub的PullRequest共享出来。

有了Timeline LIB之后,如果要实现一个IM或者Feed流,只需要创建两种类型Timeline(存储类,同步类),然后调用Timeline的读写接口即可。

接口

接下来,我们看下Timeline LIB的API。
Timeline LIB中面向最终用户的是Timeline类,用于对每个Timeline做读写操作。

Timeline的接口主要分为三类:

  • 写:

    • store:同步写入
    • storeAsync:异步写入
    • batch:批量写入
  • 读:

    • get:同步读
    • getAsync:异步读
  • 范围读:

    • scan:同步范围读取

定义

   /**
     * Timeline的构造函数。
     * @param timelineID    此Timeline对应的ID。唯一标识一个Timeline,需要全局唯一,如果业务场景中需要多个字段才能唯一标识一个TimelineID,此时可以将多个字段拼接成一个字段。
     * @param store         此Timeline关联的Store,一般为存储Store或同步Store。实现了IStore接口类的对象,目前LIB中默认实现了DistributeTimelineStore,可以使用此store。除此之外,用户还可以自己实现自己的Store类,用于适配其他系统。
     */
    public Timeline(String timelineID, IStore store);

    /**
     * 写入一个消息到此Timeline中。
     * @param message   消息对象,需实现IMessage接口。用户需要通过实现IMessage接口创造符合自己业务场景的消息类,LIB中默认实现了StringMessage类。
     * @return          完整的TimelineEntry,包括消息和顺序ID。
     */
    public TimelineEntry store(IMessage message);
    
    /**
     * 异步写入消息接口。
     * @param message     消息对象,需实现IMessage接口。
     * @param callback    回调函数。
     * @return            Future对象,异步模式下,Future和callback需要二选一。
     */
    public Future<TimelineEntry> storeAsync(IMessage message, TimelineCallback<IMessage> callback);    
    
   /**
     * 批量写入消息接口。
     * 此接口只是把消息加入到本地的一个buffer中,当buffer满或者超时(默认10s,可配置)才会统一写入。
     * 此接口返回时并不一定消息已经写入成功。
     * @param message 消息对象,需实现IMessage接口。
     */
    public void batch(IMessage message) {

   /**
     * 同步读取接口,通过制定一个唯一的顺序ID读取目标TimelineEntry。
     * @param sequenceID    顺序ID。此顺序ID可由store或scan接口获取到。
     * @return              完整的TimelineEntry,包括消息和顺序ID。
     */
    public TimelineEntry get(Long sequenceID);
    
    /**
     * 异步读取接口,通过制定一个唯一的顺序ID读取目标TimelineEntry。
     * @param sequenceID    顺序ID。
     * @param callback      读取结束后的回调函数。
     * @return              Future对象,异步模式下,Future和Callback需要二选一。
     */
    public Future<TimelineEntry> getAsync(Long sequenceID, TimelineCallback<Long> callback);

    /**
     * 顺序读取一段范围内或固定数目的消息,支持逆序,正序。
     * @param parameter     顺序读取的参数,包括方向、from、to和maxCount。
     * @return              TimelineEntry的迭代器,通过迭代器可以遍历到待读取的所有消息。
     */
    public Iterator<TimelineEntry> scan(ScanParameter parameter);
AI 代码解读

get和scan

  • get接口一般用于读取单条消息,在IM和Feed流中可使用的场景非常少,甚至可以不使用。
  • scan是读取消息或Feed流消息的主要途径,通过scan可以读取到已经产生,但还未消费的消息。

如何用

简介

安装

在 Maven 工程中使用 Timeline LIB 只需在 pom.xml 中加入相应依赖即可:

<dependency>
    <groupId>com.aliyun.openservices.tablestore</groupId>
    <artifactId>timeline</artifactId>
    <version>1.0.0</version>
</dependency>
AI 代码解读

实现自己的IMessage

使用之前,需要先实现一个满足自己业务特点的Message类,此Message类能表示业务中的一条完整消息。
需要实现IMessage的下列接口:

  • String getMessageID():

    • 生成一个消息ID。
    • 消息ID的作用主要是用于消息去重,最大使用场景是IM中的多人群。对于Timeline模型,消息ID只需要在一段时间内(比如1小时或1天内)当前会话和收件人的消息中唯一即可。比如在IM中,只需要在某个会话或者群里面唯一即可,这时候其实更好的方式是由客户端生成这个消息ID。最简单的方式是客户端循环使用0~10000之间的值和用户ID拼接后作为消息ID即可满足要求。
    • 如果不想自己生成消息ID,可以直接继承DistinctMessage,DistinctMessage类的消息ID = machineID + 进程ID + 循环递增ID[0, Integer.MAX_VALUE]。
  • void setMessageID(String messageID):

    • 在get和scan接口中设置消息ID。
  • IMessage newInstance():

    • 使用默认构造函数,生成一个同类型实例。用于在Store中自动生成同类型实例对象。
  • byte[] serialize():

    • 需要将消息内容完整序列号为字节数组。这个接口会被store接口在写入的时候调用。
  • void deserialize(byte[] input):

    • 反序列化接口,这个接口会被store在读取的时候调用。

生成Store

在一个IM或Feed流产品中,一般会有两个子系统,一个是存储系统,一个是同步系统。

需要为这两个系统各自生成一个Store对象。

  • 存储系统Store:存储数据时间长,数据量大,成本很重要。如果不使用读写扩散相结合的方式,那么存储系统的store可以使用成本更低的混合存储(SSD + SATA)。
  • 同步系统Store:存储受时间段,但是延时敏感,可以使用高性能存储(SSD),读取性能更稳定。
    如果首次使用,需要调用store的create接口创建相应存储表。

Timeline读写

Store生成好后就可以构造最终的Timeline对象了,Timeline对象分为两类,一类是存储库Timeline,一个是同步库Timeline。

当在IM中发布消息或者Feed流产品中发布状态时,就是对相应存储库Timeline和同步库Timeline的消息写入(store/storeAsync)。

当在IM或Feed流产品中读取最新消息时,就是对相应同步库Timeline的范围读取(scan)。

当在IM或Feed流产品中读取历史消息时,就是对相应存储库Timeline的范围读取(scan)。

如果是推拉结合的微博模式,则读取最新消息时,就是对相应存储库Timeline和同步库Timeline的同时范围读取(scan)。

错误处理

Timeline LIB中会抛出TimelineException,TimelineException提供了两种接口:getType()和getMessage(),getType()返回此TimelineException的类型,包括了TET_ABORT,TET_RETRY,TET_INVALID_USE,TET_UNKNOWN:

  • TET_ABORT:目前只有内部OutputStrem被非预期关闭后才会出现。
  • TET_RETRY:网络不稳定或者底层存储系统在负载均衡,可以继续做重试。
  • TET_INVALID_USE:使用方式不对,建议直接报错进程推出。
  • TET_UNKNOWN:未知错误,可以到Timeline@GitHub给我们提issue。

示例

这一节会演示下如何使用Timeline LIB实现IM的群组功能。

场景

  • 多人群组,号码是11789671。
  • 群里有用户:user_A,user_B,user_C。

初始化

构造两个store,一个用来存储,一个用例同步。

IStore store = new DistributeTimelineStore(storeConfig);
IStore sync = new DistributeTimelineStore(syncConfig);

// 构造群成员列表,群成员列表可以存储在Table Store。
List<String> groupMembers = Arrays.asList("user_A", "user_B", "user_C");
AI 代码解读

发消息

user_A发一条群消息:“有人吗”。

// 存储会话消息
Timeline timeline1 = new Timeline("11789671", store);
timeline1.store(new StringMessage("user_A:有人吗"));

// 发送同步消息
for (String user : groupMembers) {
    Timeline timeline = new Timeline(user, sync);
    timeline.store(new StringMessage("user_A:有人吗"));
}
AI 代码解读

读取同步消息

user_C读取自己最新的同步消息

Timeline timeline = new Timeline("user_C", sync);
ScanParameter scanParameter = ScanParameterBuilder
                .scanForward()
                .from(last_sequence_id)
                .to(Long.MAX_VALUE)
                .maxCount(100)
                .build();
                
Iterator<TimelineEntry> iterator = timeline.scan(scanParameter);
while(iterator.hasNext()) {
    TimelineEntry entry = iterator.next();
    // 处理消息
}
AI 代码解读

上面的示例演示了如何用Timeline LIB实现IM中的群组功能。其他的朋友圈,微博等也类似,这里就不赘述了。

我们目前在Timeline Samples@GitHub上实现两个场景的实例:

也欢迎大家共享其他场景的实现代码。

性能

我们使用阿里云ECS做了性能测试,效果较理想。
不同接口的写入性能上,批量batch接口最快,其次是异步writeAsync,最后是同步write接口。

在阿里云共享型1核1G的ECS机器上,使用DistributeTimelineStore,Timeline LIB的storeAsync接口可以完成每秒1.2万消息的写入,如果使用batch批量接口,则可以完成每秒5.3万消息的写入。

如果使用一台8核的ECS,只需要3秒钟就可以完成100万条消息的写入。

由于DistributeTimelineStore使用了Table Store作为存储和同步系统,Table Store是阿里云的一款服务化NoSQL服务,支持的TPS在理论上无上限,实际中仅受限于集群大小,所以整个Timeline LIB的写入能力和压力器的CPU成正比。

下面的图展示了不同机型上完成1000万条消息写入的延迟:
timeline2

在一台8核ECS上只需要27秒就可以完成1000万写入,由于写入能力和CPU成线线性关系,如果用两台16核的,则只需要7秒就可以完成1000万消息的写入。

我们再来看一下scan读取的性能,读取20条1KB长度消息,LIB端延迟一直稳定在3.4ms,Table Store服务端延迟稳定在2ms。

这个量级和能力,可以撑得住目前所有的IM和Feed流产品的压力。

和TableStore关系

Timeline LIB的想法来源于Table Store的真实场景需求,并且为了用户可以更加简单的使用,增加了主键列自增功能。

目前Timeline LIB的store层实现了DistributeTimelineStore类,DistributeTimelineStore可同时适用于存储store和同步store。

DistributeTimelineStore是基于Table Store的,但是为了便于用户使用其他系统,在Timeline LIB中将Store层独立了出来。

如果用户希望使用其他系统,比如MySQL作为存储系统,可以实现IStore接口构造自己的Store类。我们也欢迎大家提供自己的各种Store层实现,最终希望为社交场景的架构师和开发者提供一套完整的易用性开发框架。

接下来计划

  • Store的扩展:

    • 目前仅支持一种store:DistributeTimelineStore,虽然可以同时支持存储和同步,但是功能还是比较简单,接下来会支持更多的Store,比如全球多向同步的GlobalTimelineStore等。
  • 更易用性的接口

    • 目前Timeline接口比较偏向系统层,后续会提供更偏向业务层的接口,比如适用于IM的接口,适用于Feed流的接口等。

联系我们

如果在使用过程中有任何问题或者建议,可以通过下列途径联系我们:

目录
打赏
0
2
1
8
3249
分享
相关文章
智能客服系统的技术栈解析-唯一客服系统技术架构优势
“唯一客服系统”采用 Vue.js 2.x + ElementUI 构建前端,实现响应式界面,支持多端适配;后端基于 Golang + Gin + GORM,具备高性能与高并发处理能力。系统支持私有化部署,提供灵活定制、AI 扩展能力,技术栈简洁易维护,兼顾开发者友好与企业级应用需求。
25 1
可二次开发的在线客服系统-前后端混合渲染模式
服务端渲染(SSR)结合API交互,提升首屏加载速度与SEO友好性,适用于混合渲染模式的Web应用。
18 0
客服系统命令行程序-Cobra 命令行应用入口
唯一客服系统是基于 Go 语言与 Cobra 框架构建的命令行工具,用于管理在线客服系统。支持启动、安装和停止服务,具备清晰的命令结构和良好的扩展性,便于维护与功能拓展。
22 0
|
7天前
|
客服系统程序入口文件解析-唯一客服系统源码开发
该代码为 Go 语言编写的客服系统命令行程序入口,结构清晰,使用 cmd 包启动业务逻辑,可能基于 cobra 框架实现,具备良好可扩展性与可维护性,适用于服务启动与管理。
119 69
纯血鸿蒙NEXT即时通讯/IM系统:RinbowTalk正式发布,全源码、纯ArkTS编写
RainbowTalk是一套基于MobileIMSDK的产品级鸿蒙NEXT端IM系统,目前已正式发布。纯ArkTS、从零编写,无套壳、没走捷径,每一行代码都够“纯”(详见:《RainbowTalk详细介绍》)。 MobileIMSDK是一整套开源IM即时通讯框架,历经10年,超轻量级、高度提炼,一套API优雅支持 UDP 、TCP 、WebSocket 三种协议,支持 iOS、Android、H5、标准Java、小程序、Uniapp、鸿蒙NEXT,服务端基于Netty编写。
60 1
AI 智能客服搭建实战:如何建立一个企业级智能客服系统?
2025 年全球 AI 客服市场规模超 800 亿美元,企业面临意图识别误差、多语言支持等挑战。合力亿捷通过混合云架构、双引擎驱动等四层技术架构,结合小样本微调 + 主动学习等策略,实现服务效率提升 50% 以上、人工成本降低 40%,助力企业突破 “人工智障” 困境。
196 3
深度解析:AI语音客服系统如何重塑客户服务体验与主流解决方案探析
在数字化浪潮下,AI语音客服凭借高效、便捷、24小时在线的优势,成为企业提升服务效率、优化体验的重要工具。本文详解其核心技术、应用价值、选型要点及市场主流方案,如阿里云通义晓蜜、合力亿捷等,助力企业智能化升级。
120 1
基于通义大模型的智能客服系统构建实战:从模型微调到API部署
本文详细解析了基于通义大模型的智能客服系统构建全流程,涵盖数据准备、模型微调、性能优化及API部署等关键环节。通过实战案例与代码演示,展示了如何针对客服场景优化训练数据、高效微调大模型、解决部署中的延迟与并发问题,以及构建完整的API服务与监控体系。文章还探讨了性能优化进阶技术,如模型量化压缩和缓存策略,并提供了安全与合规实践建议。最终总结显示,微调后模型意图识别准确率提升14.3%,QPS从12.3提升至86.7,延迟降低74%。
332 14
2025年国内主流智能客服系统:技术架构与能力深度解析
本文分析了2025年国内智能客服市场的技术格局与系统能力,从核心技术栈(NLP、知识图谱、语音技术等)到市场梯队划分,深入探讨了第一梯队的综合型平台和第二梯队的场景化服务。以合力亿捷为例,剖析其端到端AI技术栈、大模型融合、全渠道融合及运营优化能力,并对比国际顶尖通用AI在语义理解、多模态交互和自主学习方面的启示。最后为企业提供选择智能客服系统的五大考量因素,强调技术与业务场景的深度融合,助力企业实现更高效、智能的客户服务体验。
345 2
工单系统【WGCAT】如何设置工单类型和标签
**简介**:WGCAT 是一款免费、易用的轻量级工单系统,适合多种业务场景。通过修改配置文件 `config/application.yml` 中的 `workType` 参数,可自定义工单类型(如故障处理、安装服务等),重启程序后生效。此外,创建工单时还可设置标签以增强分类管理。下载地址:www.wgstart.com。

云存储

+关注
AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等