网站改版 301跳转,市场营销策略,餐厅网站建设文案书,在线做c 题的网站阿里妹导读#xff1a;消息类场景是表格存储#xff08;Tablestore#xff09;主推的方向之一#xff0c;因其数据存储结构在消息类数据存储上具有天然优势。为了方便用户基于Tablestore为消息类场景建模#xff0c;Tablestore封装Timeline模型#xff0c;旨在让用户更快… 阿里妹导读消息类场景是表格存储Tablestore主推的方向之一因其数据存储结构在消息类数据存储上具有天然优势。为了方便用户基于Tablestore为消息类场景建模Tablestore封装Timeline模型旨在让用户更快捷的实现消息类场景需求。在推出Timelinev1、v2两个版本模型以来受到了大量用户关注。但依然会有用户困惑“框架、结构、模型等概念介绍了这么多该如何基于Timeline模型实现具体场景呢”。本文详细讲解如何实现一个简易的IM系统。 梗概
生活中最常见的即时聊天类软件如钉钉、微信等都可以描述为实现了即时通讯能力的聊天工具。其中聊天会话可分为两大类分别是单聊、群聊公众号类似单聊。这里我们以钉钉Ding Talk的功能为参照详细说明相应的功能基于Tablestore的Timeline模型如何实现。如新消息提醒未读消息数统计查看会话中更久的聊天内容群名模糊检索关键字查询历史记录以及多客户端同步等。让用户在实现方案上有更清晰的认识对模型的抽象概念、接口有更好的理解。
下面会按照聊天系统的功能模块分段分别介绍每一部分的功能、方案介绍、表设计以及实现代码等。功能模块主要分为消息存储、关系维护、即时感知、多端同步。
功能模块
消息存储
消息系统中消息存储是最基本的功能。对于消息存储提供消息的读、写、持久化一方面需要持久化写入保证消息数据的不丢失另一方面适合用户的快速、高效查询。在IM场景中写入方式通常是单行、批量写入而读取需要按照消息队列范围读取。有时用户还有对于历史消息的模糊查询需求这时就需要使用多维检索、全文检索的能力。
样例中消息数据的表结构见下图表设计im_timeline_store_table 存储库
功能会话窗口消息展示
存储库是聊天会话消息所对应的存储表消息以会话分类存储每个会话是一个消息队列。单个消息队列TimelineQueue通过timelineId唯一标识所有消息基于sequenceId有序排列。消息体中含有发送人、消息id消息去重、消息发送时间、消息体内容、消息类型类型包含图片、文件、普通文本本文仅适用文本等。 如上图当用户点击某一个会话时窗口会展示相应会话的最新一页消息。图片里的消息都是从存储库拉取的通过timelineId获取该会话的Queue实例然后调用Queue的scan接口与ScanParam参数sequenceId范围倒序拉取最新的一页消息。当用户向上滚动展示完这一页消息后客户端会基于第一次请求的最小sequencId发起第二次请求获取第二页消息记录单页消息数通常选择20-30条。会话的消息可以选择在客户端持久化然后在感知到新消息之后更新本地消息增加缓存减少网络IO。
|核心代码 public ListAppMessage fetchConversationMessage(String timelineId, long sequenceId) {TimelineStore store timelineV2.getTimelineStoreTableInstance();TimelineIdentifier identifier new TimelineIdentifier.Builder().addField(timeline_id, timelineId).build();ScanParameter parameter new ScanParameter().scanBackward(sequenceId).maxCount(30);IteratorTimelineEntry iterator store.createTimelineQueue(identifier).scan(parameter);ListAppMessage appMessages new LinkedListAppMessage();while (iterator.hasNext() counter 30) {TimelineEntry timelineEntry iterator.next();AppMessage appMessage new AppMessage(timelineId, timelineEntry);appMessages.add(appMessage);}return appMessages;} 存储库的消息需要永久保存是整个应用的全量消息存储。存储库数据过期时间TTL需要设为-1。
|功能多维组合、全文检索
全文检索能力就是对存储库的消息内容做模糊查询因而需要对存储库的数据建立多元索引。具体索引字段需要根据设计需求设计。如钉钉公开群的检索需要对群ID、消息发送人、消息类型、消息内容、以及时间建立索引其中消息内容需要使用分词字符串类型从而提供模糊查询的能力。 |核心代码
public ListAppMessage fetchConversationMessage(String timelineId, long sequenceId) {TimelineStore store timelineV2.getTimelineStoreTableInstance();TimelineIdentifier identifier new TimelineIdentifier.Builder().addField(timeline_id, timelineId).build();ScanParameter parameter new ScanParameter().scanBackward(sequenceId).maxCount(30);IteratorTimelineEntry iterator store.createTimelineQueue(identifier).scan(parameter);ListAppMessage appMessages new LinkedListAppMessage();int counter 0;while (iterator.hasNext() counter 30) {TimelineEntry timelineEntry iterator.next();AppMessage appMessage new AppMessage(timelineId, timelineEntry);appMessages.add(appMessage);}return appMessages;
} 另外为了做消息的权限管理仅允许用户检索自己有权限查看的消息可在消息体字段中扩展接收人ID数组这样对所有群做检索时需要增加接收人字段为自己的用户ID这一必要条件即可实现消息内容的权限限制。样例中没有实现这一功能用户可根据需求自己增加、修改。
同步库
|功能新消息即时统计
当客户端在线时应用的系统服务会维护客户端的长连接因而可以感知客户端在线。当用户的同步库有新消息写入时即有新消息应用会发出信号通知客户端有新消息然后客户端会基于同步库checkpoint点拉取同步库中该sequenceId之后的所有新消息统计各会话的新消息数并更新checkpoint点。 如上图对于一个在线客户端每个会话都会维护一个未读消息的计数小红点也会有一个总未读数的计数这个数量一般会存储在客户端本地或者通过redis持久化。这些未读消息指的就是通过同步库拉取并统计过但是还未被用户点开的消息数量。在拉取到新消息列表后客户端或应用层会遍历所有新消息然后将新消息所对应会话的未读计数累加1这样实现了未读消息的即时感知与更新。只有当用户点开会话后会话的未读计数才会清零。
在更新未读数的同时会话列表中还会有最新消息的简短摘要信息以及最新消息的发送时间等。这些可以在遍历新消息列表时不断更新。这些统计、摘要都是依托同步库而非存储库实现的。
|核心代码
public ListAppMessage fetchSyncMessage(String userId, long lastSequenceId) {TimelineStore sync timelineV2.getTimelineSyncTableInstance();TimelineIdentifier identifier new TimelineIdentifier.Builder().addField(timeline_id, userId).build();ScanParameter parameter new ScanParameter().scanForward(lastSequenceId).maxCount(30);IteratorTimelineEntry iterator sync.createTimelineQueue(identifier).scan(parameter);ListAppMessage appMessages new LinkedListAppMessage();int counter 0;while (iterator.hasNext() counter 30) {AppMessage appMessage new AppMessage(userId, iterator.next());appMessages.add(appMessage);}return appMessages;
}
在统计到会话列表中不存在的会话时客户端会做一次额外请求。通过timelineID获取会话的基本描述信息如群头像或好友的头像、群名称等并初始化未读数计时器0然后累加新消息数、更新最新消息摘要等。 同步库对于IM场景下的新消息即时感知统计这一核心功能就是通过写入冗余的方式提升新消息读取统计的效率与速度。对于IM场景没有收件箱的概念因而同步库中冗余消息并没有永久保存的价值提供7天过期时间已经足够保证功能正常。用户可以根据自身需求调整同步库的数据过期时间TTL。 |功能异步写扩散
在本文的样例中单聊会话的消息在写完存储库后同时写入了同步库只有两行的写入开销很小。但是对于群会话写完存储库后要获取群用户列表然后依次写入相应用户的同步库。这种方式在群少、用户少时不会有问题但随着用户体量、活跃度的增加同步的写的方式就会面临性能问题因此建议用户对群写扩散使用异步任务实现。
用户可以基于表格存储实现一个任务队列将写扩散任务写入队列中后直接返回然后由其他进程保证任务队列的执行。任务队列保存了群ID、消息的完整信息消费进程不断轮询读取新任务获取任务后才会从群关系表中获取完整的群成员列表并做相应的写扩散。
任务队列可以直接基于Tablestore实现表设计为两列主键第一列为topic第二列为自增列一个topic对应一个队列任务会被有序写入单个队列中。当并发量持续膨胀后可对任务做hash分桶随机写入多个topic。这样可以增加消费者数量消费并发量提升写扩散效率。对应任务队列消费用户只需要维护每个topic的checkpoint点。checkpoint点之前的为已完成任务通过getRange的方式顺序获取checkpoint点之后未执行的新任务保证任务的执行。失败的任务可以重新写入任务队列来提升容错并增加重试计数。出现多次失败后放弃重写然后将该任务写入特殊的问题队列方便应用的开发者们查询、定位问题。
元数据管理
所谓元数据就是描述数据的数据。在这里主要体现为两类用户元数据、会话元数据。这里群的元数据信息群ID复用群的timelineId、群名称、创建时间等信息可以直接基于timelineMeta的管理表完成实现所有Group类型的TimelineMeta可以映射为一个Group。但是用户的元数据却不能复用TimelineMeta所以需要单独的表实现。
|用户元数据
即用户的属性信息通过用户ID识别特定用户。在上面提到的用户关系中通过用户的标识ID确认用户身份但用户的属性信息如性别、签名、头像等信息还是需要单独维护。因此需要单独维护。
表设计im_user_table 用户元数据以user_id为标识与同步库中的timeline_id一一对应。用户同步新消息时只会拉取同步库中自己对应的单个消息队列TimelineQueue。因此为了唯一ID的方便管理我们可以选择user_id与用户同步库的timeline_id使用同一个值。这样一来在消息写扩散时只需知道群内用户的user_id列表回好友user_id即可以完成写扩算。 |功能用户检索
对于用户添加好友的需求有很多种这里我们只需要维护用户表并且创建多元索引即可轻松实现。样例中没有实现用户可以根据自己需求配置不同的索引字段设置这里我们仅简单分析一下需求
通过用户ID主键查询二维码含用户ID信息主键查询用户姓名多元索引用户名字段设置分词字符串用户标签多元索引数组字符串索引提供签检索、嵌套索引提供多标签打分检索排序附近的人多元索引GEO索引查询附近、特定地理围栏的人
|会话元数据
即会话的属性信息通过唯一会话ID识别特定会话属性信息会包含会话类别群、单聊、公众号等、群名称、公告、创建时间等。同时通过群名称模糊查找群也会是会话元数需要的重要能力。
在Timeline模型中提供了Timeline Meta的管理能力只需通过相应的接口便可实现会话meta的管理。 存储库中管理的是会话的消息队列TimelineQueue这里与会话元数据中的行一一对应。客户端用户选中特定会话后应用从相应的消息队列倒序批量拉取消息展示到客户端群聊单聊的使用方式一样因而并不做会话类型的区分。
|功能群检索
用户如果有加入群的需求首先需要查询到特定的群。查询群的方式与用户查询方式类似功能也可以做相同的实现。用户可以根据自己需求定制不同的索引字段设置需求实现方式如下
群ID主键查询二维码含用户ID信息主键查询群名多元索引用户名字段设置分词字符串群标签多元索引数组字符串索引提供签检索、嵌套索引提供多标签打分检索排序
注会话元数据可以直接维护单聊会话与人的映射关系。对于单聊的meta增加一列users字段存放两个用户ID这样不用额外维护关系表基于单聊关系表im_user_relation_table创建timeline_id为第一列主键的二级索引。
关系维护
完成了元数据管理以及用户和群的检索剩下的就是如何添加好友、加入群聊了。这里就涉及到IM体统中另一个重要的功能点。关系维护包含人与人的关系、人与群的关系以及人与会话的。下面我们介绍如何基于Tablestore解决这一关系维护的需求。
单聊关系
|功能人与单聊会话的关系
单聊场景下参与者仅有两个人同时不考虑顺序。无论是我联系小明或是小明联系我对应的会话必须有且仅有一个。如果使用表格存储维护这个关系建议用如下的设计方式。
第一列为主用户ID、第二列为次用户ID在两个人成为好友后关系表中需要插入两行数据分别以自己的用户ID为main_user以好友的用户ID为sub_user然后将共同的会话timline_id作为属性列并且可以维护相互之间不同的昵称、显示。
表设计im_user_relation_table 基于该单聊关系表还可以建立多元索引方便用户好友列表的获取同时支持加好友时间排序、昵称排序等功能。如果考虑到延时、费用等因素即时使用多元索引直接通过getRange接口也可以快速拉、高效的获取自己所有好友列表实现好友关系的维护与查询。
|功能人与人的关系
借助以上表人与人的关系可以很简单实现比如我判断我与小明的好友关系直接通过单行查询知道我们的好友关系是否存在如果存在就不会展示加好友按钮。而如果非好友这是完成好友添加后写入两行不同主键顺序行并生成一个唯一的timelineId即可。这个设计的好处在于用户可以直接通过自己的ID与好友的ID快速获取会话信息。只要用户在写入两行时做好一致性维护。
如果好友关系一旦解除可以直接拼出关系表中两行主键对用户关系通过做物理删除删除行或逻辑删除属性列状态修改结束两两个人的好友关系即可。
|核心代码 public void establishFriendship(String userA, String userB, String timelineId) {PrimaryKey primaryKeyA PrimaryKeyBuilder.createPrimaryKeyBuilder().addPrimaryKeyColumn(main_user, PrimaryKeyValue.fromString(userA)).addPrimaryKeyColumn(sub_user, PrimaryKeyValue.fromString(userB)).build();RowPutChange rowPutChangeA new RowPutChange(userRelationTable, primaryKeyA);rowPutChangeA.addColumn(timeline_id, ColumnValue.fromString(timelineId));PrimaryKey primaryKeyB PrimaryKeyBuilder.createPrimaryKeyBuilder().addPrimaryKeyColumn(main_user, PrimaryKeyValue.fromString(userB)).addPrimaryKeyColumn(sub_user, PrimaryKeyValue.fromString(userA)).build();RowPutChange rowPutChangeB new RowPutChange(userRelationTable, primaryKeyB);rowPutChangeB.addColumn(timeline_id, ColumnValue.fromString(timelineId));BatchWriteRowRequest request new BatchWriteRowRequest();request.addRowChange(rowPutChangeA);request.addRowChange(rowPutChangeB);syncClient.batchWriteRow(request);
}public void breakupFriendship(String userA, String userB) {PrimaryKey primaryKeyA PrimaryKeyBuilder.createPrimaryKeyBuilder().addPrimaryKeyColumn(main_user, PrimaryKeyValue.fromString(userA)).addPrimaryKeyColumn(sub_user, PrimaryKeyValue.fromString(userB)).build();RowDeleteChange rowPutChangeA new RowDeleteChange(userRelationTable, primaryKeyA);PrimaryKey primaryKeyB PrimaryKeyBuilder.createPrimaryKeyBuilder().addPrimaryKeyColumn(main_user, PrimaryKeyValue.fromString(userB)).addPrimaryKeyColumn(sub_user, PrimaryKeyValue.fromString(userA)).build();RowDeleteChange rowPutChangeB new RowDeleteChange(userRelationTable, primaryKeyB);BatchWriteRowRequest request new BatchWriteRowRequest();request.addRowChange(rowPutChangeA);request.addRowChange(rowPutChangeB);syncClient.batchWriteRow(request);
} 群聊关系
|功能群聊会话与人的关系
群聊时主要的查询需求还是获取当前群内用户的列表。一方面方便群属性的展示另一方面为应用做写扩散提供快速获取收件人列表的查询。因而在表设计上我们会建议用户使用两列主键第一列为群ID第二列为用户ID。通过这样的设计可以直接给予getRange接口拉取群所有用户的信息。
群聊关系表解决了群到用户的映射关系但我们还需要用户到群的映射关系。如果为了查询用户所在群的列表而新键一张表冗余成本、一致性维护成本就很高。这里可以使用两种索引来解决反向的映射关系。样例中我们使用了二级索引将用户ID字段作为索引主键从而可以直接基于索引查询单用户的群列表。同步实时性更好成本更低。
当然用户也可以使用多元索引对群、用户、入群时间做索引可以查询到某用户的所有在群列表并且基于入群时间排序。
表设计im_group_relation_table 基于群关系表可以直接基于关系主表通过getRange的方式获取单个群内所有的用户。在做写扩散时可以直接获取群内用户ID列表提升写扩散的效率。同时也方便展示群内用户列表。
|核心代码
public ListConversation listMySingleConversations(String userId) {PrimaryKey start PrimaryKeyBuilder.createPrimaryKeyBuilder().addPrimaryKeyColumn(main_user, PrimaryKeyValue.fromString(userId)).addPrimaryKeyColumn(sub_user, PrimaryKeyValue.INF_MIN).build();PrimaryKey end PrimaryKeyBuilder.createPrimaryKeyBuilder().addPrimaryKeyColumn(main_user, PrimaryKeyValue.fromString(userId)).addPrimaryKeyColumn(sub_user, PrimaryKeyValue.INF_MAX).build();RangeRowQueryCriteria criteria new RangeRowQueryCriteria(userRelationTable);criteria.setInclusiveStartPrimaryKey(start);criteria.setExclusiveEndPrimaryKey(end);criteria.setMaxVersions(1);criteria.setLimit(100);criteria.setDirection(Direction.FORWARD);criteria.addColumnsToGet(new String[] {timeline_id});GetRangeRequest request new GetRangeRequest(criteria);GetRangeResponse response syncClient.getRange(request);ListConversation singleConversations new ArrayListConversation(response.getRows().size());for (Row row : response.getRows()) {String timelineId row.getColumn(timeline_id).get(0).getValue().asString();String subUserId row.getPrimaryKey().getPrimaryKeyColumn(sub_user).getValue().asString();User friend describeUser(subUserId);Conversation conversation new Conversation(timelineId, friend);singleConversations.add(conversation);}return singleConversations;
} |功能人与群聊会话的关系
获取单用户所有加入群列表可以基于主表创建二级索引将用户字段设为索引的第一列主键。索引的数据结构见下图。这样基于二级索引可以直接通过getRange的方式获取单用户加入的群的TimlineId列表。
二级索引im_group_relation_global_index 核心代码
public ListConversation listMyGroupConversations(String userId) {PrimaryKey start PrimaryKeyBuilder.createPrimaryKeyBuilder().addPrimaryKeyColumn(user_id, PrimaryKeyValue.fromString(userId)).addPrimaryKeyColumn(group_id, PrimaryKeyValue.INF_MIN).build();PrimaryKey end PrimaryKeyBuilder.createPrimaryKeyBuilder().addPrimaryKeyColumn(user_id, PrimaryKeyValue.fromString(userId)).addPrimaryKeyColumn(group_id, PrimaryKeyValue.INF_MAX).build();RangeRowQueryCriteria criteria new RangeRowQueryCriteria(groupRelationGlobalIndex);criteria.setInclusiveStartPrimaryKey(start);criteria.setExclusiveEndPrimaryKey(end);criteria.setMaxVersions(1);criteria.setLimit(100);criteria.setDirection(Direction.FORWARD);criteria.addColumnsToGet(new String[] {group_id});GetRangeRequest request new GetRangeRequest(criteria);GetRangeResponse response syncClient.getRange(request);ListConversation groupConversations new ArrayListConversation(response.getRows().size());for (Row row : response.getRows()) {String timelineId row.getPrimaryKey().getPrimaryKeyColumn(group_id).getValue().asString();Group group describeGroup(timelineId);Conversation conversation new Conversation(timelineId, group);groupConversations.add(conversation);}return groupConversations;
} 即时感知
会话池方案
即时感知新消息正是IMInstant Message场景下核心所在。让客户端及时感知到新信息的到来然后客户端接收到通知后才会从同步库中拉取更新的消息让用户更快速、更及时地提醒用户阅读新消息。可是接受者如何才能快速感知到自己有了新消息呢
让在线的客户端周期性的刷新拉取这样的方式毫无疑问可以满足需求但伴随而来的是大量无效的网络资源浪费。同时应用的压力也会随着用户量的不断增长变得更沉重。而当白天大量非活跃用户在线时压力更为明显。面对这一问题应用通常会维护一个推送会话池。会话池记录了在线客户端与用户信息当在线用户有新的消息写入通过推送池获取该用户的会话然后通知客户端拉取同步库新消息。这样同步消息的压力只会随着真实消息量而增长避免了大量不必要的同步库查询请求。
实现会话推送池的方案很多可以使用内存型数据库也可以直接使用表格存储同时保证会话推送池的持久化。
在即时感知上最直观的就是会话表中变动的未读消息数统计了。统计新消息的实现方式上已在本文的【消息存储 第二类同步库 新消息即时统计】部分做了详尽描述不理解的可返回去重新看一下。持久化未读消息数是很必要的否则在更换设备或重新登录后。未读消息数被清零将会忽略很多新消息提醒这是我们不能接受的。
其他
多端同步
实现了以上功能IM系统的基本需求已经完成。但实现多端数据同步上还有两个注意事项。
其一我们对于单客户端情况下用户同步库做了一个checkpoint点的持久化对应的概念是“已读最新消息的sequenceId”。此时checkpoint点无客户端的区分如果使用本地做持久化多端同步时就会出现问题不同客户端统计的未读消息数就会不一致。这是需要通过应用服务端维护checkpoint点同时会话的未读消息数也需要在应用服务侧维护这样才能保证多端统计数一致。同时当有未读消息的会话被点击会话未读数清0时要让服务有感知然后通知到其他在线端维护实时一致性。
其二多端情况下自己在一个客户端发送了新消息其他客户端在没有其他新消息时是无法感知并刷新自己的发送消息这在多端同步中也是要解决的小问题。这时简单的解决方案就是将自己发送的消息也写入自己的同步库。只要再统计未读信息时对自己的信息不计数但在最新消息摘要中需要做更新。这样多端同步问题很容易实现。
添加好友、入群申请
添加好友或入群不是主动发起请求就会直接完成的这里需要主动方申请后审核方完成统一才会真实完成。因而只有在审核方才会有权限发起关系的创建。
那如何让被添加用户或群主感知到申请当然是借助同步库作为一种新的消息类型或者特殊的会话让用户即时感知到新申请尽早完成审批。申请列表如果需要持久化也可单独建表维护只要保证用户新申请的即时感知即可。
样例实操
本位为了与用户一起梳理IM系统应用的功能点基于Tablestore实现的样例简单功能完整的样例代码已完成开源。用户可以结合文章、代码一起阅读。代码在本地运行使用前请确保
开通服务、创建实例获取AK设置样例配置文件实例支持二级索引需要主动申请
开源地址
基于Tablestore实现的样例简单功能完整的样例代码已完成开源。开源地址https://github.com/aliyun/tablestore-examples/tree/master/demos/ImChart
样例配置
在home目录下创建tablestoreCong.json文件填写相应参数如下
# mac 或 linux系统下/home/userhome/tablestoreCong.json
# windows系统下: C:\Documents and Settings\%用户名%\tablestoreCong.json
{endpoint: http://instanceName.cn-hangzhou.ots.aliyuncs.com,accessId: ***********,accessKey: ***********************,instanceName: instanceName
}
endpoint实例的接入地址控制台实例详情页获取 accessIdAK的ID获取AK链接提供 accessKeyAK的密码获取AK链接提供 instanceName使用的实例名
样例入口
样例中共有三个入口用户需要根据先后顺序执行使用后及时释放资源避免不必要的费用浪费。 项目结构 原文链接 本文为云栖社区原创内容未经允许不得转载。
