Метод рефакторинга сущности, чтобы избежать проблем с параллелизмом

Question

Я приближаюсь к концу проекта, для которого я пытаюсь использовать 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 обеспечивает это, но с каждой загруженной частью параллельно это вызывает проблемы для моей логики сущности.

Answer 1

Если несколько потоков могут одновременно вызвать метод 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)); 
     } 
    } 
} 

Answer 2

Я думаю, что вы можете реорганизовать Contribution так, что он не будет обрабатывать SetUploadResult. Он будет отделять объект Contribution, а побочные эффекты SetUploadResult изолированы, сохраняя технический интерес из модели домена Contribution.

Создайте класс диспетчера, который содержит то, что делает SetUploadResult.

После того как объект Contribution завершит выполнение своей логики, поток выполнения вернется в службу приложения. Именно в этот момент события от субъекта могут быть отправлены в диспетчер.

Если они длительные, вы можете добавить их в виде набора задач и выполнить их асинхронно. Тогда вы можете просто ждать, когда все задачи будут выполнены. Вы можете искать в SO о том, как это сделать.

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