钉钉源码php技巧_若何造一个钉钉谈谈消息系统架构的实现

阿里妹导读：类场景是表格存储（Tablestore）主推的方向之一，因其数据存储构造在类数据存储上具有天然上风。
为了方便用户基于Tablestore为类场景建模，Tablestore封装Timeline模型，旨在让用户更快捷的实现类场景需求。
在推出Timeline（v1、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 List<AppMessage> 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); Iterator<TimelineEntry> iterator = store.createTimelineQueue(identifier).scan(parameter); List<AppMessage> appMessages = new LinkedList<AppMessage>(); while (iterator.hasNext() && counter++ <= 30) { TimelineEntry timelineEntry = iterator.next(); AppMessage appMessage = new AppMessage(timelineId, timelineEntry); appMessages.add(appMessage); } return appMessages; }

存储库的须要永久保存，是全体运用的全量存储。
存储库数据过期韶光（TTL）须要设为-1。

|功能：多维组合、全文检索

全文检索能力便是对存储库的内容做模糊查询，因而须要对存储库的数据建立多元索引。
详细索引字段，须要根据设计需求设计。
如钉钉公开群的检索，须要对群ID、发送人、类型、内容、以及时间建立索引，个中内容须要利用分词字符串类型，从而供应模糊查询的能力。

|核心代码

public List<AppMessage> 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); Iterator<TimelineEntry> iterator = store.createTimelineQueue(identifier).scan(parameter); List<AppMessage> appMessages = new LinkedList<AppMessage>(); 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 List<AppMessage> 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); Iterator<TimelineEntry> iterator = sync.createTimelineQueue(identifier).scan(parameter); List<AppMessage> appMessages = new LinkedList<AppMessage>(); 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识别特定会话，属性信息会包含：会话种别（群、单聊、公众号等）、群名称、公告、创建韶光等。
同时，通过群名称模糊查找群，也会是会话元数须要的主要能力。

在Timeline模型中，供应了Timeline Meta的管理能力，只需通过相应的接口便可实现会话meta的管理。

存储库中管理的是会话的行列步队（TimelineQueue），这里与会话元数据中的行逐一对应。
客户端用户选中特定会话后，运用从相应的行列步队倒序批量拉取消息展示到客户端，群聊单聊的利用办法一样，因而并不做会话类型的区分。

|功能：群检索

用户如果有加入群的需求，首先须要查询到特定的群。
查询群的办法与用户查询办法类似，功能也可以做相同的实现。
用户可以根据自己需求定制不同的索引字段设置，需求实现办法如下：

注：会话元数据可以直接掩护单聊会话与人的映射关系。
对付单聊的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 List<Conversation> 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); List<Conversation> singleConversations = new ArrayList<Conversation>(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 List<Conversation> 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); List<Conversation> groupConversations = new ArrayList<Conversation>(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;}

会话池方案

即时感知新正是IM（Instant Message）场景下核心所在。
让客户端及时感知到新信息的到来，然后客户端吸收到关照后才会从同步库中拉取更新的，让用户更快速、更及时地提醒用户阅读新。
可是，接管者如何才能快速感知到自己有了新呢？

让在线的客户端周期性的刷新拉取？这样的办法毫无疑问可以知足需求，但伴随而来的是大量无效的网络资源摧残浪费蹂躏。
同时运用的压力也会随着用户量的不断增长变得更沉重。
而当白天大量非生动用户在线时，压力更为明显。
面对这一问题，运用常日会掩护一个推送会话池。
会话池记录了在线客户端与用户信息，当在线用户有新的写入，通过推送池获取该用户的会话，然后关照客户端拉取同步库新。
这样同步的压力只会随着真实量而增长，避免了大量不必要的同步库查询要求。

实现会话推送池的方案很多，可以利用内存型数据库，也可以直策应用表格存储，同时担保会话推送池的持久化。

在即时感知上，最直不雅观的便是会话表中变动的未读数统计了。
统计新的实现办法上，已在本文的【存储 > 第二类：同步库 > 新即时统计】部分做了详尽描述，不理解的可返回去重新看一下。
持久化未读数是很必要的，否则在改换设备或重新登录后。
未读数被清零，将会忽略很多新提醒，这是我们不能接管的。

多端同步

实现了以上功能，IM系统的基本需求已经完成。
但实现多端数据同步上，还有两个把稳事变。

其一，我们对付单客户端情形下，用户同步库做了一个checkpoint点的持久化，对应的观点是：“已读最新的sequenceId”。
此时，checkpoint点无客户真个区分，如果利用本地做持久化，多端同步时就会涌现问题，不同客户端统计的未读数就会不一致。
这是须要通过运用做事端掩护checkpoint点，同时会话的未读数也须要在运用做事侧掩护，这样才能担保多端统计数同等。
同时，当有未读的会话被点击，会话未读数清0时，要让做事有感知，然后关照到其他在线端，掩护实时同等性。

其二，多端情形下，自己在一个客户端发送了新，其他客户端在没有其他新时，是无法感知并刷新自己的发送，这在多端同步中也是要办理的小问题。
这时，大略的办理方案便是将自己发送的，也写入自己的同步库。
只要再统计未读信息时，对自己的信息不计数，但在最新择要中须要做更新。
这样，多端同步问题很随意马虎实现。

添加好友、入群申请

添加好友或入群，不是主动发起要求就会直接完成的，这里须要主动方申请后，审核方完成统一才会真实完成。
因而只有在审核方才会有权限发起关系的创建。

那如何让被添加用户或群主感知到申请？当然是借助同步库，作为一种新的类型或者分外的会话，让用户即时感知到新申请，尽早完成审批。
申请列表如果须要持久化，也可单独建表掩护，只要担保用户新申请的即时感知即可。

本位为了与用户一起梳理IM系统运用的功能点，基于Tablestore实现的样例大略功能，完全的样例代码已完成开源。
用户可以结合文章、代码一起阅读。
代码在本地运行，利用前请确保：

开源地址

基于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：实例的接入地址，掌握台实例详情页获取；

accessId：AK的ID，获取AK链接供应；

accessKey：AK的密码，获取AK链接供应；

instanceName：利用的实例名；

样例入口

样例中共有三个入口，用户须要根据先后顺序实行，利用后及时开释资源，避免不必要的用度摧残浪费蹂躏。

项目构造

作者：潭潭