最复杂的getSearcher逻辑回顾
//注意 getSearcher部分理解务必结合原代码头部的E文注释
//注意 使用了单线程的ExecutorService 它会排队new SolrIndexSearcherJob、job之间又依赖对象lock同步。 另外，整个SolrIndexSearcher 正常情况是共享的，只是计数增加，在commit之后或者new core 或者new SolrIndexSearcher才会新打开。新打开之后curr的就改变了。已经打开的多个SolrIndexSearcher保存在 _searchers 队列里面，get是从last获取。
//每一次新打开一个SolrIndexSearcher 里面对应SolrIndexReader，这个Reader来自ReaderFactory，不管如何改写ReaderFactory，里面最终newReader 需要注入一个SolrIndexReader，这个SolrIndexReader可能是multiIndexReader，或者自定义的Reader对象。
//SolrIndexSearcher返回默认是整个core下的索引视图，由SolrIndexReader来保证视图的最新。每次commit会重新newReader 从而SolrIndexSearcher。
//getSearcher流程大体可以这样理解：（不知道如何将代码流程文字化出来，有好的idea 留言交流）
(1)优先从当前_searcher 获取SolrIndexSearcher，本质是计数增加，而没有新建，因为readeronly，从而无线程安全问题。
(2)之后从_searchers中获取已经注册的SolrIndexSearcher，并赋给_searcher 返回
(3)每次 new SolrIndexSearcher都会注册起来，加入_searchers 中
* Get a SolrIndexSearcher} or start the process of creating a new one.
* The registered searcher is the default searcher used to service queries.
* A searcher will normally be registered after all of the warming
and event handlers (newSearcher or firstSearcher events) have run.
* In the case where there is no registered searcher, the newly created searcher willbe registered before running the event handlers (a slow searcher is better than no searcher).
* These searchers contain read-only IndexReaders. To access a non read-only IndexReader, see newSearcher(String name, boolean readOnly).
* If forceNew==true then
A new searcher will be opened and registered regardless of whether there is already a registered searcher or other searchers in the process of being created.
* If forceNew==false then:
If a searcher is already registered, that searcher will be returned
If no searcher is currently registered, but at least one is in the process of being created, then this call will block until the first searcher is registered
If no searcher is currently registered, and no searchers in the process of being registered, a new searcher will be created.
* If returnSearcher==true then a RefCounted; will be returned with the reference count incremented. It must be decremented when no longer needed.
* If waitSearcher!=null and a new SolrIndexSearcher was created,
then it is filled in with a Future that will return after the searcher is registered. The Future may be set to null in which case the SolrIndexSearcher created has already been registered at the time
this method returned.
* @param waitSearcher if non-null, will be filled in with a {@link Future}that will return after the new searcher is registered.
public RefCounted getSearcher(boolean forceNew, boolean returnSearcher, final Future[] waitSearcher) throwsIOException {
// it may take some time to open an index.... we may need to make
// sure that two threads aren't trying to open one at the same time
// if it isn't necessary. 如此存在直接返回，
synchronized(searcherLock) {
// see if we can return the current searcher
if(_searcher!=null && !forceNew) {
if (returnSearcher) {//不返回returnSearcher，加载预热不返回，从solrCore提供出去的都是返回的
_searcher.incref();
return _searcher;
} else {
return null;
}
}
// check to see if we can wait for someone else's searcher to be set
此时_searcher 为null，或者forceNew=true
if(onDeckSearchers>0 && !forceNew && _searcher==null) {
try {
searcherLock.wait();
} catch (InterruptedException e) {
log.info(SolrException.toStr(e));
}
}
// check again: see if we can return right now
if(_searcher!=null && !forceNew) {
if (returnSearcher) {
_searcher.incref();
return _searcher;// 当前的Search
} else {
return null;
}
}
// At this point, we know we need to open a new searcher...
// first: increment count to signal other threads that we are
// opening a new searcher.
onDeckSearchers++;
if(onDeckSearchers< 1) {
// should never happen... just a sanity check
log.error(logid+"ERROR!!! onDeckSearchers is " + onDeckSearchers);
onDeckSearchers=1; // reset
} elseif(onDeckSearchers> maxWarmingSearchers) {
onDeckSearchers--;
String msg="Error opening new searcher. exceeded limit of maxWarmingSearchers="+maxWarmingSearchers+ ", try again later.";
log.warn(logid+""+ msg);
// HTTP 503==service unavailable, or 409==Conflict
throw newSolrException(SolrException.ErrorCode.SERVICE_UNAVAILABLE,msg,true);
} elseif(onDeckSearchers> 1) {
log.info(logid+"PERFORMANCE WARNING: Overlapping onDeckSearchers="+onDeckSearchers);
}
}//同步块结束，先写计数，后面新建
// open the index synchronously
// if this fails, we need to decrement onDeckSearchers again.
SolrIndexSearcher tmp;
RefCounted newestSearcher = null;
try{
newestSearcher = getNewestSearcher(false);//从已经打开的里面返回
String newIndexDir = getNewIndexDir();
File indexDirFile = new File(getIndexDir()).getCanonicalFile();
File newIndexDirFile = new File(newIndexDir).getCanonicalFile();
if(newestSearcher != null && solrConfig.reopenReaders
&& indexDirFile.equals(newIndexDirFile)) {
IndexReader currentReader = newestSearcher.get().getReader();
IndexReader newReader = currentReader.reopen();
if (newReader == currentReader) {
currentReader.incRef();
}
tmp = new SolrIndexSearcher(this, schema, "main", newReader, true, true);
} else{//每次Commit才会
//这里ReaderFactory是用户solrconfig.xml中配置的，对应newReader方法也改写了，每次传入的newIndexDir是相同的，在重定义的newReader里面需要依赖newIndexDir来解析新的路径并生成新的reader.返回的是一个reader，尽管里面有多个reader。
IndexReader reader = getIndexReaderFactory().newReader(getDirectoryFactory().open(newIndexDir), true);
tmp = new SolrIndexSearcher(this, schema, "main", reader, true, true);
}
} catch(Throwable th) {
synchronized(searcherLock) {//注意对onDeckSearchers 和searcherLock的异常处理
onDeckSearchers--;
// notify another waiter to continue... it may succeed
// and wake any others.
searcherLock.notify();
}
// need to close the searcher here??? we shouldn't have to.
thrownewRuntimeException(th);
} finally{
if(newestSearcher != null) {
newestSearcher.decref();
}
}
finalSolrIndexSearcher newSearcher=tmp;
RefCounted currSearcherHolder=null;
finalRefCounted newSearchHolder=newHolder(newSearcher);
if(returnSearcher) newSearchHolder.incref();
// a signal to decrement onDeckSearchers if something goes wrong.
finalboolean[] decrementOnDeckCount=new boolean[1];
decrementOnDeckCount[0]=true;
try{
boolean alreadyRegistered = false;
synchronized(searcherLock) {
_searchers.add(newSearchHolder);
if (_searcher == null) {
// if there isn't a current searcher then we may
// want to register this one before warming is complete instead of waiting.
if (solrConfig.useColdSearcher) {
registerSearcher(newSearchHolder);//并发hashmap保存活对象
decrementOnDeckCount[0]=false;
alreadyRegistered=true;
}
} else {
// get a reference to the current searcher for purposes of autowarming.
currSearcherHolder=_searcher;
currSearcherHolder.incref();
}
}
final SolrIndexSearcher currSearcher = currSearcherHolder==null ? null : currSearcherHolder.get();
// Note! if we registered the new searcher (but didn't increment it's
// reference count because returnSearcher==false, it's possible for
// someone else to register another searcher, and thus cause newSearcher
// to close while we are warming.
// Should we protect against that by incrementing the reference count?
// Maybe we should just let it fail? After all, if returnSearcher==false
// and newSearcher has been de-registered, what's the point of continuing?
Future future=null;
// warm the new searcher based on the current searcher.
// should this go before the other event handlers or after?
if(currSearcher != null) {
future = searcherExecutor.submit(
new Callable() {
publicObject call() throws Exception {
try {
newSearcher.warm(currSearcher);
} catch (Throwable e) {
SolrException.logOnce(log,null,e);
}
returnnull;
}
}
);
}
if(currSearcher==null&& firstSearcherListeners.size() > 0) {
future = searcherExecutor.submit(
new Callable() {
publicObject call() throws Exception {
try {
for(SolrEventListener listener : firstSearcherListeners) {
listener.newSearcher(newSearcher,null);
}
} catch (Throwable e) {
SolrException.logOnce(log,null,e);
}
returnnull;
}
}
);
}
if(currSearcher!=null&& newSearcherListeners.size() > 0) {
future = searcherExecutor.submit(
new Callable() {
publicObject call() throws Exception {
try {
for(SolrEventListener listener : newSearcherListeners) {
listener.newSearcher(newSearcher, currSearcher);
}
} catch (Throwable e) {
SolrException.logOnce(log,null,e);
}
returnnull;
}
}
);
}
// WARNING: this code assumes a single threaded executor (that all tasks
// queued will finish first).
finalRefCounted currSearcherHolderF = currSearcherHolder;
if(!alreadyRegistered) {
future = searcherExecutor.submit(
new Callable() {
publicObject call() throws Exception {
try {
// signal that we no longer need to decrement
// the count *before* registering the searcher since
// registerSearcher will decrement even if it errors.
decrementOnDeckCount[0]=false;
registerSearcher(newSearchHolder);
} catch (Throwable e) {
SolrException.logOnce(log,null,e);
} finally{
// we are all done with the old searcher we used
// for warming...
if (currSearcherHolderF!=null) currSearcherHolderF.decref();
}
returnnull;
}
}
);
}
if(waitSearcher != null) {
waitSearcher[0] = future;
}
// Return the searcher as the warming tasks run in parallel
// callers may wait on the waitSearcher future returned.
returnreturnSearcher ? newSearchHolder : null;
} catch(Exception e) {
SolrException.logOnce(log,null,e);
if(currSearcherHolder != null) currSearcherHolder.decref();
synchronized(searcherLock) {
if (decrementOnDeckCount[0]) {
onDeckSearchers--;
}
if (onDeckSearchers < 0) {
// sanity check... should never happen
log.error(logid+"ERROR!!! onDeckSearchers after decrement=" +onDeckSearchers);
onDeckSearchers=0; // try and recover
}
// if we failed, we need to wake up at least one waiter to continue the process
searcherLock.notify();
}
// since the indexreader was already opened, assume we can continue on
// even though we got an exception.
returnreturnSearcher ? newSearchHolder : null;
}
