Skip to content

Otter调度模型

agapple edited this page Aug 19, 2013 · 2 revisions

背景

在介绍调度模型之前,首先了解一下otter系统要解决的异地机房的网络环境.

  1. 中美网络延迟 (平均200ms)
  2. 中美传输速度 (2~6MB/s)

网络因素是一个很重要的问题,

    a. 比如中美网络延迟RTT,平均为200ms,这直接会影响整个系统的架构设计。

试想一下,发送一条binlog一次RTT 200ms,那是否意味着单线程1秒钟只能发送5条。不过tcp在解决这类问题时,有自己的一套优化算法,叫做滑动窗口,它发送数据可能一次性发了10条,然后一起等返回结果,这样可以提升传输效率,但始终无法满足1秒传输1w+记录的需求。这也就决定了,需要对发送的binlog做批处理,一次性发送尽可能多的数据,然后一起等结果.

    b . 比如中美单socket的带宽只有2~6MB,这直接会影响整个系统的架构设计。

试想一下,假定一binlog平均1kb,那6MB最多只有6000条数据,也就意味着最大的同步tps只有6000? 而且很多业务的mysql都是共享,也就是6000个binlog对象中,可能只有三分之一或者四分之一是某个业务的,这肯定是无法满足要求的。所以,基于带宽的问题,决定otter架构必须是双节点部署,在杭州一个节点,美国一个节点,杭州这边对数据做加速同步处理,然后快速传递到美国.

基于这两个因素,决定了: batch处理 + 双节点部署的架构.

调度模型

在正式介绍otter调度模型之前,我们首先得了解TCP/IP协议在解决此类"差网络"环境的一些处理方案,从中借鉴相应的方案.

Nagle算法

通过Canal解决Nagle算法,Canal之前是做为otter的一个子项目,为解决otter的数据增量获取的机制,并为otter项目的特点而量身打造了几个feature.

Canal的处理:

   a. 构建RingBuffer (可以基于内存控制模式/数量控制模式)

   b. 允许客户端指定batchSize获取

       i. 内存大小

       ii. 记录数

   c. 指定定batchSize + timeout获取

       i. timeout = -1 ,即时获取,有多少取多少

       ii. timeout = 0,阻塞至满足batchSize条件

       iii. timeout > 0,阻塞指定的时间或者满足batchSize.

建议值:batchSize=4000(约4M) , timeout=500,内存控制模式

滑动窗口


说明:

  1. otter通过select模块串行获取canal的批数据,注意是串行获取,每批次获取到的数据,就会有一个全局标识,otter里称之为processId.
  2. select模块获取到数据后,将其传递给后续的ETL模型. 这里E和T模块会是一个并行处理
  3. 将数据最后传递到Load时,会根据每批数据对应的processId,按照顺序进行串行加载。 ( 比如有一个processId=2的数据先到了Load模块,但会阻塞等processId=1的数据Load完成后才会被执行)

简单一点说,Select/Load模块会是一个串行机制来保证binlog处理的顺序性,Extract/Transform会是一个并行,加速传输效率.

并行度

类似于tcp滑动窗口大小,比如整个滑动窗口设置了并行度为5时,只有等第一个processId Load完成后,第6个Select才会去获取数据。

数据可靠性

  • 如何保证数据不丢:2pc. (get/ack)
  • 如何处理重传协议:get/ack/rollback
  • 如何支持并行化:多get cursor+ack curosr (可以参看Canal的异步ACK模型)


编程模型抽象(SEDA模型)

说明: 将并行化调度的串行/并行处理,进行隐藏,抽象了await/single的接口,整个调度称之为仲裁器。(有了这层抽象,不同的仲裁器实现可以解决同机房,异地机房的同步需求)

模型接口:

  • await模拟object获取锁操作
  • notify被唤醒后提交任务到thread pools
  • single模拟object释放锁操作,触发下一个stage

这里使用了SEDA模型的优势:

  • 共享thread pool,解决流控机制
  • 划分多stage,提升资源利用率
  • 统一编程模型,支持同机房,跨机房不同的调度算法

仲裁器算法

主要包括: 令牌生成(processId) + 事件通知.

令牌生成:

  • 基于AtomicLong.inc()机制,(纯内存机制,解决同机房,单节点同步需求,不需要多节点交互)
  • 基于zookeeper的自增id机制,(解决异地机房,多节点协作同步需求)

事件通知: (简单原理: 每个stage都会有个block queue,接收上一个stage的single信号通知,当前stage会阻塞在该block queue上,直到有信号通知)

  • block queue + put/take方法,(纯内存机制)
  • block queue + rpc + put/take方法 (两个stage对应的node不同,需要rpc调用,需要依赖负载均衡算法解决node节点的选择问题)
  • block queue + zookeeper watcher ()

负载均衡算法:

  • Stick : 类似于session stick技术,一旦第一次选择了node,下一次选择会继续使用该node. (有一个好处,资源上下文缓存命中率高)
  • Random : 随机算法
  • RoundRbin : 轮询算法

注意点:每个node节点,都会在zookeeper中生成Ephemeral节点,每个node都会缓存住当前存活的node列表,node节点消失,通过zookeeper watcher机制刷新每个node机器的内存。然后针对每次负载均衡选择时只针对当前存活的节点,保证调度的可靠性。

调度算法成本估算

中美网络RTT = 200ms , zookeeper一次写入=10ms

调度成本估算:

    a. zookeeper + zookeeper watch (完全分布式)

       10 * 4 + 200 * 2 + 200 = 640ms

    b. zookeeper + rpc (sticky分布式,尽可能选择同节点)

    10 + 100 + 200 = 310ms

   c. memory + memory (内存调度,单机房)

       0ms

   d. memory + rpc (跨机房调度,最优实现,待完成??)

       0 + 100 + 100 = 200ms

数据传输

有了一层SEDA调度模型的抽象,S/E/T/L模块之间互不感知,那几个模块之间的数据传递,需要有一个机制来处理,这里抽象了一个pipe(管道)的概念.

原理:

stage | pipe | stage

基于pipe实现:

  • in memory (两个stage经过仲裁器调度算法选择为同一个node时,直接使用内存传输)
  • rpc call (<1MB)
  • file(gzip) + http多线程下载

在pipe中,通过对数据进行TTL控制,解决TCP协议中的丢包问题控制.