[转载]sensei分布式实时搜索系统源码解析(三) 分布式index

 sensei分布式实时搜索系统源码解析(三) 分布式index 

 

前两回写完之后，这篇关于sensei如何建立索引的部分，至今日才补上，有些惭愧。一方面，初期没有细看index这块，另外，其他事情导致精力有所分散。话转正题。

一、提供流数据的GateWay

       sensei 处理建立索引的过程，可以有多种方式，总体而言，提供了一个SenseiGateWay的抽象类，

 

目前由如上4种实现类，并且通过注册至SenseiGatewayRegistry，来提供静态的获取getDataProviderBuilder的方法。

      根据官方的文档，下图中的4种方式均可作为sensei的数据输入的gateway。分别对应上面4个实现类。

 

对这四类对简要的分析可以发现：

1. LinedFileDataProvider 通过行读文件解析成jsonObject来提供流读取数据。通常来说，此种通过file读文件类无法或者说不方便做实时的更新，比较适合批量全量重建索引；

2. JdbcDataProvider ，通过对数据库jdbc连接来进行流数据的读取，其实是需要实现SenseiJDBCAdaptor这个接口的buildStatment和extractVersion的方法，来实现对流的控制，主要就是version的控制，熟悉zoie的开发者应该都知道，version相当于数据的一个标签。作为数据消费到的offset，build查询语句的时候也要带上这个参数。

      eg： select pid,title,content,author from post where lastmodify>"2011-09-21";   这样一条语句，让业务层保证对每一条数据的更新均有唯一的标示可以作为查询语句的一个限制条件，这样才能获取到新的数据来进行实时数据的索引更新。比较普遍的做法是设计数据表结果的时候，增加lastmodify字段来标示数据的最后修改时间来作为version。

     但现实状况是一个索引的document实体数据可能来源于不同的表，而且业务处理上很可能一些表并不能提供一个唯一的version的标识。因此， JdbcDataProvider 也不太适合做实时索引的更新。

3. JMS的dataprovider，这个比较方便，采用JMS的标准来提供数据。使用Topic/Subscribe的方式来进行消息的订阅。一方面能够实现了多patition对同一数据的消费需求，另一方面也提供了应用独立实现jms的可能性。 我就是采用这种方法来进行更新，JMS选用Activemq，成熟稳定，支持持久化，主要自己之前用过，比较熟悉。 推荐大家采用该方法来进行更新。

4. KafkaDataprovider 这个是linkedin开源的另一个消息队列。看介绍其实现比较独特，据说效率挺高，自己没有细看过，不好发表意见，以后打算研究研究这个kafka。

5. 除开以上4中dataprovider外，sensei中还提供了一个httpStreamDataProvider的类，通过http的请求来获取next()数据，相当于这样一个流程：httpclient 定时(retryTime)去抓取一个version之后的批量数据DataEvent list，每次next()获取当前dataEventList的一条数据，当消费完dataEventList，再请求下一批数据，如果version之后没有数据，则仍旧按照sleep retryTime后再获取。

      个人感觉上此种更新方式比较适合从第三方的网站来更新数据的情形。

 

 

二、深入sensei的索引更新流程

    第一部分，描述了sensei提供的几种建立索引的方式，但这些流数据在sensei内部是如何被更新到索引里面去的呢？此处主要讲解处理数据的流程。

    索引的更新流程，还是得从senseiServer启动的时候说起，在SenseiServerBuilder在进行buildCore的时候，会根据读取的配置文件来初始化两个建立索引的核心类：SenseiZoieSystemFactory和DefaultStreamingIndexingManager。

      SenseiZoieSystemFactory提供每个node产生多个数据分片(partition)的zoieSystem的工厂方法。一个partition对应一个zoieSystem，相当于对更新数据的消费者，用于建立索引。SenseiZoieSystemFactory主要由SenseiIndexReaderDecorator和DefaultJsonSchemaInterpreter来初始化。 sensei的indexReaderDecorator在初始化时读取配置文件来构建一些facethandler和runtimeFacetHandler，用于bobo做切面，是lucene中indexReader的装饰者类。而sensei默认的是jsonInterproter将provider提供的jsonObject转化为lucene的Document，可喜的是在Sensei3.0.0以后提供了一个CustomIndexingPipeline的接口，可以将实现我自己的对索引的特殊处理，如按照业务需求对特定的document加权，之类的。最近正好一个工作中特殊的业务需求通过该接口来实现了，感觉很不错。注意，该接口在sensei2.7.0的版本中未提供，3.0以上才有，如果需要请update最新版的源码。

    DefaultStreamingIndexingManager则是sensei索引更新的最核心类。肩负着流数据的消费，sharding策略的数据分发，zoieSystem消费数据的管理功能。其由senseiSchame，conf等参数构建，sharding策略默认是FieldModShardingStrategy按照uid取模进行分片。在SenseiCore start的时候，其完成了真正的初始化，SenseiZoieSystemFactory根据本node上的配置的partition来得到多个zoieSystem，indexingManager调用initialize时，根据配置文件中maxpartition初始化DataDispatcher，这个dispatcher作为streamDataprovider的消费者，来做数据的分发；而真正消费数据的zoieSystemMap也被set上。

   索引消费流程：

1. JmsStreamDataProvider 调用next()，触发flush方法；

2. dataprovider调用其consumer的consume方法，即作为DefaultStreamingIndexingManager的内部类DataDispatcher的consume方法；

3. dispatch根据sharding策略将该批数据分发至对应的_dataCollectorMap里；

4. 由本node上parition数据的zoieSystem对进行_dataCollectorMap中的数据进行消费，建索引。

   原本对于zoie中来说，dataProvider和zoieSystem是一对经典的生产者/消费者组合；而在sensei里面，dataprovider与dispatcher是一对生产者/消费者，而dispatch与zoieSystem是另一组生产者/消费者。

再看zoieSystem是如何来消费这个数据的呢？这个就得跟进zoie的源码里面。

zoiesystem的父类AsyncDataConsumer 做consume的时候给其自身的_batch里添加数据，同时，ConsumerThread 将_batch 数据交给更_consumer来处理。

当走realtime search的时候，由RealtimeIndexDataLoader来完成。

RealtimeIndexDataLoader 就是包含有RAMLuceneIndexDataLoader，当RealtimeIndexDataLoader 消费的时候首先由ramloader来进行消费，但RealtimeIndexDataLoader 的父类有个loadThread，在执行processBatch也就是说loadthread在判断ramloader消费的数据是否到达batchsize，到达的话，将触发working状态，由diskloader与该ramloader进行merge索引。

三、关于Sensei JMSDateprovider sharding策略前置的思考

   在看sensei源码的过程中，发现在每个node上都会订阅所有的data数据，只是在本地的indexmanager根据当前node上配置的partition来进行zoieSystem的消费。意味着所有data均传输到了所有的节点，而对于某个node上partition不需要的数据，其实是一种浪费。

   我在思考另一种方案：由发送索引更新数据的应用来负责sharding策略，在sharding后，写入到对应的partition的topic队列里，不同的node根据本node上的partition来决定订阅哪些topic。这样可以减少数据的传输。同样有弊端，先就两种方案进行对比：

1.topic/sub, 所有的topic 均被每个node订阅，由每个node来进行patition的判断来确定是否来索引该数据。
restrict： 数据需要传输至每个node，发布传输订阅的开销增大。
postive： 数据写入方不care MQ之后node的结构，对发布应用来说透明。

node扩展对前端发布应用服务器透明

2.如果将sharding strategy提前至发布应用服务器,则后端每个node索引服务器只需要接受其对应的patition的数据。
postive： 数据冗余发送介绍，特别是在索引node增多的情况下，传输尤为明显。
restrict： 发布应用服务器需要zookeeperclient来通知后台node的patition的变更，结构相对复杂，node扩展改动大

    综合来看，应该是sensei的开发者认为让逻辑结构更清晰，让外部应用写数据不care 索引后端的sharding策略来得更合理吧。

 

 

来源：

http://johnnychenjun.blog.163.com/blog/static/137493406201182615540576/

 

后记： 这个貌似也比较老了，sensei的新版本1.1.2-SNAPSHOT，整个包名都修改了。不过整体体系结构还是有点类似。