返回给调用者的异步结果句柄

Question

我有一个方法将某些工作排队以异步执行。我想返回调用者的某种句柄，可以轮询，等待或用于从操作中获取返回值，但找不到适合该任务的类或接口。

BackgroundWorker靠近了，但它适用于工作人员拥有自己的专用线程的情况，这在我的情况中并非如此。 IAsyncResult看起来很有前途，但提供的AsyncResult实现对我来说也是无法使用的。我应该自己实现IAsyncResult吗？

澄清：

我有一个类，概念上看起来是这样的：

class AsyncScheduler 
{ 

    private List<object> _workList = new List<object>(); 
    private bool _finished = false; 

    public SomeHandle QueueAsyncWork(object workObject) 
    { 
     // simplified for the sake of example 
     _workList.Add(workObject); 
     return SomeHandle; 
    } 

    private void WorkThread() 
    { 
     // simplified for the sake of example 
     while (!_finished) 
     { 
      foreach (object workObject in _workList) 
      { 
       if (!workObject.IsFinished) 
       { 
        workObject.DoSomeWork(); 
       } 
      } 
      Thread.Sleep(1000); 
     } 
    } 
} 

的QueueAsyncWork函数把一个工作项目到轮询名单有专门的工作线程，其中将有只有一个。我的问题不在于编写QueueAsyncWork函数 - 这很好。我的问题是，我会返回给调用者的是什么？ SomeHandle应该是什么？

现有的.Net类适用于异步操作可封装在返回的单个方法调用中的情况。情况并非如此 - 所有工作对象都在同一个线程上工作，而完整的工作操作可能会跨越多次调用workObject.DoSomeWork()。在这种情况下，向调用者提供进度通知，完成和获得操作的最终结果的一些处理的合理方法是什么？

Answer 1

是的，实现IAsyncResult（或更确切地说，它的扩展版本，以提供进度报告）。

public class WorkObjectHandle : IAsyncResult, IDisposable 
{ 
    private int _percentComplete; 
    private ManualResetEvent _waitHandle; 
    public int PercentComplete { 
     get {return _percentComplete;} 
     set 
     { 
      if (value < 0 || value > 100) throw new InvalidArgumentException("Percent complete should be between 0 and 100"); 
      if (_percentComplete = 100) throw new InvalidOperationException("Already complete"); 
      if (value == 100 && Complete != null) Complete(this, new CompleteArgs(WorkObject)); 
      _percentComplete = value; 
     } 
    public IWorkObject WorkObject {get; private set;} 
    public object AsyncState {get {return WorkObject;}} 
    public bool IsCompleted {get {return _percentComplete == 100;}} 
    public event EventHandler<CompleteArgs> Complete; // CompleteArgs in a usual pattern 
    // you may also want to have Progress event 
    public bool CompletedSynchronously {get {return false;}} 
    public WaitHandle 
    { 
     get 
     { 
      // initialize it lazily 
      if (_waitHandle == null) 
      { 
       ManualResetEvent newWaitHandle = new ManualResetEvent(false); 
       if (Interlocked.CompareExchange(ref _waitHandle, newWaitHandle, null) != null) 
        newWaitHandle.Dispose(); 
      } 
      return _waitHandle; 
     } 
    } 

    public void Dispose() 
    { 
     if (_waitHandle != null) 
      _waitHandle.Dispose(); 
     // dispose _workObject too, if needed 
    } 

    public WorkObjectHandle(IWorkObject workObject) 
    { 
     WorkObject = workObject; 
     _percentComplete = 0; 
    } 
} 

public class AsyncScheduler 
{ 
    private Queue<WorkObjectHandle> _workQueue = new Queue<WorkObjectHandle>(); 
    private bool _finished = false; 

    public WorkObjectHandle QueueAsyncWork(IWorkObject workObject) 
    { 
     var handle = new WorkObjectHandle(workObject); 
     lock(_workQueue) 
     { 
      _workQueue.Enqueue(handle); 
     } 
     return handle; 
    } 

    private void WorkThread() 
    { 
     // simplified for the sake of example 
     while (!_finished) 
     { 
      WorkObjectHandle handle; 
      lock(_workQueue) 
      { 
       if (_workQueue.Count == 0) break; 
       handle = _workQueue.Dequeue(); 
      } 
      try 
      { 
       var workObject = handle.WorkObject; 
       // do whatever you want with workObject, set handle.PercentCompleted, etc. 
      } 
      finally 
      { 
       handle.Dispose(); 
      } 
     } 
    } 
} 

Answer 2

最简单的方法是here。假设你有一个方法string DoSomeWork(int)。然后，您创建正确类型的委托，例如：

Func<int, string> myDelegate = DoSomeWork; 

然后调用BeginInvoke方法的委托：一旦你的异步调用已完成

int parameter = 10; 
myDelegate.BeginInvoke(parameter, Callback, null); 

回调委托将被调用。您可以按如下定义这个方法：

void Callback(IAsyncResult result) 
{ 
    var asyncResult = (AsyncResult) result; 
    var @delegate = (Func<int, string>) asyncResult.AsyncDelegate; 
    string methodReturnValue = @delegate.EndInvoke(result); 
} 

使用所描述的场景，您也可以查询结果，或等待他们。看看我提供的更多信息。

问候， 罗纳德

Answer 3

如果你不想使用异步回调，您可以使用一个明确的WaitHandle，如ManualResetEvent的：

public abstract class WorkObject : IDispose 
{ 
    ManualResetEvent _waitHandle = new ManualResetEvent(false); 

    public void DoSomeWork() 
    { 
     try 
     { 
      this.DoSomeWorkOverride(); 
     } 
     finally 
     { 
      _waitHandle.Set(); 
     } 
    } 

    protected abstract DoSomeWorkOverride(); 

    public void WaitForCompletion() 
    { 
     _waitHandle.WaitOne(); 
    } 

    public void Dispose() 
    { 
     _waitHandle.Dispose(); 
    } 
} 

而在你的代码，你可以说

using (var workObject = new SomeConcreteWorkObject()) 
{ 
    asyncScheduler.QueueAsyncWork(workObject); 
    workObject.WaitForCompletion(); 
} 

不要忘记调用你的workObject Dispose。

您可以随时使用它创建一个这样的包装为每一个工作对象和谁打电话_waitHandle.Dispose（）在WaitForCompletion（可选的实施方式），你可以懒洋洋地实例等待句柄（请注意：比赛条件提前）等（这几乎是BeginInvoke为代表做的事情。）

Answer 4

如果我理解正确，您有一组工作对象（IWorkObject），每个工作对象通过多次调用DoSomeWork方法完成一项任务。当某个对象完成其工作时，您希望以某种方式对此进行响应，并且在此过程中您希望响应任何已报告的进度？

在这种情况下，我建议你采取稍微不同的方法。你可以看看Parallel Extension framework（blog）。使用框架，你可以写这样的事情：

public void QueueWork(IWorkObject workObject) 
{ 
    Task.TaskFactory.StartNew(() => 
     { 
      while (!workObject.Finished) 
      { 
       int progress = workObject.DoSomeWork(); 
       DoSomethingWithReportedProgress(workObject, progress); 
      } 
      WorkObjectIsFinished(workObject); 
     }); 
} 

需要注意以下几点：

• QueueWork现在返回void。原因在于报告进度或任务完成时发生的操作已成为执行工作的线程的一部分。您当然可以返回工厂创建的Task，并从方法返回（例如启用轮询）。
• 进度报告和完成处理现在是线程的一部分，因为您应该尽可能避免轮询。轮询更加昂贵，因为通常你轮询太频繁（太早）或不够频繁（太迟）。没有理由不能在运行任务的线程中报告任务的进度和完成情况。
• 以上也可以使用（较低级别）ThreadPool.QueueUserWorkItem方法来实现。

使用QueueUserWorkItem：

public void QueueWork(IWorkObject workObject) 
{ 
    ThreadPool.QueueUserWorkItem(() => 
     { 
      while (!workObject.Finished) 
      { 
       int progress = workObject.DoSomeWork(); 
       DoSomethingWithReportedProgress(workObject, progress); 
      } 
      WorkObjectIsFinished(workObject); 
     }); 
} 

Answer 5

的工作对象类可以包含需要跟踪的特性。

public class WorkObject 
{ 
    public PercentComplete { get; private set; } 
    public IsFinished { get; private set; } 

    public void DoSomeWork() 
    { 
     // work done here 

     this.PercentComplete = 50; 

     // some more work done here 

     this.PercentComplete = 100; 
     this.IsFinished = true; 
    } 
} 

然后在您的例子：

• 变化从一个列表集合到可容纳GUID值（或唯一标识值的任何其他装置）的字典。
• 通过让主叫方通过它从QueueAsyncWork收到的Guid来公开正确的WorkObject属性。

我假设你会开始WorkThread异步（虽然，唯一的异步线程）;另外，您必须检索字典值和WorkObject属性线程安全。

private Dictionary<Guid, WorkObject> _workList = 
    new Dictionary<Guid, WorkObject>(); 

private bool _finished = false; 

public Guid QueueAsyncWork(WorkObject workObject) 
{ 
    Guid guid = Guid.NewGuid(); 
    // simplified for the sake of example 
    _workList.Add(guid, workObject); 
    return guid; 
} 

private void WorkThread() 
{ 
    // simplified for the sake of example 
    while (!_finished) 
    { 
     foreach (WorkObject workObject in _workList) 
     { 
      if (!workObject.IsFinished) 
      { 
       workObject.DoSomeWork(); 
      } 
     } 
     Thread.Sleep(1000); 
    } 
} 

// an example of getting the WorkObject's property 
public int GetPercentComplete(Guid guid) 
{ 
    WorkObject workObject = null; 
    if (!_workList.TryGetValue(guid, out workObject) 
     throw new Exception("Unable to find Guid"); 

    return workObject.PercentComplete; 
}