一步一步跟我学习lucene（19）---lucene增量更新和NRT(near-real-time)Query近实时查询

这两天加班，不能兼顾博客的更新，请大家见谅。

有时候我们创建完索引之后，数据源可能有更新的内容，而我们又想像数据库那样能直接体现在查询中，这里就是我们所说的增量索引。对于这样的需求我们怎么来实现呢？lucene内部是没有提供这种增量索引的实现的；

这里我们一般可能会想到，将之前的索引全部删除，然后进行索引的重建。对于这种做法，如果数据源的条数不是特别大的情况下倒还可以，如果数据源的条数特别大的话，势必会造成查询数据耗时，同时索引的构建也是比较耗时的，几相叠加，势必可能造成查询的时候数据缺失的情况，这势必严重影响用户的体验；

比较常见的增量索引的实现是：

• 设置一个定时器，定时从数据源中读取比现有索引文件中新的内容或是数据源中带有更新标示的数据。
• 对数据转换成需要的document并进行索引

这样做较以上的那种全删除索引然后重建的好处在于：

• 数据源查询扫描的数据量小
• 相应的更新索引的条数也少，减少了大量的IndexWriter的commit和close这些耗时操作

以上解决了增量的问题，但是实时性的问题还是存在的：

• 索引的变更只有在IndexWriter的commit执行之后才可以体现出来

那么我们怎样对实时性有个提升呢，大家都知道lucene索引可以以文件索引和内存索引两种方式存在，相较于文件索引，内存索引的执行效率要高于文件索引的构建，因为文件索引是要频繁的IO操作的；结合以上的考虑，我们采用文件索引+内存索引的形式来进行lucene的增量更新；其实现机制如下：

• 定时任务扫描数据源的变更
• 对获得的数据源列表放在内存中
• 内存中的document达到数量限制的时候，以队列的方式删除内存中的索引，并将之添加到文件索引
• 查询的时候采用文件+内存索引联合查询的方式以达到NRT效果

定时任务调度器

java内置了TimerTask，此类是可以提供定时任务的，但是有一点就是TimerTask的任务是无状态的，我们还需要对任务进行并行的设置；了解到quartz任务调度框架提供了有状态的任务StatefulJob，即在本次调度任务没有执行完毕时，下次任务不会执行；

常见的我们启动一个quartz任务的方式如下：

[java] view plain copy

1. Date runTime = DateBuilder.evenSecondDate(new Date());  
2.    StdSchedulerFactory sf = new StdSchedulerFactory();  
3.       Scheduler scheduler = sf.getScheduler();  
4.    JobDetail job = JobBuilder.newJob(XXX.class).build();  
5.       Trigger trigger = TriggerBuilder.newTrigger().startAt(runTime).withSchedule(SimpleScheduleBuilder.simpleSchedule().withIntervalInSeconds(3).repeatForever()).forJob(job).build();  
6.       scheduler.scheduleJob(job, trigger);  
7.         
8.       scheduler.start();</span>  

以上我们是设置了每三秒执行一次定时任务，而任务类是XXX

任务类通用方法

这里我定义了一个XXX的父类，其定义如下：

[java] view plain copy

1. package com.chechong.lucene.indexcreasement;  
2.   
3. import java.util.List;  
4. import java.util.TimerTask;  
5.   
6. import org.apache.lucene.store.RAMDirectory;  
7. import org.quartz.Job;  
8. import org.quartz.StatefulJob;  
9.   
10. /**有状态的任务：串行执行，即不允许上次执行没有完成即开始本次如果需要并行给接口改为Job即可 
11.  * @author lenovo 
12.  * 
13.  */  
14. public abstract class BaseInCreasementIndex implements StatefulJob {  
15.     /** 
16.      * 内存索引 
17.      */  
18.     private RAMDirectory ramDirectory;  
19.     public BaseInCreasementIndex() {  
20.     }  
21.     public BaseInCreasementIndex(RAMDirectory ramDirectory) {  
22.         super();  
23.         this.ramDirectory = ramDirectory;  
24.     }  
25.       
26.     /**更新索引 
27.      * @throws Exception 
28.      */  
29.     public abstract void updateIndexData() throws Exception;  
30.     /**消费数据 
31.      * @param list 
32.      */  
33.     public abstract void consume(List list) throws Exception;  
34. }  

任务类相关实现,以下方法是获取待添加索引的数据源XXXInCreasementIndex

[java] view plain copy

1. @Override  
2.     public void execute(JobExecutionContext context) throws JobExecutionException {  
3.         try {  
4.             XXXInCreasementIndex index = new XXXInCreasementIndex(Constants.XXX_INDEX_PATH, XXXDao.getInstance(), RamDirectoryControl.getRAMDireactory());  
5.             index.updateIndexData();  
6.         } catch (Exception e) {  
7.             // TODO Auto-generated catch block  
8.             e.printStackTrace();  
9.         }  
10.     }  

[java] view plain copy

1. @Override  
2.     public void updateIndexData() throws Exception {  
3.         int maxBeanID = SearchUtil.getLastIndexBeanID();  
4.         System.out.println(maxBeanID);  
5.         List<XXX> sources = XXXDao.getListInfoBefore(maxBeanID);、、  
6.         if (sources != null && sources.size() > 0) {  
7.             this.consume(sources);  
8.         }  
9.     }  

这里，XXX代表我们要获取数据的实体类对象

consume方法主要是做两件事：

• 数据存放到内存索引
• 判断内存索引数量，超出限制的话以队列方式取出超出的数量，并将之存放到文件索引

[java] view plain copy

1. @Override  
2.     public void consume(List list) throws Exception {  
3.         IndexWriter writer = RamDirectoryControl.getRAMIndexWriter();  
4.         RamDirectoryControl.consume(writer,list);  
5.     }  

上边我们将内存索引和队列的实现放在了RamDirectoryControl中

内存索引控制器

首先我们对内存索引的IndexWriter进行初始化，在初始化的时候需要注意先执行一次commit，否则会提示no segments的异常

[java] view plain copy

1. private static IndexWriter ramIndexWriter;  
2.     private static RAMDirectory directory;  
3.     static{  
4.         directory = new RAMDirectory();  
5.         try {  
6.             ramIndexWriter = getRAMIndexWriter();  
7.         } catch (Exception e) {  
8.             // TODO Auto-generated catch block  
9.             e.printStackTrace();  
10.         }  
11.     }  
12.     public static RAMDirectory getRAMDireactory(){  
13.         return directory;  
14.     }  
15.     public static IndexSearcher getIndexSearcher() throws IOException{  
16.         IndexReader reader = null;  
17.         IndexSearcher searcher = null;  
18.         try {  
19.             reader = DirectoryReader.open(directory);  
20.         } catch (IOException e) {  
21.             e.printStackTrace();  
22.         }  
23.         searcher =  new IndexSearcher(reader);  
24.         return searcher;  
25.     }  
26.     /**单例模式获取ramIndexWriter 
27.      * @return 
28.      * @throws Exception  
29.      */  
30.     public static IndexWriter getRAMIndexWriter() throws Exception{  
31.             if(ramIndexWriter == null){  
32.                 synchronized (IndexWriter.class) {  
33.                     Analyzer analyzer = new IKAnalyzer();  
34.                     IndexWriterConfig iwConfig = new IndexWriterConfig(analyzer);     
35.                     iwConfig.setOpenMode(OpenMode.CREATE_OR_APPEND);    
36.                     try {  
37.                         ramIndexWriter = new IndexWriter(directory, iwConfig);  
38.                         ramIndexWriter.commit();  
39.                         ramIndexWriter.close();  
40.                         iwConfig = new IndexWriterConfig(analyzer);     
41.                         iwConfig.setOpenMode(OpenMode.CREATE_OR_APPEND);    
42.                         ramIndexWriter = new IndexWriter(directory, iwConfig);  
43.                     } catch (IOException e) {  
44.                         // TODO Auto-generated catch block  
45.                         e.printStackTrace();  
46.                     }  
47.                 }  
48.             }  
49.           
50.         return ramIndexWriter;  
51.     }  

定义一个获取内存索引中数据条数的方法

[java] view plain copy

1. /**根据查询器、查询条件、每页数、排序条件进行查询 
2.      * @param query 查询条件 
3.      * @param first 起始值 
4.      * @param max 最大值 
5.      * @param sort 排序条件 
6.      * @return 
7.      */  
8.     public static TopDocs getScoreDocsByPerPageAndSortField(IndexSearcher searcher,Query query, int first,int max, Sort sort){  
9.         try {  
10.             if(query == null){  
11.                 System.out.println(" Query is null return null ");  
12.                 return null;  
13.             }  
14.             TopFieldCollector collector = null;  
15.             if(sort != null){  
16.                 collector = TopFieldCollector.create(sort, first+max, false, false, false);  
17.             }else{  
18.                 SortField[] sortField = new SortField[1];  
19.                 sortField[0] = new SortField("createTime",SortField.Type.STRING,true);  
20.                 Sort defaultSort = new Sort(sortField);  
21.                 collector = TopFieldCollector.create(defaultSort,first+max, false, false, false);  
22.             }  
23.             searcher.search(query, collector);  
24.             return collector.topDocs(first, max);  
25.         } catch (IOException e) {  
26.             // TODO Auto-generated catch block  
27.         }  
28.         return null;  
29.     }  

此方法返回结果为TopDocs，我们根据TopDocs的totalHits来获取内存索引中的数据条数，以此来鉴别内存占用，防止内存溢出。

consume方法的实现如下：

[java] view plain copy

1. /**消费数据 
2.      * @param docs  
3.      * @param listSize  
4.      * @param writer 
5.      * @param list 
6.      * @throws Exception  
7.      */  
8.     public static void consume(IndexWriter writer, List list) throws Exception {  
9.         Query query = new MatchAllDocsQuery();  
10.         IndexSearcher searcher = getIndexSearcher();  
11.         System.out.println(directory);  
12.         TopDocs topDocs = getScoreDocsByPerPageAndSortField(searcher,query, 1, 1, null);  
13.         int currentTotal = topDocs.totalHits;  
14.         if(currentTotal+list.size() > Constants.XXX_RAM_LIMIT){  
15.             //超出内存限制  
16.             int pulCount = Constants.XXX_RAM_LIMIT - currentTotal;  
17.             List<Document> docs = new LinkedList<Document>();  
18.               
19.             if(pulCount <= 0){  
20.                 //直接处理集合的内容  
21.                 TopDocs allDocs = SearchUtil.getScoreDocsByPerPageAndSortField(searcher, query, 0,currentTotal, null);  
22.                 ScoreDoc[] scores = allDocs.scoreDocs;  
23.                 for(int i = 0 ;i < scores.length ; i ++){  
24.                     //取出内存中的数据  
25.                     Document doc1 = searcher.doc(scores[i].doc);  
26.                     Integer pollId = Integer.parseInt(doc1.get("id"));  
27.                     Document doc = delDocumentFromRAMDirectory(pollId);  
28.                     if(doc != null){  
29.                         XXX carSource = (XXX) BeanTransferUtil.doc2Bean(doc, XXX.class);  
30.                         Document doc2 = carSource2Document(carSource);  
31.                         if(doc2 != null){  
32.                             docs.add(doc2);  
33.                         }  
34.                     }  
35.                 }  
36.                 addDocumentToFSDirectory(docs);  
37.                 writer = getRAMIndexWriter();  
38.                 consume(writer, list);  
39.             }else{  
40.                 //先取出未达到内存的部分  
41.                 List subProcessList = list.subList(0, pulCount);  
42.                 consume(writer, subProcessList);  
43.                 List leaveList = list.subList(pulCount, list.size());  
44.                 consume(writer, leaveList);  
45.             }  
46.         }else{//未超出限制，直接存放到内存  
47.             int listSize = list.size();  
48.             if(listSize > 0){  
49.                 //存放到内存  
50.   
51.             }  
52.         }  
53.           
54.     }  

上边的逻辑为：

1. 根据getScoreDocsByPerPageAndSortField获取当前内存中的数据条数
2. 根据内存中数据数量A和本次获取的数据源的总数B和内存中限制的数量C进行比较
3. 如果A+B<=C则未超出内存索引的限制，所有数据均存放到内存
4. 反之，判断当前内存中的数据是否已经达到限制，如果已经超出，则直接处理取出内存中的内容，然后回调此方法。
5. 如果未达到限制，先取出未达到限制的部分，然后对剩余的进行回调。

这里我们的BeanTransferUtil是根据document转换成对应的bean的方法，此处用到了反射和commons-beanutils.jar

[java] view plain copy

1. package com.chechong.util;  
2.   
3. import java.lang.reflect.Field;  
4. import java.lang.reflect.InvocationTargetException;  
5.   
6. import org.apache.commons.beanutils.BeanUtils;  
7. import org.apache.lucene.document.Document;  
8.   
9.   
10. public class BeanTransferUtil {  
11.   
12.     public static Object doc2Bean(Document doc, Class clazz) {  
13.         try {  
14.             Object obj = clazz.newInstance();  
15.             Field[] fields = clazz.getDeclaredFields();  
16.             for (Field field : fields) {  
17.                 field.setAccessible(true);  
18.                 String fieldName = field.getName();  
19.                 BeanUtils.setProperty(obj, fieldName, doc.get(fieldName));  
20.             }  
21.             return obj;  
22.         } catch (InstantiationException e) {  
23.             // TODO Auto-generated catch block  
24.             e.printStackTrace();  
25.         } catch (IllegalAccessException e) {  
26.             // TODO Auto-generated catch block  
27.             e.printStackTrace();  
28.         } catch (InvocationTargetException e) {  
29.             // TODO Auto-generated catch block  
30.             e.printStackTrace();  
31.         }  
32.         return null;  
33.     }  
34. }  

从内存索引中读取索引的方法如下：

[java] view plain copy

1. /**从内存索引中删除指定的doc 
2.      * @param pollId 
3.      * @throws IOException  
4.      */  
5.     private static Document delDocumentFromRAMDirectory(Integer pollId) throws IOException {  
6.         Document doc = null;  
7.         Query query = SearchUtil.getQuery("id", "int", pollId+"", false);  
8.         IndexSearcher searcher = getIndexSearcher();  
9.         try {  
10.             TopDocs queryDoc = SearchUtil.getScoreDocsByPerPageAndSortField(searcher, query, 0, 1, null);  
11.             ScoreDoc[] docs = queryDoc.scoreDocs;  
12.             System.out.println(docs.length);  
13.             if(docs.length > 0){  
14.                 doc = searcher.doc(docs[0].doc);  
15.                 System.out.println(doc);  
16.                 ramIndexWriter.deleteDocuments(query);  
17.                 ramIndexWriter.commit();  
18.             }  
19.             return doc;  
20.         } catch (IOException e) {  
21.             // TODO Auto-generated catch block  
22.             e.printStackTrace();  
23.         }  
24.         return null;  
25.     }  

此处是根据id来读取内存索引中的内容，然后将它转换成document同时删除内存中的对应记录。

NRT近实时查询的实现

对于上边的索引我们要采用适当的查询方法，这里查询时候为了达到近实时的效果，需要将内存索引添加到查询的范围中，即IndexReader中。

这里的IndexSearcher的获取方法如下：

[java] view plain copy

1. /**多目录多线程查询 
2.      * @param parentPath 父级索引目录 
3.      * @param service 多线程查询 
4.      * @param isAddRamDirectory 是否增加内存索引查询 
5.      * @return 
6.      * @throws IOException 
7.      */  
8.     public static IndexSearcher getMultiSearcher(String parentPath,ExecutorService service, boolean isAddRamDirectory) throws IOException{  
9.         File file = new File(parentPath);  
10.         File[] files = file.listFiles();  
11.       
12.         IndexReader[] readers = null;  
13.         if(!isAddRamDirectory){  
14.             readers = new IndexReader[files.length];  
15.         }else{  
16.             readers = new IndexReader[files.length+1];  
17.         }  
18.         for (int i = 0 ; i < files.length ; i ++) {  
19.             readers[i] = DirectoryReader.open(FSDirectory.open(Paths.get(files[i].getPath(), new String[0])));  
20.         }  
21.         if(isAddRamDirectory){  
22.             readers[files.length] = DirectoryReader.open(RamDirectoryControl.getRAMDireactory());  
23.         }  
24.           
25.         MultiReader multiReader = new MultiReader(readers);  
26.         IndexSearcher searcher = new IndexSearcher(multiReader,service);  
27.         return searcher;  
28.     }