随着业务的发展, 用户、资源和其他共享数据业务库散落在不同的库表, 甚至是不同的服务器和机房,为了做一些统计分析工作, 需要将这些数据汇集到搜索引擎或者hdfs等, 因此需要对散落在各处的元数据进行抽取和监听。 其中包括存量数据(历史数据)和增量数据(变更数据)。
- 存量数据多线程从MySQL或者MongoDB中读取, 以Task格式然后塞入队列中, 默认使用activeMQ;
- 增量数据通过监听数据变化的方式,将变更信息以Task格式塞入队列; 其中MySQL监听binlog的变更, MongoDB通过读取oplog的形式获取变更信息;
- 消费端通过轮询读取队列或者监听队列(默认)的方式获取Task格式的任务, 然后经过一系列filters的处理, 将数据放入存储引擎或者其他存储单元;
- 扩展性。各生产者和消费者只需要完成插件, 放到生产者容器或者消费者容器内即可。
- 灵活性。可以灵活的完成各个插件的加载和卸载; 消费者可以顺序进行一些列的处理。
- 一致性。使用hash算法将同一业务id的Task塞入同一队列, 增量数据单线程生产。
- 高吞吐。考虑生产者生产速度可能远远大于消费者消费速度, 因此消费者端多线程监听并处理任务。
- 高可用。使用zookeeper作为服务注册中心以及插件注册中心。
生产者和消费者均采用插件化处理方案, 各个生产者/消费者之间相互不受影响, 在单个生产者/消费者逻辑变更时可以不用重启整个服务,只需要将当前生产者/消费者 从zookeeper中移除该插件所在的节点, 然后替换为新插件注册到zookeeper即可。插件化的好处自然是高度解耦和不停服更新插件带来的稳定性。
除了插件化带来的动态加载或者卸载生产者/消费者外, 针对生产/消费过程中使用的各个中间件均是可以动态选择的, 只要按照接口定义完成相应的实现即可。另外, 针对存量数据和增量数据的特点, 我们同时也考虑了插件的生产顺序, 即插件的依赖关系。 只需要在各自插件中配置依赖插件名称即可。消费插件在消费中可以自定义各种处理逻辑。
对于高一致性的场景, 我们的做法是: 首先将原始数据进行落地处理(可以存储在关系型数据库中), 接着发给broker(假设queue使用的是mq), 只有正确的发给了broker且被消费者成功消费, 才会更新落地的记录。 为了确保数据一致性和较小的消耗, 生产增量数据时使用单线程处理, 确保入队顺序和用户行为完全一致。 另外, 在消费端, 首先检查落地记录, 该消息是否已经被消费, 如果被消费则放弃继续处理以防止重复消费; 接着检查该消息的前置消息(相同业务Id的前一条发生的消息)是否被正确消费, 如果没有被正确消费, 为了保证消费的顺序性, 则暂时不进行处理, 等待一段时间后继续查看前置消息是否被消费。 为了确保消息总是能正确消费, 有个定时任务作为生产者不断扫描未能正确消费的任务, 重新发给消费者消费。
生产者和消费者本身已经插件化处理, 也是无状态的服务, 本身可以确保高可用。生产者和消费者容器本身使用zookeeper作为注册中心, 当运行节点出现问题时, 重新选出节点作为运行节点。