Chinaunix首页 | 论坛 | 博客
  • 博客访问: 471265
  • 博文数量: 153
  • 博客积分: 0
  • 博客等级: 民兵
  • 技术积分: 1575
  • 用 户 组: 普通用户
  • 注册时间: 2016-12-20 17:02
文章分类

全部博文(153)

文章存档

2017年(111)

2016年(42)

我的朋友

分类: Java

2016-12-29 16:08:37

Worker模式

想解决的问题

异步执行一些任务,有返回或无返回结果

使用动机

有些时候想执行一些异步任务,如异步网络通信、daemon任务,但又不想去管理这任务的生命周。这个时候可以使用Worker模式,它会帮您管理与执行任务,并能非常方便地获取结果

结构

很多人可能为觉得这与executor很像,但executor是多线程的,它的作用更像是一个规划中心。而Worker则只是个搬运工,它自己本身只有一个线程的。每个worker有自己的任务处理逻辑,为了实现这个目的,有两种方式

1. 建立一个抽象的AbstractWorker,不同逻辑的worker对其进行不同的实现;

2. 对worker新增一个TaskProcessor不同的任务传入不同的processor即可。

第二种方式worker的角色可以很方便地改变,而且可以随时更换processor,可以理解成可”刷机”的worker
^ ^。这里我们使用第二种方式来介绍此模式的整体结构。


详细介绍一下几个角色:

  • ConfigurableWorker:顾名思义这个就是真正干活的worker了。要实现自我生命周期管理,需要实现Runable,这样其才能以单独的线程运行,需要注意的是work最好以daemon线程的方式运行。worker里面还包括几个其它成员:taskQueue,一个阻塞性质的queue,一般BlockingArrayList就可以了,这样任务是FIFO(先进先出)的,如果要考虑任务的优先级,则可以考虑使用PriorityBlockingQueue;listeners,根据事件进行划分的事件监听者,以便于当一个任务完成的时候进行处理,需要注意的是,为了较高效地进行listener遍历,这里我推荐使用CopyOnWriteArrayList,免得每次都复制。其对应的方法有addlistener、addTask等配套方法,这个都不多说了,更详细的可以看后面的示例代码。
  • WorkerTask:实际上这是一个抽象的工内容,其包括基本的id与,task的ID是Worker生成的,相当于递wtte后的一个执回,当数据执行完了的时候需要使用这个id来取结果。而后面真正实现的实体task则包含任务处理时需要的数据。
  • Processor:为了实现可”刷机”的worker,我们将处理逻辑与worker分开来,processor的本职工作很简单,只需要加工传入的task数据即可,加工完成后触发fireEvent(WorkerEvent.TASK_COMPLETE)事件,之后通过Future的get即可得到最终的数据。

另外再说一点,对于addTask,可以有一个overload的方法,即在输入task的同时,传入一个RejectPolice,这样可以在size过大的时候做出拒绝操作,有效避免被撑死。

适用性/问题

这种设计能自动处理任务,并能根据任务的优先级自动调节任务的执行顺序,一个完全独立的thread,你完全可以将其理解成一专门负责干某种活的”机器人”。它可以用于处理一些定时、请求量固定均匀且对实时性要求不是太高的任务,如日志记录,数据分析等。当然,如果想提高任务处理的数据,可以生成多个worker,就相当于雇佣更多的人来为你干活,非常直观的。当然这样一来,谁来维护这worker便成了一个问题,另外就目前这种设计下worker之间是没有通信与协同的,这些都是改进点。

那么对于多个worker,有什么组织方式呢?这里我介绍三种,算是抛砖引玉:

流水线式worker(assembly-line worker)

就像生产车间上的流水线工人一样,将任务切分成几个小块,每个worker负责自己的一部分,以提高整体的生产、产出效率,如下图:

假设完成任务 t 需要的时间为:W(t)=n,那么将任务分解成m份,流水线式的执行,每小份需要的时间便为 W(t/m)=n/m,那么执行1000条任务的时间,单个为1000n,流水线长度为L,则用这种方式所用的时间为(1000-1)*(m-L+1)*n/m+n
其中L

多级反馈队列(Multilevel Feedback Queue)

这是一个有Q1、Q2...Qn个多重流水线方式,从高到低分别代码不同的优先级,高优先级的worker要多于低优先级的,一般是2的倍数,即Q4有16个worker、Q3有8个,后面类推。任务根据预先估计好的优先级进入,如果任务在某步的执行过长,直接踢到下一级,让出最快的资源。


显然这种方式的好处就在于可以动态地调整任务的优级,及时做出反应。当然,为了实现更好的高度,我们可以在低级里增加一个阀值,使得放偶然放入低级的task可以有复活的机会^
^。

MapReduce式

流水线虽然有一定的并行性,但总体来说仍然是串行的,因为只要有一个节点出了问题,那都是致命的错误。MapReduce是Google率先实现的一个分布式算法,有非常好的并行执行效率。


只要我们将Map与Reduce都改成Worker就行了,如MapWorker与ReduceWorker。这样,可以看见,Map的过程是完全并行的,当然这样就需要在Map与Reduce上的分配与数据组合上稍稍下一点功夫了。


阅读全文请点击:
阅读(1548) | 评论(0) | 转发(0) |
给主人留下些什么吧!~~