近期这段时间在设计和实现日志系统。在整个日志系统系统中Zookeeper的作用非常重要——它用于协调各个分布式组件并提供必要的配置信息和元数据。这篇文章主要分享一下Zookeeper的使用场景。
这里主要涉及到Zookeeper在日志系统中的使用，但事实上它在我们的消息总线和搜索模块中也相同非常重要。
日志元数据
日志的类型和日志的字段这里我们统称为日志的元数据。我们构建日志系统的目的终于主要是为了：日志搜索，日志分析。这两大块我们非常大程度上依赖于——ElasticSearch（关于什么是ElasticSearch，这里就不多做介绍了）。
关于日志字段
日志的字段定义，在ElasticSearch中是一个索引中某个mapping的Schema。ElasticSearch是一个号称Schema Free的分布式全文检索系统。这里须要正确地理解Schema Free，它并非不须要Schema。也不是强制要求你必须有明白的Schema，有没有都能够做全文检索。可是像聚合、分析等非常多高级功能都建立在明白的Schema的基础上。因此。从分析统计的角度看，我们对日志进行明白的字段定义是非常有必要的。
日志与搜索模块的协同
日志元数据是日志系统的基础信息。我们在web管控台管理它并同步至Zookeeper供其它模块使用。比方搜索模块。由于上文提到日志的类型、字段事实上跟ElasticSearch的Mapping Type是对等的映射关系。所以搜索模块会重度依赖日志元数据。
另外。为了保证新的日志元数据（一般是一个新的日志类型被创建）尽快同步至ElasticSearch。我们利用了Zookeeper的事件Push机制（或者叫Pub/Sub机制）来实时获知日志元数据的变化。一旦有新的日志元数据产生。我们的搜索模块会马上得到事件通知，它会获取最新的日志元数据。然后为其在ElasticSearch的indices中新建一个mapping。为这样的日志类型的日志存入ElasticSearch做好准备。
这样的方式带来了哪些优点。眼下来看至少有三点：
- 实时性：搜索模块能第一时间感知到新的日志类型的创建
- 低耦合性：管控台上日志模块跟搜索模块，没有由于信息的依赖而产生较强的耦合性；它们通过Zookeeper进行了解耦
- Mapping Type的可控性：ElasticSearch有个非常好的特性，就是当你将一个文档存入某个mapping type，假设该文档中存在mapping未曾定义的字段，ElasticSearch将会为你自己主动加入该字段的定义。我们觉得这样的机制将会使日志字段变得不可控。因此我们通过统一日志元数据再加上后面基于相同的解析行为来保证Schema的可控性。
日志採集
从之前的文章中，你应该能够找到日志採集器的选型，出于多种原因（跟消息总线的集成、可定制性、支持从zookeeper获取配置等），我们选择了Flume-ng。它在最新的稳定版（1.6.0）提供了从zookeeper获取採集配置的功能。
于是，日志採集花费在运维上的成本就大大减少了。由于有了配置之后，运维人员就不须要手动在目标server上改动配置文件，完毕各种配置，仅仅须要固定键入例如以下的启动命令就可以：
sudo bin/flume-ng agent --conf conf -z 127.0.0.1:2181 -p /component/log/mysqlSlowquery/flume -name mysql-slowquery-30 -Dflume.root.logger=DEBUG,console而上面这个命令中，唯一须要变动的仅仅有以下几个部分：
- zookeeper的server（集群）信息
- 即将收集的日志的类型的flume路径
- 即将收集的日志的flume配置的znode名称，如上例是
mysql-slowquery-30
事实上原先须要手动改动配置文件的部分參数项将在提供的管控台中进行配置，但基于web的表单填写显然要比在server上以命令行的方式来得easy得多。
这里我们的做法是拆解了flume的配置文件，将其固定不变的部分做成模板，将其可变部分做成表单。在提交之前，通过模板引擎将模板跟配置信息进行合并为完整的配置并推送到Zookeeper中去。
当中须要配置的部分參数有：
- 日志文件所在目标server的路径位置
- 日志文件名的格式
- 日志是单行模式还是多行模式
- 消息源的secret(消息总线部分)
- 消息槽的名称（消息总线部分）
- 消息流的token（消息总线部分）
- ….
日志解析
相同在之前的文章中我也提及我们在日志解析上的选择是morphline。
morphline是个在Hadoop生态系统中的ETL Framework。morphline也有一个配置文件用于定义一系列的Commands。而它也有固定部分和可变部分，由于解析主要应用了morphline的Grok命令，所以针对这个命令，可变部分主要是：
- 解析字典
- 解析的正則表達式
我们的做法相同相似于日志採集模块。将morphline的配置文件的固定部分做成固定模板。然后将可变部分在管控台上进行配置。终于合并提交到Zookeeper中。
日志解析服务（归属于以下的后台服务）。在启动时会依据自己的日志类型，从Zookeeper的特定节点下找到该日志类型的morphline的配置，将其获取下来并保存在本地文件系统中，然后构建Mrophline对象（由于morphline眼下仅仅提供基于File对象的构造方式。所以多了一个先保存至本地文件再基于文件构造Morphline对象的步骤）进行解析。
后台服务
日志解析这边仅仅是给解析任务提供了 元数据 。真正的解析由后台的解析任务来完毕，我们将相似的这些同意在后台的全部任务笼统得归结为 后台服务 。
后台服务遵循：任务组->任务->工作线程的层次性的组织方式。依照服务的业务类型（说白了就是同一套处理逻辑）。将其划分为不同的任务组（比方：ETL组、日志索引组等）。不同的任务组下会有 至少 一个任务。比方ETL任务组下就会有非常多个任务（比方：nginx訪问日志解析任务、mysql慢查询日志解析任务等）。而某一个任务下又存在至少一个工作线程。
任务元数据
在管控台有一个后台服务模块。专门用于管理任务对象以及它们的元数据。
运行任务的工作者线程本身是无状态的。它们在启动的时候会去Zookeeper上下载它们运行任务所须要的元数据。
热备机制
通常，为了服务的可用性我们会为每一个任务配备不少于一个工作者线程。
当然，这里的 不少于一个 并不仅仅是基于一个运行后台服务的JVM进程或一个主机节点来计数的，而是针对由多个节点组成的集群而言。
这通过一个配置来实现：
worker.minimumNumPerTask=2它的意义是：对每一个task而言，启动的最少的worker线程数。假设一个主机节点上启动两个后台服务的JVM进程，那么这个task就会相应4个工作者线程。
正常情况下，每一个任务在同一时刻仅仅有一个处于active状态的工作者线程，而其它抢占失败的都会将自己切换为standby模式。作为备援随时待命。
这个机制是怎样实现的？这得益于Zookeeper提供的 暂时顺序 节点。
当多个工作者线程去竞争一个任务的时候，它们首先去该任务的path下创建一个子path，并注冊自己的主机等信息。
注意这里创建的子path的类型不同于其它Zookeeper使用场景的path类型（其它path通常都是持久型的），它是暂时、顺序的。这两个属性非常重要：
- 暂时： 当一个工作者线程挂掉之后。它本地的Zookeeper会话也会随之失效。在其会话失效之后。暂时节点将会消失。
- 顺序：它能仲裁出创建path的client的先后顺序，并在新建的path中追加标识
各个工作者线程创建暂时顺序的path后，由于具有 顺序 性。Zookeeper会依照它们创建的顺序在path后追加带有从1開始递增的编号。各个工作者创建完毕后会得到各自的编号，然后它们作一个顺序推断，谁是最小的谁就会获得任务的运行机会并成为active工作者，而其它抢占失败的将默认切换到standby模式，这些抢占失败的工作者线程会注冊成为它们抢占task的 子节点变更 watcher。这时 暂时 属性就派上用场了，当处于active模式的工作者线程丢失会话之后，这些standby将会收到通知，而这时它们会再次去推断自己的编号是不是最小的。假设是那么就能够接替之前的工作者线程成为active的了。这时假设有新加入的工作者线程也会触发变更通知，但这并不会影响正常的逻辑。
当然这里还存在一些问题有待完好：
- active线程由于网络延迟出现短时会话丢失的问题，可能会导致Zookeeper错误的推断
- 子节点频繁变更可能会产生广播风暴
动态运行新任务
每一个任务组都会有一个watcher来监控是否有新的任务被创建（比方一种新的日志类型被提交）。假设有新任务则会在其所属的线程池中新建新的工作者来运行新任务。
眼下暂时这个watcher默认仅仅关注新增任务，而针对任务被移除或者任务的元数据变更，watcher暂时还没有相应的响应机制。这也是兴许须要考虑和完好的部分。
后台服务配置模板化
这样的机制之前已经分别应用于日志採集、日志解析模块了，在这里也是为了简化后台服务启动时配置的问题。
总结
综上，整个设计的Zookeeper 拓扑图大致例如以下：
从上面的分析能够看到。我们最大程度地将各种可变的參数配置到Zookeeper中。使其成为串联起整个分布式系统的配置中心，而我们的管控台某种意义上退化成了“配置系统”——将配置界面化、持久化。
同一时候。我们也利用了Zookeeper的实时事件Push机制，来进行分布式协调。
