RDBMS的lucene爬虫

主要介绍如何周期性尽量实时地从RDBMS爬数据然后建索引，不涉及AOP或ORM Framework的listener方式。

先决条件：

1. Lucene索引是从无到有的，一开始所有数据都是存储在RDBMS（Oracle）中。
2. 数据表有一列是updateTime或称为lastModifiedTime用来存储最后一次更新时间，并建有db索引
3. 主表必须要有主键，这个主键也用来唯一确定一个Lucene document。

该策略大致可以分为以下几个部分：

1．索引结构

2．初始化索引

3．增量索引

4．补偿操作

5．删除检测

6．备份

7．注意点

索引结构：

我们是做全文搜索的，所以db中的用户需要全文搜索的字段值都会拼接成一个string存储在lucene的一个字段中称为content。其他的需要存储的字段或者字段分析后的token是有意义单词的我们称为metadata，每个metadata是document的一个field。

从索引结构看来，我们有一个主表，不同的主表OID的记录对应不同的lucene document。主表依赖的所有子表中的数据如果是需要做全文搜索的就需要append在一起然后用Ngram进行分词处理。有的数据是需要存储的或有特殊用途的，比如访问控制，就不分词或者用特殊的analyzer分词。

简单的介绍一下，具体会在以后的文章中谈到。

初始化索引

初始化索引，是索引从无到有的一个过程。也是系统第一次初始化时做的事情。初始化时我们会首先获取当前时间，然后将updateTime早于当前时间的所有数据取出来。最后把当前时间存储到timeTrace.properties文件中。作为增量索引的起始依据。

取什么？

用户预先在配置文件配置一个主SQL，通过这个sql我们取出需做全文索引的content和metadata。对于某些特定的metadata不希望被build到全文索引中的，可以单独配置sql。

如何取，多线程分批取。我们会有一个线程池，每个分批都是作为该线程池的一个task被提交执行。

·         查询出当前时间之前的记录总数n

·         按照rownum分批。假设分批的大小是2000，则需要将rownum为0, 2000，4000，6000…..max(rownum)的所有OID都一次拿回来。

Select oid from x where rownum=0 or rownum=2000,………….or rownum=n

如果直接根据rownum分批取，会出现幻影读的问题，因为rownum每次查询都会发生变化，如果有新数据插入，改用当时snapshot的OID去取，避免这个问题，因为OID是不变的。有人会说，这样可能查到新插入的数据，当然是这样的，但是查到新加入的不会有影响，新加入的也是迟早需要爬的，但是用rownum还会丢失数据。

·         根据拿回来的断点的OID，分批取。比如令先前获取的rownum为2000的oid是10231，拿回来的4000的oid是14215

前两个分批的sql就是：

Select * from x where oid >= 0 and oid < 10231;

Select * from x where oid >=10231 and oid < 14215;

具体的sql和oid是哪张表的哪列都是在配置文件中预先配置的。

·         当然每个task还要负责将从db中爬出的数据建索引，建完索引后提交对索引的改动

·         所有task都跑完之后，主线程将一开始获取的当前时间更新到timeTrace.properties文件中。

当中可能会出现问题，如果出现问题，就需要重新初始化，初始化之前会清除所有已建好的部分脏索引。因为当前时间没有更新到timeTrace.properties文件中。我们测试下来百万级的数据这个过程大概需要10分钟。

增量索引

在初始化索引成功后，当时的时间已经被更新到timeTrace.properties。增量索引是从这个时间点开始定期地被触发执行，可以使用quartz来管理这个timing job。增量索引称为incrementalIndexService，增量索引服务的不同任务调度之间需要同步执行，用quartz的stateful job可以实现，或者使用内存，文件或DB锁。

·         第一步，从timeTrace文件中获取已爬数据的截至时间Tlast，获取当前时间Tcurrent。

·         Select count(*) from X where updateTime > Tlast and updateTime <= Tcurrent, 结果记为n，如果n小于分批的大小就直接爬出这段的索引数据。如果n大于分批大小，就需要将这个n个结果分批.

·         这次我们按照时间段分批，如果n/2000 = 3但有余数, 那就说明要分四批拿，将这个时间段Tcurrent-Tlast平均分为四段。每个线程处理其中的某段。

·         某线程将它负责的某段数据拿回来之后，首先判断这个OID是否在index已经存在，如果存在就说明在这个时间段里这条记录是被用户update过的，index也做相应的update。如果这个OID在index中不存在，则说明这条记录是新加入到db中的，index也做add操作。做完之后提交。

·         当所有分批都完成之后，更新timeTrace文件，把时间更新为Tcurrent。

·         一旦有分批出现问题失败，整个时间段就认为是不成功的，需要重新爬一遍。

一般增量服务我们设置的间隔都小于1分钟，因为需要拿出最实时的数据，而且每次获取数据的结束时间都是当前时间。保证数据的实时性。

补偿操作

补偿操作在整个爬虫策略中是最复杂的一个环节。采用增量索引看似天衣无缝，其实还是有风险的。因为记录到db的updateTime往往都是有延迟的，一般情况下是java端的时间或是记录写入DB的时间，都早于commit时间，但一般数据库的隔离级别都是read committed。只有在数据被提交后才可能被增量服务看到。这样的话3点跑的增量服务，先前的结束时间是2点58分。这时它需要获取2.58到3.00之间的数据，但是此时有可能java端正有一条记录生成，它的updateTime是 2.59，但是它一直没有commit，因为transaction的超时时间是10分钟。悲剧了发生了，这条数据将永远不会被爬出来，除非遥远的将来有人再次更新它。因为这个时间段已经被爬过了，按照增量服务，它是永远不会再爬timeTrace文件中记录的时间之前的数据的。

此时，补偿服务隆重登场。它存在的价值就是把所有可能被遗漏的数据都查出来。关键点就是要找出在补偿服务运行时，哪个时间段的数据是可能被遗漏的而哪个时间段的数据又是永远不会被丢失的。那个永远不会丢失的时间段就没必要再去管它。我们关心的只是可能被遗漏的那段时间段的数据。

我们来看个例子：

·         增量服务每1分钟跑一次, 周期记为P(N) = 1

·         Transaction Timeout时间3分钟，记为TO

·         补偿服务每x分钟跑一次，P(C)  = x >= P(N)

下图的第一条时间轴T(N)是增量服务的，第二条是补偿服务的T(C)。

 

 

 

 

在这个分级策略中级数n, 每一级的P(c), TO, Te都是可调的，但需要注意最后一级的TO是不可调的必须等于真正的transaction timeout时间，Te的取值范围是[Last(N)-TO, Last(N)]。

 

调优的依据是我们会记录每次补偿操作的历史记录，比如每次补偿成功的个数，补偿运行的开始，结束时间等。

              

删除检测

 

增量索引服务只是负责update和add的检测，它并不判定索引中document对应的记录在DB中是否已经被删除，索引中会积累很多在DB中已经被清除的数据。这些document也要及时地从索引中删除。所以会有一个定期的删除检测服务，检测出那些在索引中有，而在DB中已经被物理删除的记录。

 

删除检测服务的步骤：

l           从索引中分批取出所有OID，根据OID排序

l           用每个分批的最小值和最大值到DB中取出此OID段DB中存在(没有被删除)的所有OID，也根据OID排序

Select OID from X where oid>12001 and oid<24100 order by oid;

l           将索引中查处的OID      有序集合和DB中获得的OID有序集合进行对比,如果DB中没有索引中有的就添加到deletedList中.

l           把deletedList中的所有记录对应document从索引中删除

 

对于软删除，它们的状态属性active如果已经被爬到索引中，直接从索引中选择出那些active=0的document删除，如果没有，可以将删除检测的sql语句改成

Select OID from X where active = 0 and oid>12001 and oid<24100 order by oid;

其他步骤同上面硬删除的部分

 

备份

备份时需要注意不仅要备份最后一次commit之前的所有索引,而且需要备份timeTrace文件. 恢复后只需要从timeTrace的时间开始爬就可以了.

 

注意点

主表updateTime没有更新

有时候，业务逻辑更新了子对象，比如JobOrder对象包含了很多个Container对象，一个JobOrder对应一个Lucene Document，当Container对象更新时，它并没有更新JobOrder的updateTime，只是更新了Container的updateTime。这也没关系，我们再增量服务和补偿策略中同时也会查出子表updateTime在当前时间段的所有主表数据。

但container如果有删除，就必须约定application必须要update主表的updateTime。否则用户就会搜出他本不能访问的被删除的container。

算法

补偿服务的算法主要目的就是比较出遗漏的数据。为了比较有无遗漏，我们需要把db中的数据和增量服务已经爬过的数据进行比较才知道。我们会在内存中存放增量服务已经爬过的数据的oid和updateTime，在内存中存放是为了提高性能。每次补偿服务运行时，也会把完全可见区从内存中清除。

 

·         增量服务每次执行后就会将爬出的数据的OID和updateTime保存在内存中，内存中有一棵二叉排序树维护OID和updateTime的pair。排序的key是OID。二叉排序树可以用TreeSet实现。

·         补偿服务运行时，先创建两个List一个用来存放需要update数据的oid，记为updateList，另一个用来存放add数据的oid，记为addList

·         接着从DB中取出Ts到Te之间所有数据的OID和updateTime，也是根据OID排序的。

·         计算下一次补偿服务的开始时间，Next(Ts) = Last(N) –TO;

·         现在就是要比较两个有序集合。树的访问者应该写成：

if(OID(C) == OID(N))

     if(updateTime(C) > updateTime (N))

     {

                  更新OID(N)的updateTime;

                   updateList.add(OID(N));

        }

        else{

       addList.add(OID(N))

       }

        //清除完全可见区，下一次补偿服务开始时间之前的数据

           if(updateTime(N) < Next(Ts)) {

           tressSetIterator.remove();

          }

       

·         将updateList中OID对应的所有数据从db中获取并update到索引中

·         将addList中OID对应的所有的数据从db中获取并add到索引中

·         Commit索引，并将Next(Ts)记录到updateTrace文件中

 

意外情况

补偿服务很久没有被调度

一般不会出现，因为我们会将增量服务和补偿服务的线程优先级设为相同的。应该会被分时处理。如果很久没有不会被调度，正确性是可以保证的，因为开始时间都是记录在文件中的，如果一直没有跑，只是一下子补偿的时间段很长，并不会丢失补偿的时间段。但是不排除内存溢出的风险，因为存储在内存中的treeset在这种情况下会很大。在treeset很大时，我们可以检测，如果超过一定的节点数，就可以将treeset序列化到一个internal索引中，下次取出来时也是有序的。甚至可以分块取出比较。

 

Server突然shutdown

Server shutdown突然shutdown，线程被interrupt掉，没有执行完，内存中的树也没了。这时就需要每次启动时，这个时间段内的所有数据都会认为是需要add到索引的，这样就会出问题。所以需要提前检测，每次系统启动时，补偿服务需要把这段时间内的所有记录的OID和updateTime从db中获取，直接和索引中的进行比较，比较效率要低一些。但也不会出现数据丢失的情况。

 

 

初始化值

系统初始化时，补偿服务初始化Ts（被扫描数据的起始时间）是，初始化索引的当前时间-TO。   而补偿服务本身的开始时间是在增量服务开始之后。之后多少可以调。

 

优化

优化的重点放在了以下几个方向。

·         对DB压力

·         补偿延迟

·         消耗内存的大小

·         比较次数

 

以上选项之间有的都是矛盾的，比如说补偿延迟要小，则补偿服务的P(C)就要小，则查询DB的次数就增加，对DB压力就增大。

所以针对不同的使用情况，比如DB资源，延迟的可接受程度，应用服务器资源等，我们可能需要采用不同的策略，这就要我们的补偿策略可调。

为了可调，我们不仅使一些参数可以配置，而且引入了分级补偿服务的方案。在分级方案中，如果分n级，则n-1级的TO输入值推荐和P(C)是相同的，但也是可调的。

 

举个例子：一个三级补偿服务，

第一级：为了使补偿的延迟最小，极端情况下我们可以采用和增量服务相同的周期假设为1分钟，此时TO的输入值也是周期值。此级的启动时间也是初始化当前时间记为Tinitial+P(N).

 

第二级: 业务场景中绝大多数事务都是在3分钟内完成的，如果TO是3分钟，基本上绝大多数事务都可以及时的补偿到。此级的启动时间是Tinitial+3

 

第三级：也是最后一级，在App Server中配置的真正的TO是10分钟，为了保证正确性，TO的输入值一定要是10分钟，因为只需要保证正确性所以它的频率也不需要太频繁周期也设为10分钟。从前文中可知Last(N) – TO < Te < Last(N). 此级没有必要多实时，所以Te就取最小值=Last(N)-TO.

 

我们将这个三级策略和一级策略进行比较，我们假设一级策略的周期为2分钟。假设整个时间段是10分钟。

 

比较项	一级(2)	三级(1, 3, 10)
DB访问次数	5	14
延迟	2	<2
内存	11分钟数据的OID 和updateTime	17
访问DB的数据量	12×5=60	10+6×10/3+2×10=50
比较次数	60	50

对于补偿服务我们需要确定每次的开始时间Ts，结束时间Te，周期P(C), 算法，初始化值，意外情况，优化等方面。

 

结束时间Te

补偿服务的目的是对增量索引进行补偿，所以它所补偿的时间区间一定是增量服务已经处理过的。所以它的结束时间一定是timeTrace文件中最后一次增量服务记录的时间我们记为Last(N). Last(N)之后的数据或者正在被增量服务处理或者没有被增量服务处理，如果补偿服务去涉及这些数据，那肯定全是要补偿的，但是增量服务也会去处理，一个是会重复处理，一旦需要补偿，我们会把这条记录的所有数据都从DB端取过来，建索引，重复的代价也是很大的。为了避免这个代价: Te < Last(N).

 

如果Te = Last(N), 我们就会拿到最新的需要补偿的数据，补偿服务的延迟就最小。Te < Last(N), 每次都没有不尝到最新的数据，遗漏数据被检测到就会有延迟，只有等下一次补偿服务触发时才能被检测出。

 

开始时间Ts

上图中，transaction timeout之前的那个时间段，如果有数据生成，T(N) = 1, 要么在图上标出的[T(N) =1，T(N) = 4]之间被提交，要么transaction超时该数据也不会写入到DB中。所以Last(N) – TO之前的数据在这一次补偿服务的时候已经是完全可见的，肯定都会被补偿的。对于下一次来说这块也是不需要再被补偿的。完全可见区（针对下一次补偿操作）必须满足下面两个条件：

·         被增量服务处理过并且更新已经完全对本次补偿服务可见

·         已经被补偿服务处理过

 

则下一次补偿操作就不会再关心Last(N)-TO之前的数据了。我们把上一次补偿服务记为Last(C), 而此次的增量服务记为Last(C Last(N)).  则下一次补偿服务的开始时间<=Last(N)-TO.因为大于这个时间的所有数据都是需要被补偿的。换个表达方式，此次的补偿服务的开始时间是由上一次补偿服务计算的得到的，为Last(C Last(N)) – TO.  Ts <= Last(C Last(N)) – TO

 

同时我们需要注意的是，Last(N) – TO 到Last(N)每次补偿服务是必须要检测的，不然就会有遗漏，因为我们假设了不可见区，前提条件就是每次都会检测这个区域。所以结束时间: Last(N) – TO < Te < Last(N).

 

开始时间是由上一次补偿服务计算得到的，那这个值就需要保存下来。保存在文件中可以避免系统down掉后丢失。我们也会将这个时间值保存到timeTrace.properties的compensation属性上。