elasticsearch索引index之engine讀寫控制結(jié)構實現(xiàn)

更新時間：2022年04月22日 10:14:31 作者：zziawan

這篇文章主要為大家介紹了elasticsearch索引index之engine讀寫控制結(jié)構實現(xiàn)，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪

engine的實現(xiàn)結(jié)構

elasticsearch對于索引中的數(shù)據(jù)操作如讀寫get等接口都封裝在engine中，同時engine還封裝了索引的讀寫控制，如流量、錯誤處理等。engine是離lucene最近的一部分。

engine的實現(xiàn)結(jié)構如下所示：

engine接口有三個實現(xiàn)類，主要邏輯都在InternalEngine中。

ShadowEngine之實現(xiàn)了engine接口的部分讀方法，主要用于對于索引的讀操作。

shardFSEngine在InternalEngine的基礎上實現(xiàn)了recovery方法，它的功能跟InternalEngine基本相同只是它的recovery過程有區(qū)別，不會對Translog和index進行快照存儲。

Engine類定義了一些index操作的主要方法和內(nèi)部類，方法如create，index等。內(nèi)部類如index，delete等。這些方法的實現(xiàn)是在子類中，這些方法的參數(shù)是這些內(nèi)部類。

Engine類的方法：

public abstract void create(Create create) throws EngineException;
    public abstract void index(Index index) throws EngineException;
    public abstract void delete(Delete delete) throws EngineException;
    public abstract void delete(DeleteByQuery delete) throws EngineException;

這些抽象方法都在子類中實現(xiàn)，它們的參數(shù)都是一類，這些都是Engine的內(nèi)部類，這些內(nèi)部類類似于實體類，沒有相關邏輯只是由很多filed及get方法構成。如Create和Index都繼承自IndexOperation，它們所有信息都存儲到IndexOperation的相關Field中，IndexOperation如下所示：

public static abstract class IndexingOperation implements Operation {
        private final DocumentMapper docMapper;
        private final Term uid;
        private final ParsedDocument doc;
        private long version;
        private final VersionType versionType;
        private final Origin origin;
        private final boolean canHaveDuplicates;
        private final long startTime;
        private long endTime;
    ………………
}

無論是Index還是Create，相關數(shù)據(jù)和配置都在doc中，根據(jù)doc和docMapper就能夠獲取本次操作的所有信息，另外的一些字段如version，uid都是在類初始化時構建。這樣傳給實際方法的是一個class，在方法內(nèi)部根據(jù)需求獲取到相應的數(shù)據(jù)

如index方法的實現(xiàn)：

private void innerIndex(Index index) throws IOException {
        synchronized (dirtyLock(index.uid())) {
            final long currentVersion;
            VersionValue versionValue = versionMap.getUnderLock(index.uid().bytes());
            if (versionValue == null) {
                currentVersion = loadCurrentVersionFromIndex(index.uid());
            } else {
                if (engineConfig.isEnableGcDeletes() &amp;&amp; versionValue.delete() &amp;&amp; (engineConfig.getThreadPool().estimatedTimeInMillis() - versionValue.time()) &gt; engineConfig.getGcDeletesInMillis()) {
                    currentVersion = Versions.NOT_FOUND; // deleted, and GC
                } else {
                    currentVersion = versionValue.version();
                }
            }
            long updatedVersion;
            long expectedVersion = index.version();
            if (index.versionType().isVersionConflictForWrites(currentVersion, expectedVersion)) {
                if (index.origin() == Operation.Origin.RECOVERY) {
                    return;
                } else {
                    throw new VersionConflictEngineException(shardId, index.type(), index.id(), currentVersion, expectedVersion);
                }
            }
            updatedVersion = index.versionType().updateVersion(currentVersion, expectedVersion);
            index.updateVersion(updatedVersion);
            if (currentVersion == Versions.NOT_FOUND) {
                // document does not exists, we can optimize for create
                index.created(true);
                if (index.docs().size() &gt; 1) {
                    indexWriter.addDocuments(index.docs(), index.analyzer());
                } else {
                    indexWriter.addDocument(index.docs().get(0), index.analyzer());
                }
            } else {
                if (versionValue != null) {
                    index.created(versionValue.delete()); // we have a delete which is not GC'ed...
                }
                if (index.docs().size() &gt; 1) {
                    indexWriter.updateDocuments(index.uid(), index.docs(), index.analyzer());//獲取IndexOperation中doc中字段更新索引
                } else {
                    indexWriter.updateDocument(index.uid(), index.docs().get(0), index.analyzer());
                }
            }
            Translog.Location translogLocation = translog.add(new Translog.Index(index));//寫translog
            versionMap.putUnderLock(index.uid().bytes(), new VersionValue(updatedVersion, translogLocation));
            indexingService.postIndexUnderLock(index);
        }
    }

這就是Engine中create、index這些方法的實現(xiàn)方式。后面分析索引過程中會有更加詳細說明。Engine中還有獲取索引狀態(tài)（元數(shù)據(jù)）及索引操作的方法如merge。這些方法也是在子類中調(diào)用lucene的相關接口，跟create，index，get很類似。因為沒有深入Engine的方法實現(xiàn)，因此這里的分析比較簡單，后面的分析會涉及這里面很多方法。

總結(jié)

這里只是從結(jié)構上對indexEngine進行了簡單說明，它里面的方法是es對lucene索引操作方法的封裝，只是增加了一下處理方面的邏輯如寫translog，異常處理等。它的操作對象是shard，es所有對shard的寫操作都是通過Engine來實現(xiàn)，后面的分析會有所體現(xiàn)。

以上就是elasticsearch索引index之engine讀寫控制結(jié)構實現(xiàn)的詳細內(nèi)容，更多關于elasticsearch索引index engine讀寫控制的資料請關注腳本之家其它相關文章！

您可能感興趣的文章: