从上文中的QueryTraverser对象的BatchResult runBatch(BatchSize batchSize)方法上溯到CancelableBatch类，该类实现了TimedCancelable接口，后者又extends了TimedCancelable接口，后者又extends了Cancelable接口，后者又extends了Runnable接口
Cancelable接口源码
/*** A {@link Runnable} that supports cancellation.*/ public interface Cancelable extends Runnable {/*** Cancel the operation performed by this {@link Runnable}.* While this {@link Runnable#run} method is running in one thread this* may be called in another so implementors must provide any needed* synchronization.*/public void cancel(); }
TimedCancelable接口源码
/*** A {@link Runnable} that supports cancellation and timeout.*/ public interface TimedCancelable extends Cancelable {/*** Complete the operations performed by this {@link Runnable} due to the* expiration of its time interval. While this {@link Runnable#run} method is* running in one thread this may be called in another so implementors must* provide any needed synchronization.*/public void timeout(TaskHandle taskHandle); }
下面才是CancelableBatch类，实现了TimedCancelable接口
/*** A {@link TimedCancelable} for running a {@link Connector} batch using* a {@link Traverser}*/ class CancelableBatch implements TimedCancelable {private static final Logger LOGGER =Logger.getLogger(CancelableBatch.class.getName());final Traverser traverser;final String traverserName;final BatchResultRecorder batchResultRecorder;final BatchTimeout batchTimeout;final BatchSize batchSize;/*** Construct a {@link CancelableBatch}.** @param traverser {@link Traverser} for running the batch.* @param traverserName traverser name for logging purposes.* @param batchResultRecorder {@link BatchResultRecorder} for recording* the result of running the batch.* @param batchSize hint and constraints as to the number of documents* to process in the batch.*/public CancelableBatch(Traverser traverser, String traverserName,BatchResultRecorder batchResultRecorder, BatchTimeout batchTimeout,BatchSize batchSize) {this.traverser = traverser;this.traverserName = traverserName;this.batchResultRecorder = batchResultRecorder;this.batchSize = batchSize;this.batchTimeout = batchTimeout;}/*** 取消执行*/public void cancel() {traverser.cancelBatch();}/*** 运行超时*/public void timeout(TaskHandle taskHandle) {batchTimeout.timeout();}public void run() {NDC.push("Traverse " + traverserName);try {LOGGER.fine("Begin runBatch; traverserName = " + traverserName+ " " + batchSize);BatchResult batchResult = traverser.runBatch(batchSize);LOGGER.fine("Traverser " + traverserName + " batchDone with result = "+ batchResult);batchResultRecorder.recordResult(batchResult);} finally {NDC.remove();}}@Overridepublic String toString() {return "CancelableBatch: traverser = " + traverser+ ", batchSize = " + batchSize;} }
在上面的run方法里面，调用了BatchResult batchResult = traverser.runBatch(batchSize);方法，用于获取数据源数据并向服务器推送数据
另外两方法注意下，后面会用到
/*** 取消执行*/public void cancel() {traverser.cancelBatch();}/*** 运行超时*/public void timeout(TaskHandle taskHandle) {batchTimeout.timeout();}
综上所述，CancelableBatch是一个实现了Runnable接口的线程类，姑且如是说
继续上溯到ConnectorCoordinatorImpl类，该类实现了ConnectorCoordinator接口，该接口声明了一个startBatch()方法
/*** Starts running a batch for this {@link ConnectorCoordinator} if a batch is* not already running.** @return true if this call started a batch* @throws ConnectorNotFoundException if this {@link ConnectorCoordinator}* does not exist.*/public boolean startBatch() throws ConnectorNotFoundException;
首先需要明白的是，一个连接器实例对应一个ConnectorCoordinatorImpl实例对象，ConnectorCoordinatorImpl类实在庞大，我们先分析startBatch()方法源码
/*** 开始采集* Starts running a batch for this {@link ConnectorCoordinator} if a batch is* not already running.** @return true if this call started a batch*//* @Override */public synchronized boolean startBatch() {if (!shouldRun()) {return false;}BatchSize batchSize = loadManager.determineBatchSize();if (batchSize.getHint() == 0) {return false;}try {TraversalManager traversalManager = getTraversalManager();if (traversalManager == null) {return false;}//当前标识currentBatchKey = new Object();BatchCoordinator batchCoordinator = new BatchCoordinator(this);//batchCoordinator 作为 TraversalStateStore stateStore角色Traverser traverser = new QueryTraverser(pusherFactory,traversalManager, batchCoordinator, name,Context.getInstance().getTraversalContext(), clock);//batchCoordinator 作为 BatchResultRecorder batchResultRecorder, BatchTimeout batchTimeout角色//调用Traverser traverser的取消方法 //BatchResultRecorder batchResultRecorder记录运行结果；[不能由外部调用]//BatchTimeout batchTimeout的超时方法TimedCancelable batch = new CancelableBatch(traverser, name,batchCoordinator, batchCoordinator, batchSize);taskHandle = threadPool.submit(batch);//threadPool.shutdown(interrupt, waitMillis)//taskHandle.cancel();return true;} catch (ConnectorNotFoundException cnfe) {LOGGER.log(Level.WARNING, "Connector not found - this is normal if you "+ " recently reconfigured your connector instance: " + cnfe);} catch (InstantiatorException ie) {LOGGER.log(Level.WARNING,"Failed to perform connector content traversal.", ie);delayTraversal(TraversalDelayPolicy.ERROR);}return false;}
我们可以看到，在该方法里面首先构造QueryTraverser对象(需要构造引用PusherFactory pusherFactory、TraversalManager traversalManager、TraversalStateStore stateStore实例)，然后构造CancelableBatch对象（构造函数传入QueryTraverser对象和BatchSize batchSize对象*批次尺寸），最后线程池对象提交CancelableBatch对象执行（到现在我们知道，一次线程执行只执行批次尺寸的数据采集，而并不一定是全部数据）
这里的BatchCoordinator batchCoordinator = new BatchCoordinator(this)对象在上面方法中充当不同的角色，即该对象实现了不同的接口
其构造函数传入了当前对象，即ConnectorCoordinatorImpl connectorCoordinator实例对象
在BatchCoordinator batchCoordinator对象实现不同接口的实现方法里面，基本上都是回调ConnectorCoordinatorImpl connectorCoordinator实例对象的方法，采用这种迂回的回马枪策略，大概是出于职责分明考虑吧，另外可能需要用到ConnectorCoordinatorImpl connectorCoordinator实例对象的状态
基本上BatchCoordinator batchCoordinator对象实现的方法在ConnectorCoordinatorImpl connectorCoordinator实例对象里面都要实现，这里设计方法采用的是一种包装器模式、或者是代理模式
可以猜想到，BatchCoordinator batchCoordinator对象实现了的接口实际上ConnectorCoordinatorImpl connectorCoordinator实例对象名义上甚至实际上也同样实现了（而类ConnectorCoordinatorImpl 实现的接口BatchCoordinator并不一定要实现）
我们先睹为快，不先做瞎猜了
BatchCoordinator类实现
class BatchCoordinator implements TraversalStateStore, BatchResultRecorder, BatchTimeout
ConnectorCoordinatorImpl类实现
class ConnectorCoordinatorImpl implementsConnectorCoordinator, ChangeHandler, BatchResultRecorder
貌似BatchCoordinator类实现的接口之中，只有BatchResultRecorder接口ConnectorCoordinatorImpl类名义上也实现了
下面逐一分析
BatchCoordinator batchCoordinator = new BatchCoordinator(this);//batchCoordinator 作为 TraversalStateStore stateStore角色Traverser traverser = new QueryTraverser(pusherFactory,traversalManager, batchCoordinator, name,Context.getInstance().getTraversalContext(), clock);
这里batchCoordinator 作为 TraversalStateStore stateStore角色，实现raversalStateStore接口方法为：
public String getTraversalState() {synchronized (connectorCoordinator) {if (connectorCoordinator.currentBatchKey == requiredBatchKey) {return cachedState;} else {throw new BatchCompletedException();}}}public void storeTraversalState(String state) {synchronized (connectorCoordinator) {// Make sure our batch is still valid and that nobody has modified// the checkpoint while we were away.try {if ((connectorCoordinator.currentBatchKey == requiredBatchKey) &&isCheckpointUnmodified()) {connectorCoordinator.setConnectorState(state);cachedState = state;} else {throw new BatchCompletedException();}} catch (ConnectorNotFoundException cnfe) {// Connector disappeared while we were away.// Don't try to store results.throw new BatchCompletedException();}}}
上面方法分别为获取断点状态和更新断点状态，需要考虑同步问题，两者都回调了ConnectorCoordinatorImpl connectorCoordinator对象的方法
获取状态
/*** Returns the Connector's traversal state.** @return String representation of the stored state, or* null if no state is stored.* @throws ConnectorNotFoundException if this {@link ConnectorCoordinator}* does not exist.*//* @Override */public synchronized String getConnectorState()throws ConnectorNotFoundException {return getInstanceInfo().getConnectorState();}
更新状态
/*** Set the Connector's traversal state.** @param state a String representation of the state to store.* If null, any previous stored state is discarded.* @throws ConnectorNotFoundException if this {@link ConnectorCoordinator}* does not exist.*//* @Override */public synchronized void setConnectorState(String state)throws ConnectorNotFoundException {getInstanceInfo().setConnectorState(state);// Must not call ChangeDetector, as this is called from a synchronized// block in BatchCoordinator.}
接下来分析BatchCoordinator batchCoordinator对象充当的其他角色
//batchCoordinator 作为 BatchResultRecorder batchResultRecorder, BatchTimeout batchTimeout角色//调用Traverser traverser的取消方法 //BatchResultRecorder batchResultRecorder记录运行结果；[不能由外部调用]//BatchTimeout batchTimeout的超时方法TimedCancelable batch = new CancelableBatch(traverser, name,batchCoordinator, batchCoordinator, batchSize);
前者作为BatchResultRecorder batchResultRecorder角色，后者作为BatchTimeout batchTimeout角色
实现BatchResultRecorder接口的方法为
public void recordResult(BatchResult result) {synchronized (connectorCoordinator) {if (connectorCoordinator.currentBatchKey == requiredBatchKey) {connectorCoordinator.recordResult(result);} else {LOGGER.fine("Ignoring a BatchResult returned from a "+ "prevously canceled traversal batch. Connector = "+ connectorCoordinator.getConnectorName()+ " result = " + result + " batchKey = " + requiredBatchKey);}}}
里面进一步回调了ConnectorCoordinatorImpl connectorCoordinator对象如下方法
/*** Records the supplied traversal batch results. Updates the* {@link LoadManager} with number of documents traversed,* and implements the requested {@link TraversalDelayPolicy}.** @param result a BatchResult*//* @Override */public synchronized void recordResult(BatchResult result) {loadManager.recordResult(result);delayTraversal(result.getDelayPolicy());}
记录执行结果及决定延迟策略
这里调用的方法名一致，我们再前面已经看到，两者都实现了BatchResultRecorder接口
实现BatchTimeout接口方法为
public void timeout() {synchronized (connectorCoordinator) {if (connectorCoordinator.currentBatchKey == requiredBatchKey) {connectorCoordinator.resetBatch();} else {LOGGER.warning("Ignoring Timeout for previously prevously canceled"+ " or completed traversal batch. Connector = "+ connectorCoordinator.getConnectorName()+ " batchKey = "+ requiredBatchKey);}}}
回调ConnectorCoordinatorImpl connectorCoordinator对象如下方法（重置采集）
/*** 取消采集* Halts any in-progess traversals for this {@link Connector} instance.* Some or all of the information collected during the current traversal* may be discarded.*/synchronized void resetBatch() {if (taskHandle != null) {taskHandle.cancel();}taskHandle = null;currentBatchKey = null;interfaces = null;// Discard cached interface instances.traversalManager = null;retriever = null;traversalSchedule = null;}
前两者分别用于状态管理和记录执行结果及延迟策略，第三者用于取消采集（里面调用了taskHandle.cancel()方法）
执行序列为
TimedCancelable类型对象（CancelableBatch对象）的timeout(TaskHandle taskHandle)方法-->
BatchTimeout类型对象(即BatchCoordinator batchCoordinator)的batchTimeout.timeout()方法-->
ConnectorCoordinatorImpl connectorCoordinator对象的resetBatch()方法-->
TaskHandle taskHandle的taskHandle.cancel()方法-->Cancelable类型对象（CancelableBatch对象）的cancel()方法-->Traverser traverser的 cancelBatch()方法
即CancelableBatch对象的timeout(TaskHandle taskHandle)方法绕来绕去最终接上了自己的cancel()方法
从下文中我们可以看到，这样处理的目的在于当一个线程超时时，由另一个监控超时的线程执行取消操作；在正常情况下，该执行序列将不会发生
TaskHandle taskHandle是一个任务执行句柄，用于对执行过程进行操控
/*** Handle for the management of a {@link Cancelable} primary task.*/ public class TaskHandle {/*** The primary {@link Cancelable} that is run by this task to* perform some useful work.*/final Cancelable cancelable;/** The {@link future} for the primary task.*/final Future<?> taskFuture;/** The time the task starts.*/final long startTime;/*** Create a TaskHandle.** @param cancelable {@link Cancelable} for the primary task.* @param taskFuture {@link Future} for the primary task.* @param startTime startTime for the primary task.*/TaskHandle(Cancelable cancelable, Future<?> taskFuture, long startTime) {this.cancelable = cancelable;this.taskFuture = taskFuture;this.startTime = startTime;}/*** Cancel the primary task and the time out task.*/public void cancel() {cancelable.cancel();taskFuture.cancel(true);}/*** Return true if the primary task has completed.*/public boolean isDone() {return taskFuture.isDone();} }
---------------------------------------------------------------------------
本系列企业搜索引擎开发之连接器connector系本人原创
转载请注明出处 博客园 刺猬的温驯
本人邮箱： chenying998179@163#com （#改为.）
本文链接 http://www.cnblogs.com/chenying99/p/3775591.html