solr dataimport 数据导入源码分析（三）

在介绍DocBuilder 类之前，我们先来解读数据导入对应实体处理器EntityProcessor，默认的实体处理器为SqlEntityProcessor

EntityProcessor 为抽象类，具体方法由子类实现

package org.apache.solr.handler.dataimport;

import java.util.Map;

public abstract class EntityProcessor {

public abstract void init(Context context);

public abstract Map<String, Object> nextRow();

public abstract Map<String, Object> nextModifiedRowKey();

public abstract Map<String, Object> nextDeletedRowKey();

public abstract Map<String, Object> nextModifiedParentRowKey();

public abstract void destroy();

public void postTransform(Map<String, Object> r) {

}

public void close() {

//no-op

}

EntityProcessorBase继承类封装公用逻辑，其中比较重要的是getNext()方法，用于遍历数据迭代器，供子类调用

protected Iterator<Map<String, Object>> rowIterator;

protected DIHCacheSupport cacheSupport = null;
protected Map<String, Object> getNext() {

if(cacheSupport==null) {

try {

if (rowIterator == null)

return null;

if (rowIterator.hasNext())

return rowIterator.next();

query = null;

rowIterator = null;

return null;

} catch (Exception e) {

SolrException.log(log, "getNext() failed for query '" + query + "'", e);

query = null;

rowIterator = null;

wrapAndThrow(DataImportHandlerException.WARN, e);

return null;

}

} else {

return cacheSupport.getCacheData(context, query, rowIterator);

}

}

真正的数据处理器为SqlEntityProcessor,简要代码如下

package org.apache.solr.handler.dataimport;

import java.util.Iterator;

import java.util.Map;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import java.util.regex.Matcher;

import java.util.regex.Pattern;

public class SqlEntityProcessor extends EntityProcessorBase {

protected DataSource<Iterator<Map<String, Object>>> dataSource;

@Override

@SuppressWarnings("unchecked")

public void init(Context context) {

super.init(context);

dataSource = context.getDataSource();

}

protected void initQuery(String q) {

try {

DataImporter.QUERY_COUNT.get().incrementAndGet();

rowIterator = dataSource.getData(q);

this.query = q;

} catch (DataImportHandlerException e) {

throw e;

} catch (Exception e) {

LOG.error( "The query failed '" + q + "'", e);

throw new DataImportHandlerException(DataImportHandlerException.SEVERE, e);

}

}

@Override

public Map<String, Object> nextRow() {

if (rowIterator == null) {

String q = getQuery();

initQuery(context.replaceTokens(q));

}

return getNext();

}

@Override

public Map<String, Object> nextModifiedRowKey() {

if (rowIterator == null) {

String deltaQuery = context.getEntityAttribute(DELTA_QUERY);

if (deltaQuery == null)

return null;

initQuery(context.replaceTokens(deltaQuery));

}

return getNext();

}

@Override

public Map<String, Object> nextDeletedRowKey() {

if (rowIterator == null) {

String deletedPkQuery = context.getEntityAttribute(DEL_PK_QUERY);

if (deletedPkQuery == null)

return null;

initQuery(context.replaceTokens(deletedPkQuery));

}

return getNext();

}

@Override

public Map<String, Object> nextModifiedParentRowKey() {

if (rowIterator == null) {

String parentDeltaQuery = context.getEntityAttribute(PARENT_DELTA_QUERY);

if (parentDeltaQuery == null)

return null;

LOG.info("Running parentDeltaQuery for Entity: "

+ context.getEntityAttribute("name"));

initQuery(context.replaceTokens(parentDeltaQuery));

}

return getNext();

}

}