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

我们注意到EntityProcessorWrapper的初始化方法

@Override
  public void init(Context context) {

    delegate.init(context);

  } 

这里context是上下文对象，通过context获取数据源，context是抽象类

public abstract class Context
{
public abstract DataSource getDataSource();
public abstract DataSource getDataSource(String name);
public abstract EntityProcessor getEntityProcessor();
public abstract boolean isRootEntity();

}  

继承类为ContextImpl，注意一下获取数据源的方法，调用DataImporter对象获取数据源（DataImporter 前文介绍过，参考solr dataimport 数据导入源码分析（二） ，通过getDataSourceInstance 方法实例化new JdbcDataSource() 对象）

ContextImpl.java 

 private DataSource ds;

  

private DataImporter dataImporter;

   

@Override

  public DataSource getDataSource() {

    if (ds != null) return ds;

    if(entity == null) return  null;

    if (entity.dataSrc == null) {

      entity.dataSrc = dataImporter.getDataSourceInstance(entity, entity.dataSource, this);

    }

    if (entity.dataSrc != null && docBuilder != null && docBuilder.verboseDebug &&

             Context.FULL_DUMP.equals(currentProcess())) {

      //debug is not yet implemented properly for deltas

      entity.dataSrc = docBuilder.getDebugLogger().wrapDs(entity.dataSrc);

    }

    return entity.dataSrc;

  }

再看DocBuidler是怎么初始化的EntityProcessorWrapper类的

DocBuidler.java 

private EntityProcessorWrapper getEntityProcessor(DataConfig.Entity entity) {
    if (entity.processor != null)
      return entity.processor;
    EntityProcessor entityProcessor = null;
    if (entity.proc == null) {
      entityProcessor = new SqlEntityProcessor();
    } else {
      try {
        entityProcessor = (EntityProcessor) loadClass(entity.proc, dataImporter.getCore())
                .newInstance();
      } catch (Exception e) {
        wrapAndThrow (SEVERE,e,
                "Unable to load EntityProcessor implementation for entity:" + entity.name);
      }
    }
    return entity.processor = new EntityProcessorWrapper(entityProcessor, this);

  } 

这里返回的是代理类，EntityProcessorWrapper，代理的对象默认是SqlEntityProcessor类

在 DocBuidler的buildDocument方法里面，构造并传入上下文对象，即ContextImpl对象（通过ContextImpl获取数据源）

 private void buildDocument(VariableResolverImpl vr, DocWrapper doc,

                             Map<String, Object> pk, DataConfig.Entity entity, boolean isRoot,

                             ContextImpl parentCtx, List<EntityProcessorWrapper> entitiesToDestroy) {



    EntityProcessorWrapper entityProcessor = getEntityProcessor(entity);



    ContextImpl ctx = new ContextImpl(entity, vr, null,


            pk == null ? Context.FULL_DUMP : Context.DELTA_DUMP,


            session, parentCtx, this);


    entityProcessor.init(ctx);



//其他代码略



}

在SqlEntityProcessor的初始化方法里面，调用ContextImpl 的DataSource getDataSource()的方法初始化数据源

 SqlEntityProcessor.java

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

  @Override
  @SuppressWarnings("unchecked")
  public void init(Context context) {
    super.init(context);
    dataSource = context.getDataSource();

  } 

数据查询方法如下，通过传入参数sql语句，初始化数据迭代器

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);
    }

  } 

SqlEntityProcessor 的全部导入和增量导入方法通过调用initQuery(String q) 获取数据迭代器

最后DocBuilder 类通过包装类EntityProcessorWrapper间接调用 SqlEntityProcessor的这些全部导入和增量导入方法获取数据构造SolrInputDocument对象