标签：code iter react auto pre anti ssi 对象 数据
这两天加班，不能兼顾博客的更新。请大家见谅。
有时候我们创建完索引之后，数据源可能有更新的内容。而我们又想像数据库那样能直接体如今查询中。这里就是我们所说的增量索引。对于这种需求我们怎么来实现呢？lucene内部是没有提供这种增量索引的实现的。
这里我们一般可能会想到。将之前的索引所有删除，然后进行索引的重建。
对于这样的做法。假设数据源的条数不是特别大的情况下倒还能够。假设数据源的条数特别大的话。势必会造成查询数据耗时。同一时候索引的构建也是比較耗时的，几相叠加，势必可能造成查询的时候数据缺失的情况，这势必严重影响用户的体验。
比較常见的增量索引的实现是：
这样做较以上的那种全删除索引然后重建的优点在于：
以上攻克了增量的问题，可是实时性的问题还是存在的：
那么我们如何对实时性有个提升呢，大家都知道lucene索引能够以文件索引和内存索引两种方式存在。相较于文件索引。内存索引的运行效率要高于文件索引的构建，由于文件索引是要频繁的IO操作的。结合以上的考虑，我们採用文件索引+内存索引的形式来进行lucene的增量更新；事实上现机制例如以下：
java内置了TimerTask。此类是能够提供定时任务的。可是有一点就是TimerTask的任务是无状态的。我们还须要对任务进行并行的设置；了解到quartz任务调度框架提供了有状态的任务StatefulJob。即在本次调度任务没有运行完成时，下次任务不会运行；
常见的我们启动一个quartz任务的方式例如以下：
Date runTime = DateBuilder.evenSecondDate(new Date()); StdSchedulerFactory sf = new StdSchedulerFactory(); Scheduler scheduler = sf.getScheduler(); JobDetail job = JobBuilder.newJob(XXX.class).build(); Trigger trigger = TriggerBuilder.newTrigger().startAt(runTime).withSchedule(SimpleScheduleBuilder.simpleSchedule().withIntervalInSeconds(3).repeatForever()).forJob(job).build(); scheduler.scheduleJob(job, trigger); scheduler.start();</span>
以上我们是设置了每三秒运行一次定时任务，而任务类是XXX
这里我定义了一个XXX的父类，其定义例如以下：
package com.chechong.lucene.indexcreasement;
import java.util.List;
import java.util.TimerTask;
import org.apache.lucene.store.RAMDirectory;
import org.quartz.Job;
import org.quartz.StatefulJob;
/**有状态的任务：串行运行。即不同意上次运行没有完毕即開始本次假设须要并行给接口改为Job就可以
* @author lenovo
*
*/
public abstract class BaseInCreasementIndex implements StatefulJob {
/**
* 内存索引
*/
private RAMDirectory ramDirectory;
public BaseInCreasementIndex() {
}
public BaseInCreasementIndex(RAMDirectory ramDirectory) {
super();
this.ramDirectory = ramDirectory;
}
/**更新索引
* @throws Exception
*/
public abstract void updateIndexData() throws Exception;
/**消费数据
* @param list
*/
public abstract void consume(List list) throws Exception;
}
任务类相关实现,下面方法是获取待加入索引的数据源XXXInCreasementIndex
@Override
public void execute(JobExecutionContext context) throws JobExecutionException {
try {
XXXInCreasementIndex index = new XXXInCreasementIndex(Constants.XXX_INDEX_PATH, XXXDao.getInstance(), RamDirectoryControl.getRAMDireactory());
index.updateIndexData();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
@Override
public void updateIndexData() throws Exception {
int maxBeanID = SearchUtil.getLastIndexBeanID();
System.out.println(maxBeanID);
List<XXX> sources = XXXDao.getListInfoBefore(maxBeanID);、、
if (sources != null && sources.size() > 0) {
this.consume(sources);
}
}
这里，XXX代表我们要获取数据的实体类对象
consume方法主要是做两件事：
@Override
public void consume(List list) throws Exception {
IndexWriter writer = RamDirectoryControl.getRAMIndexWriter();
RamDirectoryControl.consume(writer,list);
}
上边我们将内存索引和队列的实现放在了RamDirectoryControl中
首先我们对内存索引的IndexWriter进行初始化，在初始化的时候须要注意先运行一次commit，否则会提示no segments的异常
private static IndexWriter ramIndexWriter;
private static RAMDirectory directory;
static{
directory = new RAMDirectory();
try {
ramIndexWriter = getRAMIndexWriter();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
public static RAMDirectory getRAMDireactory(){
return directory;
}
public static IndexSearcher getIndexSearcher() throws IOException{
IndexReader reader = null;
IndexSearcher searcher = null;
try {
reader = DirectoryReader.open(directory);
} catch (IOException e) {
e.printStackTrace();
}
searcher = new IndexSearcher(reader);
return searcher;
}
/**单例模式获取ramIndexWriter
* @return
* @throws Exception
*/
public static IndexWriter getRAMIndexWriter() throws Exception{
if(ramIndexWriter == null){
synchronized (IndexWriter.class) {
Analyzer analyzer = new IKAnalyzer();
IndexWriterConfig iwConfig = new IndexWriterConfig(analyzer);
iwConfig.setOpenMode(OpenMode.CREATE_OR_APPEND);
try {
ramIndexWriter = new IndexWriter(directory, iwConfig);
ramIndexWriter.commit();
ramIndexWriter.close();
iwConfig = new IndexWriterConfig(analyzer);
iwConfig.setOpenMode(OpenMode.CREATE_OR_APPEND);
ramIndexWriter = new IndexWriter(directory, iwConfig);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
return ramIndexWriter;
}定义一个获取内存索引中数据条数的方法
/**依据查询器、查询条件、每页数、排序条件进行查询
* @param query 查询条件
* @param first 起始值
* @param max 最大值
* @param sort 排序条件
* @return
*/
public static TopDocs getScoreDocsByPerPageAndSortField(IndexSearcher searcher,Query query, int first,int max, Sort sort){
try {
if(query == null){
System.out.println(" Query is null return null ");
return null;
}
TopFieldCollector collector = null;
if(sort != null){
collector = TopFieldCollector.create(sort, first+max, false, false, false);
}else{
SortField[] sortField = new SortField[1];
sortField[0] = new SortField("createTime",SortField.Type.STRING,true);
Sort defaultSort = new Sort(sortField);
collector = TopFieldCollector.create(defaultSort,first+max, false, false, false);
}
searcher.search(query, collector);
return collector.topDocs(first, max);
} catch (IOException e) {
// TODO Auto-generated catch block
}
return null;
}
此方法返回结果为TopDocs。我们依据TopDocs的totalHits来获取内存索引中的数据条数，以此来鉴别内存占用，防止内存溢出。
consume方法的实现例如以下：
/**消费数据
* @param docs
* @param listSize
* @param writer
* @param list
* @throws Exception
*/
public static void consume(IndexWriter writer, List list) throws Exception {
Query query = new MatchAllDocsQuery();
IndexSearcher searcher = getIndexSearcher();
System.out.println(directory);
TopDocs topDocs = getScoreDocsByPerPageAndSortField(searcher,query, 1, 1, null);
int currentTotal = topDocs.totalHits;
if(currentTotal+list.size() > Constants.XXX_RAM_LIMIT){
//超出内存限制
int pulCount = Constants.XXX_RAM_LIMIT - currentTotal;
List<Document> docs = new LinkedList<Document>();
if(pulCount <= 0){
//直接处理集合的内容
TopDocs allDocs = SearchUtil.getScoreDocsByPerPageAndSortField(searcher, query, 0,currentTotal, null);
ScoreDoc[] scores = allDocs.scoreDocs;
for(int i = 0 ;i < scores.length ; i ++){
//取出内存中的数据
Document doc1 = searcher.doc(scores[i].doc);
Integer pollId = Integer.parseInt(doc1.get("id"));
Document doc = delDocumentFromRAMDirectory(pollId);
if(doc != null){
XXX carSource = (XXX) BeanTransferUtil.doc2Bean(doc, XXX.class);
Document doc2 = carSource2Document(carSource);
if(doc2 != null){
docs.add(doc2);
}
}
}
addDocumentToFSDirectory(docs);
writer = getRAMIndexWriter();
consume(writer, list);
}else{
//先取出未达到内存的部分
List subProcessList = list.subList(0, pulCount);
consume(writer, subProcessList);
List leaveList = list.subList(pulCount, list.size());
consume(writer, leaveList);
}
}else{//未超出限制。直接存放到内存
int listSize = list.size();
if(listSize > 0){
//存放到内存
}
}
}
上边的逻辑为：
这里我们的BeanTransferUtil是依据document转换成相应的bean的方法。此处用到了反射和commons-beanutils.jar
package com.chechong.util;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.lucene.document.Document;
public class BeanTransferUtil {
public static Object doc2Bean(Document doc, Class clazz) {
try {
Object obj = clazz.newInstance();
Field[] fields = clazz.getDeclaredFields();
for (Field field : fields) {
field.setAccessible(true);
String fieldName = field.getName();
BeanUtils.setProperty(obj, fieldName, doc.get(fieldName));
}
return obj;
} catch (InstantiationException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (IllegalAccessException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InvocationTargetException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return null;
}
}
从内存索引中读取索引的方法例如以下：
/**从内存索引中删除指定的doc
* @param pollId
* @throws IOException
*/
private static Document delDocumentFromRAMDirectory(Integer pollId) throws IOException {
Document doc = null;
Query query = SearchUtil.getQuery("id", "int", pollId+"", false);
IndexSearcher searcher = getIndexSearcher();
try {
TopDocs queryDoc = SearchUtil.getScoreDocsByPerPageAndSortField(searcher, query, 0, 1, null);
ScoreDoc[] docs = queryDoc.scoreDocs;
System.out.println(docs.length);
if(docs.length > 0){
doc = searcher.doc(docs[0].doc);
System.out.println(doc);
ramIndexWriter.deleteDocuments(query);
ramIndexWriter.commit();
}
return doc;
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return null;
}
此处是依据id来读取内存索引中的内容。然后将它转换成document同一时候删除内存中的相应记录。
对于上边的索引我们要採用适当的查询方法。这里查询时候为了达到近实时的效果。须要将内存索引加入到查询的范围中，即IndexReader中。
这里的IndexSearcher的获取方法例如以下：
/**多文件夹多线程查询
* @param parentPath 父级索引文件夹
* @param service 多线程查询
* @param isAddRamDirectory 是否添加内存索引查询
* @return
* @throws IOException
*/
public static IndexSearcher getMultiSearcher(String parentPath,ExecutorService service, boolean isAddRamDirectory) throws IOException{
File file = new File(parentPath);
File[] files = file.listFiles();
IndexReader[] readers = null;
if(!isAddRamDirectory){
readers = new IndexReader[files.length];
}else{
readers = new IndexReader[files.length+1];
}
for (int i = 0 ; i < files.length ; i ++) {
readers[i] = DirectoryReader.open(FSDirectory.open(Paths.get(files[i].getPath(), new String[0])));
}
if(isAddRamDirectory){
readers[files.length] = DirectoryReader.open(RamDirectoryControl.getRAMDireactory());
}
MultiReader multiReader = new MultiReader(readers);
IndexSearcher searcher = new IndexSearcher(multiReader,service);
return searcher;
}
如此。我们就能够在查询的时候既从文件索引中读取。也从内存索引中检索数据了。
一步一步跟我学习lucene是对最近做lucene索引的总结。大家有问题的话联系本人的Q-Q: 891922381，同一时候本人新建Q-Q群：106570134（lucene,solr,netty,hadoop）大家共同探讨,本人争取每日一博。希望大家持续关注。会带给大家惊喜的
一步一步跟我学习lucene（19）---lucene增量更新和NRT(near-real-time)Query近实时查询
标签：code iter react auto pre anti ssi 对象 数据
原文地址：http://www.cnblogs.com/liguangsunls/p/7136590.html
