项目背景
在推荐系统中消息队列是广泛使用的基础组件，但在我司的推荐系统中有许多成千上万个实例需要完整消费同一个topic的需求，而仅仅因为单机网络带宽的限制就部署大量的消息队列server显然是一种浪费，所以我司自研了一套客户端能够转发消息的消息队列服务BTQueue。
但BTQ在设计时考虑的比较简单，在后续使用中发现了许多问题，比如BTQ没有Sequence Number，只能依靠时间戳定位，导致基于它做数据同步的存储模块无法精准地控制数据一致，而且BTQ server端没有使用一致性协议，本身就有一致性风险，还存在稳定性、内存消耗等其它方面的问题，因此现在需要结合推荐架构中的典型数据分发和存储场景，设计一套完善的消息队列组件，目标能够简化推荐数据流-存储架构，降低资源开销。
衍生Topic
除了BTQ的数据转发功能外，新的消息队列还将支持通过轻量算子衍生新的topic的能力，衍生出的topic不需要多副本存储，由上游topic消息经过算子计算直接获得，此特性能够简化服务的部署，方便各模块间解耦，避免重复计算，解决传统中间模块消费-转发过程中转发失败难以处理的问题。
架构设计
每个集群会提前分好固定数目的partition，每个partition有独立的Sequence Number，整个集群的元数据存在一个元数据服务上（第一版用zookeeper），除了源头topic，每个衍生topic配置固定的上游topic。
每个topic的所有consumer，会互相转发消息，每个partition的转发的拓扑是一个多叉树，由一个独立的调度模块，统筹全局的转发拓扑。
除了独立的调度模块，整个架构的主体都由基于以下框架开发的实例组成
源头topic可以支持Kafka、一致性协议等多种数据源，一个源头topic会启动多个实例，调度模块会将partition均衡调度到各个实例上，并且支持多个备份容灾。
下游实例的输入端设置为OriginQueue Consumer，并指定好topic，输出端默认会输出一个与输入相同的topic，用于转发流量。使用者可以自行实现多个算子，产出衍生topic，也可以实现本地回调，作为末端消费。
流量转发
调度策略会让每个partition由尽量少的节点转发，并且转发负载尽量均衡，节点不会缓存不负责转发的partition的数据，这相比BTQ能够大幅降低内存开销。
现在BTQ遇到过多Consumer由于自身原因产生lag时，会产生大量回源流量，把源头资源耗尽，这个问题无法从根源上避免，但可以本系统可以从两点改善：
1.由于每个节点只转发缓存少量partition，故而整体上可以缓存更多的数据，短期的lag可以靠缓存覆盖，减少回源流量
2.每个topic要限制回源并发量，优先保证健康的consumer的正常消费，将有lag的消费按照Sequence Number、消费速率等因素分到不同队列中，每个队列从最小Sequence Number从上游拉取数据，只产生一份回源流量
应用举例-三级存储架构
推荐系统中常见的存储分为分布式持久化存储、分布式缓存、本地缓存三种，分布式持久化存储保存全量数据，分布式缓存保存热点数据，本地缓存保存极热数据，但市面上没有一套分布式存储系统将这三种角色完美整合。
以我司为例，分布式缓存一般使用redis或者memory cache，分布式缓存和持久化存储是完全无关的系统，需要使用方自己做数据同步；本地缓存现在在我们的推荐架构里两个典型场景是预估服务读取embedding server和推荐服务读分布式正排索引。预估服务会定期淘汰老的embedding数据，随机抽取一部分访问更新本地缓存，这种方式的问题是数据更新会有比较大的延迟，并且淘汰、请求等操作有一定性能开销。推荐服务正排索引的本地缓存，每个实例会消费整个BTQ，然后本地做更新和淘汰，这会消费大量非热点数据的消息。
但基于OriginQueue的两点新特性，可以设计一套统一的分布式存储系统，各存储实例之间通过消息队列做数据同步，而不必使用方做跨系统的数据同步，数据一致性大幅改善。由于数据源头是有一致性协议保证的消息队列，所以持久化的存储只要记录好Sequence Number，就可以不必使用WAL，从而节约一部分写io。
大部分场景下，每个调用实例的热点数据是基本一致的，所以可以由存储服务搜集统计信息，筛选出热点数据，然后产出一份只有热点数据相关operate的操作日志，请求方只需要订阅这部分操作，更新本地缓存即可。这样本地缓存即能保证实时更新，又不会为非热点数据浪费资源。
上图中，数据源是一份按照ItemID分好partition的kafka，一组OriginQueue服务把它作为输入端，通过两个定制的算子，衍生出两个topic，一个算子筛选出视频相关的用户行为，转换成通用的计数的KV Operate，另一个算子筛选出直播相关的用户行为，转换成通用的计数的KV Operate，这样存储实例只需要消费衍生的topic就可以累积数据，并且各实例之间完全按照Sequence Number对齐。全量的KV operate经过热点过滤模块过滤后，衍生出新的topic给reader的本地缓存消费，依靠数据转发的功能，能够支持上万个reader同时实时更新本地缓存。
元数据服务
zookeeper是一个广泛使用并且大家普遍对它不满的服务，主要问题有：性能差、接口丑、功能少。
由于zk每个集群全局只有一个有序队列，所以写入相当于是单线程的，但实际使用中，一个zk集群往往会存储多块互不相干的数据，这些数据共用一个队列显然是不合理的。zk的目录结构可以看成是一棵树，现实中许多zk集群，是可以把这颗树划分成互不相干的若干棵子树的，让这些子树各自拥有独立的操作队列，这些子树的祖先节点，不允许有值，只能创建和删除。
现在zk的每个watch，是注册到每个节点上的，如果需要watch的位置比较多，实现会比较复杂而且效率不高。其实客户端也可以作为一个observer，绑定一个子树，同步操作序列，在本地缓存这个子树，这样每个位置的回调只需要在本地设置，server端不需要保存一大堆watch信息了，而且使用observer模式就可以避免丢消息的问题。而基于OriginQueue封装的client，由于自带流量转发功能，上万client同时订阅大量元数据也不会有性能问题。
