详解 Elasticsearch refresh 机制

你的名字 2022-11-21 04:27 787阅读 0赞

## 概览 ##

`Refresh` 是Elasticsearch的运行机制的一个环节，保证了索引信息的增删改变化能够被及时感知检索，本篇博客我们将深入理解refresh的配置，运行机制，执行逻辑等等内容。

Elasticsearch 将 refresh 指令分发到所有的 Index Shard 不是本篇博客的关注内容，我们主要关注的是每个Index Shard 上的refresh的执行逻辑。

## refresh 配置 ##

通过在index settings 里配置 `index.refresh_interval` 来指定refresh间隔

{ 
      "my_index" : { 
        "settings" : { 
          "index" : { 
            "refresh_interval" : "2s"
          }
        }
      }
    }

refresh 间隔可以设置`秒，分，小时，天` 等等，设置方式参考`java.util.concurrent.TimeUnit`，当 `index.refresh_interval=-1` 时关闭refresh功能。

## refresh 在ES层执行逻辑 ##

### IndexService\#new ###

每个Index的分片在实例化时会创建IndexService，专门负责处理索引的各种问题，包括增删改查，分片setting是，mappings更新，refresh等等，其在实例化时会创建一个`AsyncRefreshTask` 用于负责索引分片的refresh功能

public class IndexService extends AbstractIndexComponent implements IndicesClusterStateService.AllocatedIndex<IndexShard> { 
    	//......
    	private volatile AsyncRefreshTask refreshTask;
    
    	public IndexService(......){ 
    		// ......
            this.refreshTask = new AsyncRefreshTask(this);
            // ......
    	}
    }

### AsyncRefreshTask\#new ###

`AsyncRefreshTask` 是负责refresh 索引的任务类，其在实例化时就在构造函数内执行了一些列的出发逻辑。

/** * 异步refresh任务 */
    final class AsyncRefreshTask extends BaseAsyncTask { 
    
        AsyncRefreshTask(IndexService indexService) { 
            // 索引service, refresh间隔
            super(indexService, indexService.getIndexSettings().getRefreshInterval());
        }
    
        @Override
        protected void runInternal() { 
            // 执行 refresh
            indexService.maybeRefreshEngine(false);
        }
    
        /** * 指定线程池 * * @return */
        @Override
        protected String getThreadPool() { 
            return ThreadPool.Names.REFRESH;
        }
    }

### BaseAsyncTask\#new ###

`AsyncRefreshTask` 继承 自 `BaseAsyncTask` 其在构造函数内就执行了 `rescheduleIfNecessary（）`

abstract static class BaseAsyncTask extends AbstractAsyncTask { 
    
        protected final IndexService indexService;
    
        BaseAsyncTask(final IndexService indexService, final TimeValue interval) { 
            super(indexService.logger, indexService.threadPool, interval, true);
            this.indexService = indexService;
            // 重新调度当前任务
            rescheduleIfNecessary();
        }
        //......
    }

### AbstractAsyncTask\#rescheduleIfNecessary ###

`BaseAsyncTask` 在构造函数内就执行 `rescheduleIfNecessary（）`，这方法是其父类 `AbstractAsyncTask` 定义的函数

public abstract class AbstractAsyncTask implements Runnable, Closeable { 
    	// refresh 线程池
        private final ThreadPool threadPool;
    	// index.refresh_interva 参数 指定的refresh间隔
        private volatile TimeValue interval;
    
        protected AbstractAsyncTask(Logger logger, ThreadPool threadPool, TimeValue interval, boolean autoReschedule) { 
            // ......
            this.threadPool = threadPool;
            this.interval = interval;
        }
    
        public synchronized void rescheduleIfNecessary() { 
           // ......
           if (interval.millis() > 0 && mustReschedule()) { 
               if (logger.isTraceEnabled()) { 
                   logger.trace("scheduling {} every {}", toString(), interval);
               }
               // 调度当前任务
               cancellable = threadPool.schedule(this, interval, getThreadPool());
               isScheduledOrRunning = true;
           } else { 
               logger.trace("scheduled {} disabled", toString());
               cancellable = null;
               isScheduledOrRunning = false;
           }
       }
    
    }

`AsyncRefreshTask` 在实例化时就启动调度任务，执行其 `runInternal()` 方法(方法重写逻辑这里没有详细说明)，也就是 `indexService.maybeRefreshEngine(false);` ，非强制性的执行 refresh 任务。

## refresh 在Lucene层执行逻辑 ##

Elasticsearch 曾的`refresh`最终会调用到 Lucene的 `SearcherManager#refreshIfNeeded(...)` 方法：

### SearcherManager\#refreshIfNeeded ###

public final class SearcherManager extends ReferenceManager<IndexSearcher> { 
    
    	// 在需要时刷新参数指定的Index
    	protected IndexSearcher refreshIfNeeded(IndexSearcher referenceToRefresh) throws IOException { 
            final IndexReader r = referenceToRefresh.getIndexReader();
            assert r instanceof DirectoryReader : "searcher's IndexReader should be a DirectoryReader, but got " + r;
            // 如果当前索引目录有变化，比如生成了代表修改删除的.liv 文件，内存中有新写入的未生成segment的document，对当前索引目录生成一个新的Reader
            final IndexReader newReader = DirectoryReader.openIfChanged((DirectoryReader)r);
            // 如果新的Reader == null, 也就是说目录内数据未发生变化
            if (newReader == null) { 
                return null;
            } else { 
                // 通过新的Reader 生成新的IndexSearcher
                return getSearcher(searcherFactory, newReader, r);
            }
        }
    
    }

### StandardDirectoryReader\#doOpenIfChanged ###

public final class StandardDirectoryReader extends DirectoryReader { 
    
    	private IndexWriter writer;
    
    	protected DirectoryReader doOpenIfChanged(final IndexCommit commit) throws IOException { 
            // If we were obtained by writer.getReader(), re-ask the
            // writer to get a new reader.
            if (writer != null) { 
                return doOpenFromWriter(commit);
            } else { 
                return doOpenNoWriter(commit);
            }
        }
    
    	/** * 重新打开一个Reader * * @param commit * @return * @throws IOException */
        private DirectoryReader doOpenFromWriter(IndexCommit commit) throws IOException { 
            if (commit != null) { 
                return doOpenFromCommit(commit);
            }
            // segmentInfos 是否发生变化, 具体逻辑查看下一个方法
            if (writer.nrtIsCurrent(segmentInfos)) { 
                return null;
            }
    		// 重新生成一个目录的Reader
            DirectoryReader reader = writer.getReader(applyAllDeletes, writeAllDeletes);
    
            // If in fact no changes took place, return null:
            if (reader.getVersion() == segmentInfos.getVersion()) { 
                reader.decRef();
                return null;
            }
    
            return reader;
        }
    
    }

### IndexWriter\#nrtIsCurrent ###

public class IndexWriter implements Closeable, TwoPhaseCommit, Accountable { 
    
    	/** * 当前segmengInfos版本没有发生变化, 比如是否有新的segment生成, 内存总是否有新写入的没有写入到segment的document等等 * * @param infos * @return */
        synchronized boolean nrtIsCurrent(SegmentInfos infos) { 
            ensureOpen();
            boolean isCurrent = infos.getVersion() == segmentInfos.getVersion()
                && docWriter.anyChanges() == false      // 内存中是否有document, 也就是不在segment里的 document
                && bufferedUpdatesStream.any() == false // 内存中是否有更新的数据
                && readerPool.anyChanges() == false;
            // ......
            }
            return isCurrent;
        }
    
    }

当有新的document被写入，或者修改，删除等情况时，会触发重新构建 `DirectoryReader` ，也就是重新关联索引目录下的所有数据。

### IndexWriter\#getReader ###

public class IndexWriter implements Closeable, TwoPhaseCommit, Accountable { 
    
    	final DocumentsWriter docWriter;
    
    	DirectoryReader getReader(boolean applyAllDeletes, boolean writeAllDeletes) throws IOException { 
    		// ......
    		DirectoryReader r = null;
    		try { 
                boolean success = false;
                synchronized (fullFlushLock) { 
                    try { 
                        // 1.flush 所有DWPT, Document Writer Per Thread
                        long seqNo = docWriter.flushAllThreads();
                       // ......
                        synchronized (this) { 
                            // ......
                            // 重新打开索引目录
                            r = StandardDirectoryReader.open(this, segmentInfos, applyAllDeletes, writeAllDeletes);
                            // ......
                        }
                        success = true;
                    } finally { 
                       // ......
                }
                anyChanges |= maybeMerge.getAndSet(false);
                if (anyChanges) { 
                	// 2. 触发 Merge, 也就是每次refresh 都有可能出发segment 合并
                    maybeMerge(config.getMergePolicy(), MergeTrigger.FULL_FLUSH, UNBOUNDED_MAX_MERGE_SEGMENTS);
                }
                // ......
            } catch (AbortingException | VirtualMachineError tragedy) { 
               ......
            } 
             return r;
    	}
    
    }

上述代码中，有两个步骤比较关键，就是 `flushAllThreads()` 和 `merge(......)`

### docWriter.flushAllThreads() (关键内容) ###

这里的Threads并不是线程的意思，而是`DocumentWriterPerThread(DWPT)`的对象，Lucene维护了一个`DWPT的列表`，专门用于写入Document数据，当Flush时，每个DWPT的数据都能生成一个Segment。

每个DWPT对象同一时间只能被一个线程使用，当一个Index线程在获取一个DWPT对象时，如果有空闲的DWPT对象，则从空闲列表中获取锁定一个；当没有空闲的DWPT时，创建一个，在用完后放入空闲列表中；

当Elasticsearch写入压力很大时，`Lucene的DWPT的池子容量会增大`，上限是和Elasticsearch的 `index` 线程池的线程数上限一致。所以当写入压力较大时，适当增大 `index` 的线程池的线程数上限能够增大写入吞吐量。

Elasticsearch 每次 `refresh`时，大部分场景下都会调用 Lucene 的 `flush`，每个DWPT对象持有的数据都会生成一个Segment，在Flush后释放DWPT池的所有对象，之后Index时重新创建。

每个DWPT在Flush时都会在内存中生成一个segment，`但没有实时Sync同步到磁盘`，没有调用`commit` 提交数据到磁盘，具体调用时机有操作系统自行决定，比如当前操作系统比较空闲，则能立马同步磁盘，生成相应的segment文件，比如 `.cfe，.cfs，.si` 文件。

### maybeMerge(…); ###

每一次`flush`，如果生成了新的segment，则都会触发 `Segment Merge`。触发 Merge 操作不表示一定会 merge segment，是否要merge是由 `Merge Policy` 来决定的。 Merge Policy 有`ForceMergePolicy`(ES强制merge)，`TieredMergePolicy(阶梯式的，Lucene默认策略)`等等

## refresh search idle ##

`Refresh 的作用是将内存中新写入的document生成一个segment，基于系统文件形式缓存在内存中，这样Lucene就能够检索（Lucene基于系统文件缓存的形式来读取加载数据）。`

如果`当前系统没有检索的需求`，那么即使数据不断的写入，`延迟refresh的执行时间`，那么就能一次性生成比较大的segment，而不是反复的生成小的segment，然后merge，这种情况下性能会有明显的提高，毕竟不需要多次的刷盘，文件句柄也能有效减少。`search idle` 就是基于上述原理做的一个refresh 优化策略。

在上文中我们展示了Elasticsearch 的 `AsyncRefreshTask`，即执行refresh的任务类，在创建实列时会定时调度执行refresh，具体代码如下：

### IndexShard\#scheduledRefresh ###

/** * Executes a scheduled refresh if necessary. * * @return <code>true</code> iff the engine got refreshed otherwise <code>false</code> */
        public boolean scheduledRefresh() { 
            verifyNotClosed();
            // 当前是否有refreshListener，也就是要强制refresh然后回调
            boolean listenerNeedsRefresh = refreshListeners.refreshNeeded();
            // 综合条件是否要执行当前调度的 refresh 在指定索引上是否有新的变化，比如增删改，有的话 need
            if (isReadAllowed() && (listenerNeedsRefresh || getEngine().refreshNeeded())) { 
    
                if (listenerNeedsRefresh == false                       // if we have a listener that is waiting for a refresh we need to force it
                    && isSearchIdle()                                   // 当前 index shard 是否是search idle
                    && indexSettings.isExplicitRefresh() == false       // 没有显式的设置了index.refresh_interval
                    && active.get()) {                                   // it must be active otherwise we might not free up segment memory once the shard became inactive
                    // lets skip this refresh since we are search idle and
                    // don't necessarily need to refresh. the next searcher access will register a refreshListener and that will
                    // cause the next schedule to refresh.
                    final Engine engine = getEngine();
                    engine.maybePruneDeletes(); // try to prune the deletes in the engine if we accumulated some
                    // 推迟refersh，刷新 pendingRefreshLocation, 也就是translog里refresh起始location
                    setRefreshPending(engine);
                    return false;
                } else { 
                    if (logger.isTraceEnabled()) { 
                        logger.trace("refresh with source [schedule]");
                    }
                    return getEngine().maybeRefresh("schedule");
                }
            }
            final Engine engine = getEngine();
            engine.maybePruneDeletes(); // try to prune the deletes in the engine if we accumulated some
            return false;
        }
    
        /** * 判断当前index shard 是否是检索空闲状态 * Returns true if this shards is search idle * @see IndexSettings#INDEX_SEARCH_IDLE_AFTER */
        final boolean isSearchIdle() { 
            // 当前时间 - 最新的search时间 >= 设定的检索空闲时间
            return (threadPool.relativeTimeInMillis() - lastSearcherAccess.get()) >= indexSettings.getSearchIdleAfter().getMillis();
        }
    
    	/** * 推迟refresh * @param engine */
        private void setRefreshPending(Engine engine) { 
            // 上次refresh是的transLog location
            Translog.Location lastWriteLocation = engine.getTranslogLastWriteLocation();
            Translog.Location location;
            do { 
                // 先获取上一次refresh 的 location
                // 目前这变量没有特殊作用，仅仅是做一个标识，有值时需要先 force refresh, 再 serach
                location = this.pendingRefreshLocation.get();
                // 如果之前已经设置了localtion， 那么不能讲location后退，因为会丢失数据
                if (location != null && lastWriteLocation.compareTo(location) <= 0) { 
                    break;
                }
                // 更新pending refresh 的 location
            } while (pendingRefreshLocation.compareAndSet(location, lastWriteLocation) == false);
        }

Elasticsearch 在 refresh前会做一个判断，如果索引可读且有必要refresh时才refresh，有必要的条件包括以下两步：

1.  没有注册`refreshListener`，如果注册了要强制refresh，之后回调
2.  索引上有变化，比如有写入， 修改，删除等操作

当满足了上述前置条件后，才算可以进行refresh，但是否确定要refresh还需要做进一步的条件判断，`search idle` 就在这一步骤生效。如果不立即refresh，需要满足下述条件

1.  没有注册`refreshListener`，也就是没有强制refresh的请求
2.  当前是`search idle 状态`
3.  `没有显式的设置index.refresh_interval`，也就是refresh的间隔是默认值1s
4.  当前index正常生效

当满足了上述所有条件时，会`推迟refresh`的执行时机，会`标记 translog 上一次写的位置`，下次refresh时从translog上次位置处开始回放。

上述条件中有一个不是很容易理解， `search idle` 状态，顾名思义，就是检索空闲，当一段时间没有检索请求时就是检索空闲状态，检索请求类型包括 `search, get, explain`。一段时间是多久，由参数 `index.search.idle.after` 指定，默认`30s`，也就是当30s内没有检索请求时，就会进入`search idle` 状态，会推迟refresh，那具体`推迟到什么时候执行？`

上述代码的最后一段 `setRefreshPending` ，将translog的上次写入位置标识下来，保持进变量 `pendingRefreshLocation` 中，然后在下次要执行 search 相关请求时`先refresh 然后 search`， `pendingRefreshLocation`目前`没有特殊作用，仅仅是一个标识`，当有值时需要`先force refresh，再search`，所以`refresh其实和tranlog没有什么关系` 。具体代码如下：

### IndexShard\#awaitShardSearchActive ###

/** * 注册一个listener, 当shard 又接到一个search请求时 触发执行所有的pending refresh , refresh 基于 translog location * 也就是基于translog的指定位置, 后续的index,update,delete 数据都记录在translog的location后, 从此location后开始apply * 此方法主要用作定位transLog 的 location * * Registers the given listener and invokes it once the shard is active again and all * pending refresh translog location has been refreshed. If there is no pending refresh location registered the listener will be * invoked immediately. * @param listener the listener to invoke once the pending refresh location is visible. The listener will be called with * <code>true</code> if the listener was registered to wait for a refresh. */
        public final void awaitShardSearchActive(Consumer<Boolean> listener) { 
            markSearcherAccessed(); // move the shard into non-search idle
            // 获取pending refresh 的 location
            final Translog.Location location = pendingRefreshLocation.get();
            // 如果存在则说明有必要先refresh 再 search。一般适用于 search idle 情况
            if (location != null) { 
                // 添加refresh listener, 当接收到search请求时先触发resresh, 然后执行search
                addRefreshListener(location, (b) -> { 
                	// 将translog refresh location 重置,重置 search idle 状态
                    pendingRefreshLocation.compareAndSet(location, null);
                    // 响应request
                    listener.accept(true);
                });
            } else { 
                listener.accept(false);
            }
        }
    
        public void addRefreshListener(Translog.Location location, Consumer<Boolean> listener) { 
            final boolean readAllowed;
            if (isReadAllowed()) { 
                readAllowed = true;
            } else { 
                synchronized (mutex) { 
                    readAllowed = isReadAllowed();
                }
            }
            if (readAllowed) { 
            	// force refresh， 强制refresh
                refreshListeners.addOrNotify(location, listener);
            } else { 
                // we're not yet ready fo ready for reads, just ignore refresh cycles
                listener.accept(false);
            }
        }

从上述 `search ilde` 机制中我们能够知晓，当数据批量导入时，没有检索，持续30s后就会进入search idle 状态，refresh被无限推迟直到下一次search类型请求，所以在`批量导入数据无检索请求时，修改`**index.refresh\_interval**`其实没什么屌用，哈哈`

## 总结 ##

本篇博客从多个角度分享了Elasticsearch 的 refresh机制，从es到lucene的执行流程，配置方式，search idle 等等，因该能让你对refresh有一个比较深入的理解，有疑问欢迎留言。