重构实体方法以避免并发问题答案

【问题标题】：Refactoring entity method to avoid concurrency problems重构实体方法以避免并发问题
【发布时间】：2017-02-12 16:38:12
【问题描述】：

我正在尝试使用 DDD 的项目接近尾声，但发现了一个明显的错误，我不确定如何轻松解决。

这是我的实体 - 为了简单起见，我将其简化：

public class Contribution : Entity
{
    protected Contribution()
    {
        this.Parts = new List<ContributionPart>();
    }

    internal Contribution(Guid id)
    {
        this.Id = id;
        this.Parts = new List<ContributionPart>();
    }

    public Guid Id { get; private set; }

    protected virtual IList<ContributionPart> Parts { get; private set; }

    public void UploadParts(string path, IEnumerable<long> partLengths)
    {
        if (this.Parts.Count > 0)
        {
            throw new InvalidOperationException("Parts have already been uploaded.");
        }

        long startPosition = 0;
        int partNumber = 1;

        foreach (long partLength in partLengths)
        {
            this.Parts.Add(new ContributionPart(this.Id, partNumber, partLength));
            this.Commands.Add(new UploadContributionPartCommand(this.Id, partNumber, path, startPosition, partLength));
            startPosition += partLength;
            partNumber++;
        }
    }

    public void SetUploadResult(int partNumber, string etag)
    {
        if (etag == null)
        {
            throw new ArgumentNullException(nameof(etag));
        }

        ContributionPart part = this.Parts.SingleOrDefault(p => p.PartNumber == partNumber);

        if (part == null)
        {
            throw new ContributionPartNotFoundException(this.Id, partNumber);
        }

        part.SetUploadResult(etag);

        if (this.Parts.All(p => p.IsUploaded))
        {
            IEnumerable<PartUploadedResult> results = this.Parts.Select(p => new PartUploadedResult(p.PartNumber, p.ETag));
            this.Events.Add(new ContributionUploaded(this.Id, results));
        }
    }
}

我的错误发生在 SetUploadResult 方法中。基本上，多个线程同时执行上传，然后在上传结束时调用 SetUploadResult。但由于实体是在几秒钟前加载的，每个线程都会在实体的不同实例上调用 SetUploadResult，因此测试 if (this.Parts.All(p => p.IsUploaded) 永远不会评估为 true。

我不确定如何轻松解决此问题。将多个 UploadContributionPartCommands 添加到 Commands 集合背后的想法是，每个 ContributionPart 可以并行上传 - 我的 CommandBus 确保了这一点 - 但每个部分并行上传，它会导致我的实体逻辑出现问题。

【问题讨论】：

所以你是说有多个线程在同一个 Contribution 实体实例上运行？
正确。贡献实体为每个 partLength 创建了一个 UploadContributionPartCommand，每个 UploadContributionPartCommandHandler 并行执行，因此并行调用 SetUploadResult。除了它不是实体的同一个内存实例，而是同一个实体。
每个实例都有自己的部分，为什么不是同一个“内存中的实例”？
如果每个线程都在自己的实例上运行，那么您的问题在哪里？可以被多个不同线程访问的共享字段/属性是什么？
实体应该处理像上传数据这样的横切关注点听起来不像 DDD。我认为你的问题的根源是你试图让对象在不应该由它们负责的并行动作中合作。

标签： c# multithreading entity-framework concurrency domain-driven-design

【解决方案1】：

我认为您可以重构Contribution，使其无法处理SetUploadResult。它将贡献实体解耦，SetUploadResult 的副作用被隔离，将技术问题排除在Contribution 域模型之外。

创建一个包含SetUploadResult 正在做什么的调度程序类。

一旦Contribution 实体执行完其逻辑，执行线程将返回到应用程序服务。正是在这一点上，来自实体的事件可以被馈送到调度程序中。

如果它们是长时间运行的进程，您可以将它们添加为任务集合并异步运行它们。然后，您可以等待所有任务完成。您可以在 SO 中搜索如何执行此操作。

var results = await Task.WhenAll(task1, task2,...taskN);

【讨论】：

同意。跟踪上传进度绝对是应用程序级别的问题，而不是域问题。
公平评论。我们围绕贡献和贡献的部分进行了很多讨论，我们决定将它们作为域的一部分，但质疑它们是否应该是合理的。

【解决方案2】：

如果多个线程可能同时调用SetUploadResult 方法并且您有竞争条件，您应该使用诸如锁之类的同步机制来保护临界区：https://msdn.microsoft.com/en-us/library/c5kehkcz.aspx。

如果您设置锁定字段static，它将在您的实体类型的所有实例之间共享，例如：

private static readonly object _lock = new object();
public void SetUploadResult(int partNumber, string etag)
{
    if (etag == null)
    {
        throw new ArgumentNullException(nameof(etag));
    }

    ContributionPart part = this.Parts.SingleOrDefault(p => p.PartNumber == partNumber);

    if (part == null)
    {
        throw new ContributionPartNotFoundException(this.Id, partNumber);
    }

    part.SetUploadResult(etag);

    lock (_lock) //Only one thread at a time can enter this critical section.
                 //The second thread will wait here until the first thread leaves the critical section.
    {
        if (this.Parts.All(p => p.IsUploaded))
        {
            IEnumerable<PartUploadedResult> results = this.Parts.Select(p => new PartUploadedResult(p.PartNumber, p.ETag));
            this.Events.Add(new ContributionUploaded(this.Id, results));
        }
    }
}

【讨论】：