带版本控制的 Elasticsearch 更新 - NEST/C#答案

【问题标题】：Elasticsearch Update with Versioning - NEST/C#带版本控制的 Elasticsearch 更新 - NEST/C#
【发布时间】：2017-05-06 18:25:04
【问题描述】：

我希望实现弹性搜索版本控制功能，以使用 NEST 库和 C# 更新记录。我实际上正在寻找一个帮助器，它执行以下操作：

读取现有记录。
对记录进行更改。
使用版本功能更新文档。

我进行了一些研究，但没有找到我正在寻找的信息。谁能给我一些代码示例、实现或测试？

【问题讨论】：

几年前我写了一篇关于 ES 和 NEST 的乐观并发控制的博文：forloop.co.uk/blog/… 它使用 Elasticsearch 1.x 和 NEST 1.x，但在 2.x 中基本相同和 5.x
它让我清楚地了解了如何实现该功能。非常感谢！
嗨@RussCam，在您的帖子中，我注意到您使用 ElasticsearchServerException 对象来捕获和处理重试。由于某种原因，我无法在我的代码中访问此对象。我看到 ElasticsearchClientException。能否指出如何访问服务器异常？

标签： c# elasticsearch nest

【解决方案1】：

我使用以下类，其中包括版本控制Update

public class SampleElasticClient
{
    private const string VERSION_CONFLICT_ERROR = "version_conflict_engine_exception";

    protected readonly string IndexName;

    protected readonly ElasticClient Client;

    public SampleElasticClient(Uri uri, string indexName)
    {
        Client = new ElasticClient(new ConnectionSettings(uri).DefaultIndex(indexName));
        IndexName = indexName;
    }

    public IGetResponse<T> Get<T>(Id id) where T : class
    {
        var request = new GetRequest<T>(IndexName, typeof(T), id);
        var response = Client.Get<T>(request);
        EnsureSuccessResponse(response);
        return response;
    }

    public void Update<T>(Id id, Func<T, T> update, int retriesCount = 10) where T : class
    {
        Retry(() =>
        {
            var getResponse = Get<T>(id);
            var item = update(getResponse.Source);
            return Client.Index(item, index => getResponse.Found
                ? index.Version(getResponse.Version)
                : index.OpType(OpType.Create));
        }, retriesCount);
    }

    protected void EnsureSuccessResponse(IResponse response)
    {
        if (!response.IsValid && response.ApiCall.HttpStatusCode != 404)
        {
            var errorMessage = response.ServerError != null
                ? $"ElasticSearch error: {response.ServerError.Error}\r\n" +
                               $"Http status: {response.ServerError.Status}"
                : $"ElasticSearch error. {response.DebugInformation}";
            throw new Exception(errorMessage);
        }
    }

    protected void Retry(Func<IResponse> execute, int retriesCount)
    {
        var numberOfRetry = 0;
        do
        {
            var response = execute();
            if (response.ServerError?.Error.Type != VERSION_CONFLICT_ERROR || ++numberOfRetry == retriesCount)
            {
                EnsureSuccessResponse(response);
                return;
            }
        } while (true);
    }
}

Retry 方法负责处理version_conflict_engine_exception 并重试更新。 Update 方法 enther 使用 lambda 插入或更新实体以处理从索引中检索到的实体。这是一个使用这个类的例子

var client = new SampleElasticClient(new Uri("http://localhost:9200"), indexName);
var id = 123;
client.Update<Sample>(id, entity =>
{
    if (entity == null)
        entity = new Sample { Id = id }; // Or any other action for new entity

    entity.MyField = "new value";
    return entity;
});

【讨论】：