我试图使用异步方法调用开发方法管道.管道的逻辑如下
>集合中有n个数据必须被送入管道中的m个方法
>枚举T的集合
>将第一个元素输入第一个方法
>获取输出,异步将其提供给第二个方法
>同时,将集合的第二个元素提供给第一个方法
>完成第一个方法后,将结果提供给第二个方法(如果第二个方法仍在运行,将结果放入其队列并在第一个方法开始执行第三个元素)
>当第二个方法完成执行时从队列中取出第一个元素并执行等等(每个方法都应该异步运行,没有人应该等待下一个完成)
>在第m个方法中,执行数据后,将结果存储到列表中
>在第m个方法完成第n个元素后,将结果列表(n个结果)返回到第一个级别.
我提出了如下代码,但它没有按预期工作,结果永远不会返回,而且它没有按顺序执行.
static class Program
{
static void Main(string[] args)
{
var list = new List<int> { 1, 2, 3, 4 };
var result = list.ForEachPipeline(Add, Square, Add, Square);
foreach (var element in result)
{
Console.WriteLine(element);
Console.WriteLine("---------------------");
}
Console.ReadLine();
}
private static int Add(int j)
{
return j + 1;
}
private static int Square(int j)
{
return j * j;
}
internal static void AddNotify<T>(this List<T> list, T item)
{
Console.WriteLine("Adding {0} to the list", item);
list.Add(item);
}
}
internal class Function<T>
{
private readonly Func<T, T> _func;
private readonly List<T> _result = new List<T>();
private readonly Queue<T> DataQueue = new Queue<T>();
private bool _isBusy;
static readonly object Sync = new object();
readonly ManualResetEvent _waitHandle = new ManualResetEvent(false);
internal Function(Func<T, T> func)
{
_func = func;
}
internal Function<T> Next { get; set; }
internal Function<T> Start { get; set; }
internal int Count;
internal IEnumerable<T> Execute(IEnumerable<T> source)
{
var isSingle = true;
foreach (var element in source) {
var result = _func(element);
if (Next != null)
{
Next.ExecuteAsync(result, _waitHandle);
isSingle = false;
}
else
_result.AddNotify(result);
}
if (!isSingle)
_waitHandle.WaitOne();
return _result;
}
internal void ExecuteAsync(T element, ManualResetEvent resetEvent)
{
lock(Sync)
{
if(_isBusy)
{
DataQueue.Enqueue(element);
return;
}
_isBusy = true;
_func.BeginInvoke(element, CallBack, resetEvent);
}
}
internal void CallBack(IAsyncResult result)
{
bool set = false;
var worker = (Func<T, T>) ((AsyncResult) result).AsyncDelegate;
var resultElement = worker.EndInvoke(result);
var resetEvent = result.AsyncState as ManualResetEvent;
lock(Sync)
{
_isBusy = false;
if(Next != null)
Next.ExecuteAsync(resultElement, resetEvent);
else
Start._result.AddNotify(resultElement);
if(DataQueue.Count > 1)
{
var element = DataQueue.Dequeue();
ExecuteAsync(element, resetEvent);
}
if(Start._result.Count == Count)
set = true;
}
if(set)
resetEvent.Set();
}
}
public static class Pipe
{
public static IEnumerable<T> ForEachPipeline<T>(this IEnumerable<T> source, params Func<T, T>[] pipes)
{
Function<T> start = null, previous = null;
foreach (var function in pipes.Select(pipe => new Function<T>(pipe){ Count = source.Count()}))
{
if (start == null)
{
start = previous = function;
start.Start = function;
continue;
}
function.Start = start;
previous.Next = function;
previous = function;
}
return start != null ? start.Execute(source) : null;
}
}
你能帮助我让这件事有用吗？如果这种设计不适合实际的方法管道,请随意提出不同的方法.
编辑：我必须严格遵守.Net 3.5.
最佳答案 我没有立即在您的代码中发现问题,但您可能会过度复杂化一些事情.这可能是一种更简单的方法来做你想要的.
public static class Pipe
{
public static IEnumerable<T> Execute<T>(
this IEnumerable<T> input, params Func<T, T>[] functions)
{
// each worker will put its result in this array
var results = new T[input.Count()];
// launch workers and return a WaitHandle for each one
var waitHandles = input.Select(
(element, index) =>
{
var waitHandle = new ManualResetEvent(false);
ThreadPool.QueueUserWorkItem(
delegate
{
T result = element;
foreach (var function in functions)
{
result = function(result);
}
results[index] = result;
waitHandle.Set();
});
return waitHandle;
});
// wait for each worker to finish
foreach (var waitHandle in waitHandles)
{
waitHandle.WaitOne();
}
return results;
}
}
这不会像您自己的尝试那样为管道的每个阶段创建锁定.我省略了,因为它似乎没用.但是,你可以通过包装这样的函数轻松添加它：
var wrappedFunctions = functions.Select(x => AddStageLock(x));
AddStageLock是这样的：
private static Func<T,T> AddStageLock<T>(Func<T,T> function)
{
object stageLock = new object();
Func<T, T> wrappedFunction =
x =>
{
lock (stageLock)
{
return function(x);
}
};
return wrappedFunction;
}
编辑：执行实现可能比单线程执行慢,除非为每个单独元素完成的工作使创建等待句柄和在线程池上调度任务的开销相形见绌,真正受益于您需要的多线程限制开销; .NET 4中的PLINQ在partitioning the data之前执行此操作.
