Video在线视频

我们现在来介绍流计算的

一种简单的实现

我们把它叫做

WORKER+QUEUE

这样的一种实现模式

worker很容易理解

那就是流计算中一种处理单元

它可以负责完成一些

比如说

从一个tweets中

那么抽出其中的URL这样的操作

或者说把一个已经知道

存在一个URL

我要把它在数据库里+1

这样的操作

那么这是worker

它很常见的这样一个功能

那么另外一个组件叫做queue

这是一个非常重要的组件

那么它主要具有两个功能

第一个叫做缓冲

第二个叫做路由

那我们来看一下这样一个事情

那么假如有很多的tweets

用户发了一个tweets

然后这个发布

首先它需要在网站上去呈现

它需要去更改这个

tweets本身的数据库

其次

它要接入到这个分析系统中

那么在这样的一个过程里面

实际上就会存在着

一个很重要的问题

就是速度的不匹配问题

也就是说

这边tweets产生的速度

和你这边能够处理的速度

分析系统处理的速度

这两个之间

并不一定能够在任何时候

都是匹配的

当然我们从排队论的理论

可以知道

如果这边的速度λ1

和这边的速度λ2

处理速度来讲

如果λ1平均速度

如果大于λ2的话

那么这个系统就肯定是要

丢掉很多的这个处理的

但即使在λ1小于等于λ2的

情况底下

仍然会存在着某些时刻

那么这个λ1存在着某些消息

它如果直接输入到

这个分析系统里面的话

分析系统会无法及时处理

主要是由于这个任务

到达的不均匀性造成的

那么怎么样解决这样一个问题

Queue就是用来解决这个问题的

这样一个方法

那么本质上 它是一些缓冲

也就是说

我们在这个生产者

我们把这个东西叫做生产者

producer

和这个分析系统

我们可以把这个东西

叫做消费者 consumer

在这个之间

我们设了一系列的缓冲

那么producer

它直接把这个东西

把它产生这个tweets

把它扔到这个Queue里面

而这个consumer

它可以从这个queue里面

它完成了当前的任务以后

它就把数据取出来进行处理

那么这样的一种方式

就是可以很好地匹配

当这个处理的平均速率

还是要比任务的产生的

这个平均速率高的时候

也就是在这种情况下

通过这样的缓冲

我们可以达到系统的一个匹配

那么这个是缓冲

那么另外还有一个问题是说

这个tweets

它们不见得都要进入到

同一个分析系统中

那可能还有另外一个

这是consumer1

可能还有consumer2

也就是说这边producer

它会产生很多的消息

那么这些消息

有的可能需要给consumer1

有的需要给consumer2

所以 队列也不见得只有一个

可能会有多个

那么具体说哪一个消息

应该到consumer1还是consumer2

这个就是我们所谓的路由

那么这样的一个

也就是说对于某一条tweets

它到底应该

送到哪一个队列里面

被哪个consumer所接受

那么这也是queue队列

它起到的一个重要的作用

那么这个 这副图

实际上就是

一个简单的一个示意图

也就是说我们怎么样

这个twitter（3：50）

是怎么样把每一条tweets

变成我们刚才看到的这个结果

也就是说在过去7天中

有多少tweets

含有某一个特定的URL

那么这样的过程是怎么样的

我们可以看到这个数据源

实际上就是我刚才多次用过的

这样的一个tweets1tweets2等等

它实际上就是一些短文本的文字

那么这个里面可能会含有

一些文字

也可能会含有一些URL1

可能会含有URL2

也可能没有

然后这个里面

可能URL1不只出现一次

是这样的

那么首先这一层的队列的意义

大家应该从我刚才介绍这个

队列的时候已经可以理解了

那它主要是说

tweets在产生了这个

某些人

这个人在发了这个tweets以后

这些tweets要被

我们这个整个的分析系统处理

要被后面的这个分析系统

实际上就是我们刚才讲的这个

对于这个消息来说

它是一个consumer 对吧

这个是producer

首先它把它放到了

某干个队列里面

那么在这一阶段

在我们这个特定的任务里面

这个tweets进入到不同的队列

并没有一个特殊的要求说

这个路由在这一层是

首先没有很重要的

它主要是

它可以进入到这个队列

也可以进入到这个队列

主要是起到一个负载平衡的作用

那么在中间这一层

这个这些worker

它们做的是一个什么样的处理

它们实际上是把tweets中

所包含的这个URL把它抽取出来

也就是说

比如说这条消息在这个队列里面

然后被这个worker所访问到

那么它会抽取说

这个tweets包含的URL1

那么类似的

比如这个tweets说

它会包含一个URL2

某一条消息进来

以此类推

那么在这些

当产生这个URL1以后

这个worker就要把它

放到一个对应的一个

队列里面来

这个放的时候

我们可以看到这个之间

是一个全连接的关系

就是说每一个worker都可以

连到每一个后续的这个

我们把它叫做URLQ

都是可以连到这样的

一个URLQ上面

那么这个queue

它后面又去连接了一个worker

这个worker做的是什么

它们会把数据写到

一个数据库里面

在twitter呢

它用到的是这个cossadra

这样的一个数据库

也就是说这个里面它会

这里面其实已经存了

比如说URL1它已经目前有5次

对吧

那么当我们把这个URL1的

这样的一个消息

把它放到这个队列里

然后被这个worker所获得以后

它会做一个什么操作

它会把这个库里面的URL1

对应的这个5次把它加1

来更新成6次

是这样一个操作

那么这里面需要

特别注意的一点是说

在这些worker产生的URL里面

它的URL只能对应于

如果比如说它产生了一个

它从这个消息中抽出了URL1

那个这个时候

这个URL1这个消息

它并不能够随意地对应到

下面这三个队列中的任何一个

而是只能对应其中的某一个

比如说它只能对应这一个

为什么是这样

我们可以看到后面这个关系

就是这个URL1

它实际上

就这个队列

它实际上对应的就是这个worker

那么我们希望

我们为什么希望做到

这个URL1只能对应到某一个队列

实际上是说

我们希望这个URL1只能

对应于这样一个worker

而只有这个worker会去

写数据库中对应于URL1的这一项

会去更新这样一项

那么这样做有什么样的好处

大家可以想一下

如果说这个worker

可以去更新这个URL1这一项

那么不如果URL1

也可以分配给这个queue

也可以分配给这个queue

和这个worker

那么它也可以去写这一项

那实际上这就会遇到一个

我们在分析系统中

非常常见的一个现象

就是一个原子行为一致性的问题

这时候我们就必须要对这一个

这一项上锁

因为有多个

可能有多个worker对它做更新

那么在worker1

对它做更新的时候

需要对它上锁

防止在同样的这个时候

worker2对它也做更新

这个时候就会产生这种不一致性

所以为了避免在数据库的这个

写中加锁

那么我们需要做到的是

只让某一个worker

能够去更新这个URL1

所对应的这个数据库项

那这个时候反过来就要求说

我们这个数据

这个URL只能从这个queue

特定的一个queue中

传给这个worker

那么反过来就是说

对于中间的这个分析消息的

这个worker

它产生了URL1这样的

一个消息以后

那它就只能给定给

固定的一个queue

好的 那实际上

这样的一种方式

实际上说明了一个什么问题

也就是说 从这个中间的worker

到后面这个queue

之间的这样一个映射

实际上 不是随意的映射

虽然连接是全连接的

也就是说

它可能可以映射到任何一个上面

但是对于一个给定的URL

它必须要给定一个确定的映射

这种确定性呢

实际上在实现中可以

用一个这个Hash的方式

来做这个东西对吧

我们可以通过对不对

URL来做一个Hash来做这件事

但是这样做了以后

我们实际长就可以看到

这实际上会造成一个

很严重的问题

也就是说我们讲过扩展性

假如说我们现在的消息

到达速率提高了

那这个时候我们希望能够增加

这要增加worker

这要怎么queue这也要增加worker

但是我们原来是以一个

Hash来做这个

这两部分之间的对应的

那么如果我们增加了

worker以后

这个Hash的方式

可能就很难去适应

也就是说这个扩展性会成为问题

那么另外类似的

如果说我们其中的

某一个queue

或者说某一个worker坏掉了

由于我们现在的映射是

一个固定的映射

也就是说谁个URL1

只能经过这个Q到这个worker

那么也就是说使得容错机制

也会变得相当的困难

所以 我们刚才介绍的这种

worker+queue的方式

实际上它会有若干个基本的问题

我们前面讲过说

我们评价一个系统

会从实时性 扩展性

容错能力和编程模型

这三个角度来看

那么从实时性和扩展性来讲

由于在这个worker和queue之间

这种使用Hash的这种

固定的对应关系是

使得这个系统不容易扩展

第二方面

也是同样的这样一个原因

系统也非常难容错

那么第三

由于你的编码中大量地使用了

这个queue这种中间件

还有这个worker

那么 而且它们之间的对应关系

又相当的复杂

使得这个编码也相当的复杂

刚才个讲过的

你想扩展或者想容错都会引入

非常多的这个编码的开销

所以这样一种方式

并不是一种很好的

处理流处理的方式

这种方式是Twitter公司

他们最早进行流处理的一个尝试

所以后来他们就开发了

这个Storm的这样一个系统

来解决刚才提到的这个问题

那么下面 我们也会介绍两个

专门用来做流计算的系统

来看这些流计算系统

是怎么比worker和queue这样的

直观的实现方式能够

更好的去实现流计算的

这样的扩展性 容错能力和编码

方便了这样的功能