Solr4.8.0源码分析(14)之SolrCloud索引深入(1)

 1  <updateRequestProcessorChain name="key" default="true">
 2    <processor class="package.Class1" />
 3    <processor class="package.Class2" >
 4      <str name="someInitParam1">value</str>
 5      <int name="someInitParam2">42</int>
 6    </processor>
 7    <processor class="solr.LogUpdateProcessorFactory" >
 8      <int name="maxNumToLog">100</int>
 9    </processor>
10    <processor class="solr.RunUpdateProcessorFactory" />
11  </updateRequestProcessorChain>

 1  public void init(PluginInfo info) {
 2     final String infomsg = "updateRequestProcessorChain \"" + 
 3       (null != info.name ? info.name : "") + "\"" + 
 4       (info.isDefault() ? " (default)" : "");
 5 
 6     log.info("creating " + infomsg);
 7 
 8     // wrap in an ArrayList so we know we know we can do fast index lookups 
 9     // and that add(int,Object) is supported
10     //从solrcore获取索引链的成员，存放成列表形式。索引链在Solrconfig.xml中是以插件的形式加入的
11     List<UpdateRequestProcessorFactory> list = new ArrayList
12       (solrCore.initPlugins(info.getChildren("processor"),UpdateRequestProcessorFactory.class,null));
13 
14     if(list.isEmpty()){
15       throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
16                               infomsg + " require at least one processor");
17     }
18 
19     int numDistrib = 0;
20     int runIndex = -1;
21     // hi->lo incase multiple run instances, add before first one
22     // (no idea why someone might use multiple run instances, but just in case)
23     //从后往前遍历索引链列表，寻找DistributingUpdateProcessorFactory和RunUpdateProcessorFactory
24     for (int i = list.size()-1; 0 <= i; i--) {
25       UpdateRequestProcessorFactory factory = list.get(i);
26       if (factory instanceof DistributingUpdateProcessorFactory) {
27         numDistrib++; //DistributingUpdateProcessorFactory的个数不能超过1
28       }
29       if (factory instanceof RunUpdateProcessorFactory) {
30         runIndex = i; //RunUpdateProcessorFactory的编号
31       }
32     }
33     if (1 < numDistrib) {
34       throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
35                               infomsg + " may not contain more then one " +
36                               "instance of DistributingUpdateProcessorFactory");
37     }
38     //如果存在RunUpdateProcessorFactory且没有DistributingUpdateProcessorFactory，
39     //那么会在RunUpdateProcessorFactory之前加入DistributingUpdateProcessorFactory
40     if (0 <= runIndex && 0 == numDistrib) {
41       // by default, add distrib processor immediately before run
42       DistributedUpdateProcessorFactory distrib 
43         = new DistributedUpdateProcessorFactory();
44       distrib.init(new NamedList());
45       list.add(runIndex, distrib);
46 
47       log.info("inserting DistributedUpdateProcessorFactory into " + infomsg);
48     }
49 
50     chain = list.toArray(new UpdateRequestProcessorFactory[list.size()]); 
51   }

 1 public UpdateRequestProcessor createProcessor(SolrQueryRequest req, 
 2                                                 SolrQueryResponse rsp) 
 3   {
 4     UpdateRequestProcessor processor = null;
 5     UpdateRequestProcessor last = null;
 6     //获取distribPhase 是否需要跳过DistributingUpdateProcessorFactory前面那一过程（即LogUpdateProcessor），该参数不是由客户端生成
 7     final String distribPhase = req.getParams().get(DistributingUpdateProcessorFactory.DISTRIB_UPDATE_PARAM);
 8     final boolean skipToDistrib = distribPhase != null;
 9     boolean afterDistrib = true;  // we iterate backwards, so true to start
10     //从后往前组件索引链即
11     //LogUpdateProcessor.next = DistributedUpdateProcessor
12     //DistributedUpdateProcessor.next = DirectUpdateHandler2
13     //DirectUpdateHandler2.next = null
14     for (int i = chain.length-1; i>=0; i--) {
15       UpdateRequestProcessorFactory factory = chain[i];
16 
17       if (skipToDistrib) {
18         if (afterDistrib) {
19           // 跳过DistributingUpdateProcessorFactory前面的索引链过程
20           if (factory instanceof DistributingUpdateProcessorFactory) {
21             afterDistrib = false;
22           }
23         } else if (!(factory instanceof UpdateRequestProcessorFactory.RunAlways)) {
24           // skip anything that doesn‘t have the marker interface
25           continue;
26         }
27       }
28       //创建UpdateRequestProcessorFactory对应的UpdateProcessor，并进行连接
29       processor = factory.getInstance(req, rsp, last);
30       last = processor == null ? last : processor;
31     }
32 
33     return last;
34   }

1   public UpdateRequestProcessor getInstance(SolrQueryRequest req, SolrQueryResponse rsp, UpdateRequestProcessor next) {
2     return LogUpdateProcessor.log.isInfoEnabled() ? new LogUpdateProcessor(req, rsp, this, next) : null;
3   }

1   public DistributedUpdateProcessor getInstance(SolrQueryRequest req,
2       SolrQueryResponse rsp, UpdateRequestProcessor next) {
3 
4     return new DistributedUpdateProcessor(req, rsp, next);
5   }

1   public UpdateRequestProcessor getInstance(SolrQueryRequest req, SolrQueryResponse rsp, UpdateRequestProcessor next) 
2   {
3     return new RunUpdateProcessor(req, next);
4   }