当前位置：首页 > 资讯 > 技术文档

简单实现C#异步操作

时间：2020-09-10 11:46 编辑：来源：阅读：
扫一扫，手机访问

摘要：简单实现C#异步操作

在.net4.0以后异步操作,并行计算变得异常简单,但是由于公司项目开发基于.net3.5所以无法用到4.0的并行计算以及Task等异步编程。因此,为了以后更方便的进行异步方式的开发,我封装实现了异步编程框架,通过BeginInvoke、EndInvoke的方式实现异步编程。 [b]一、框架结构[/b] 整个框架包括四个部分 1、基类抽象Opeartor 我把每个异步执行过程称为一个Operate,因此需要一个Opeartor去执行 2、FuncAsync 异步的Func 3、ActionAsync 异步的Action 4、Asynchorus 对ActionAsync和FuncAsync的封装 [b]Operator [/b]Operator是一个抽象类,实现了IOperationAsync和IContinueWithAsync两个接口。 IOperationAsync实现了异步操作,IContinueWithAsync实现了类似于Task的ContinueWith方法,在当前异步操作完成后继续进行的操作 IOperationAsync接口详解

public interface IOperationAsync
{
  IAsyncResult Invoke();
  void Wait();
  void CompletedCallBack(IAsyncResult ar);
  void CatchException(Exception exception);
}

[list] [*][b]Invoke():异步方法的调用[/b][/*] [*][b]Wait():等待异步操作执行[/b][/*] [*][b]CompletedCallBack():操作完成回调[/b][/*] [*][b]CatchException():抓取异常 [/b][/*] [/list] IContinueWithAsync接口详情

public interface IContinueWithAsync
{
  Operator Previous { get; set; }
  Operator Next { get; set; }
  Operator ContinueWithAsync(Action action);
  Operator ContinueWithAsync<TParameter>(Action<TParameter> action, TParameter parameter);
}

Previous:前一个操作 Next:下一个操作 ContinueWithAsync():异步继续操作

public abstract class Operator : IOperationAsync, IContinueWithAsync
{
  public IAsyncResult Middle;
  public readonly string Id;
  public Exception Exception { get; private set; }
  public Operator Previous { get; set; }
  public Operator Next { get; set; }
  protected Operator()
  {
    Id = Guid.NewGuid().ToString();
  }
  public abstract IAsyncResult Invoke();
  protected void SetAsyncResult(IAsyncResult result)
  {
    this.Middle = result;
  }
  public virtual void Wait()
  {
    if (!Middle.IsCompleted) Middle.AsyncWaitHandle.WaitOne();
  }
  public virtual void CompletedCallBack(IAsyncResult ar)
  {
  }
  public void CatchException(Exception exception)
  {
    this.Exception = exception;
  }
  protected Operator ContinueAsync()
  {
    if (Next != null) Next.Invoke();
    return Next;
  }
  public virtual Operator ContinueWithAsync(Action action)
  {
    Next = new ActionAsync(action);
    Next.Previous = this;
    return Next;
  }
  public virtual Operator ContinueWithAsync<TParameter>(Action<TParameter> action, TParameter parameter)
  {
    Next = new ActionAsync<TParameter>(action, parameter);
    Next.Previous = this;
    return Next;
  }
  public virtual Operator ContinueWithAsync<TResult>(Func<TResult> func)
  {
    Next = new FuncAsync<TResult>();
    Next.Previous = this;
    return Next;
  }
  public virtual Operator ContinueWithAsync<TParameter, TResult>(Func<TParameter, TResult> func,
    TParameter parameter)
  {
    Next = new FuncAsync<TParameter, TResult>(func, parameter);
    Next.Previous = this;
    return Next;
  }
}

[b]无返回异步操作 [/b]ActionAsync

public class ActionAsync : Operator
{
  private readonly Action _action;
  protected ActionAsync()
  {
  }
  public ActionAsync(Action action)
    : this()
  {
    this._action = action;
  }
  public override IAsyncResult Invoke()
  {
    var middle = _action.BeginInvoke(CompletedCallBack, null);
    SetAsyncResult(middle);
    return middle;
  }
  public override void CompletedCallBack(IAsyncResult ar)
  {
    try
    {
      _action.EndInvoke(ar);
    }
    catch (Exception exception)
    {
      this.CatchException(exception);
    }
    ContinueAsync();
  }
}
public class ActionAsync<T> : ActionAsync
{
  public T Result;
  private readonly Action<T> _action1;
  protected readonly T Parameter1;
  public ActionAsync()
  {
  }
  public ActionAsync(T parameter)
  {
    this.Parameter1 = parameter;
  }
  public ActionAsync(Action<T> action, T parameter)
  {
    this._action1 = action;
    this.Parameter1 = parameter;
  }
  public override IAsyncResult Invoke()
  {
    var result = _action1.BeginInvoke(Parameter1, CompletedCallBack, null);
    SetAsyncResult(result);
    return result;
  }
  public override void CompletedCallBack(IAsyncResult ar)
  {
    try
    {
      _action1.EndInvoke(ar);
    }
    catch (Exception exception)
    {
      this.CatchException(exception);
    }
    ContinueAsync();
  }
}

[b]有返回异步 [/b]FuncAsync实现了IFuncOperationAsync接口 IFuncOperationAsync

public interface IFuncOperationAsync<T>
{
  void SetResult(T result);
  T GetResult();
}

[list] [*][b]SetResult(T result):异步操作完成设置返回值[/b][/*] [*][b]GetResult():获取返回值[/b][/*] [/list] 1）、FuncAsync

public class FuncAsync<TResult> : Operator, IFuncOperationAsync<TResult>
{
private TResult _result;

public TResult Result
{
  get
  {
    if (!Middle.IsCompleted || _result == null)
    {
      _result = GetResult();
    }
    return _result;
  }
}
private readonly Func<TResult> _func1;
public FuncAsync()
{
}
public FuncAsync(Func<TResult> func)
{
  this._func1 = func;
}
public override IAsyncResult Invoke()
{
  var result = _func1.BeginInvoke(CompletedCallBack, null);
  SetAsyncResult(result);
  return result;
}
public override void CompletedCallBack(IAsyncResult ar)
{
  try
  {
    var result = _func1.EndInvoke(ar);
    SetResult(result);
  }
  catch (Exception exception)
  {
    this.CatchException(exception);
    SetResult(default(TResult));
  }
  ContinueAsync();
}
public virtual TResult GetResult()
{
  Wait();
  return this._result;
}
public void SetResult(TResult result)
{
  _result = result;
}
}
public class FuncAsync<T1, TResult> : FuncAsync<TResult>
{
protected readonly T1 Parameter1;
private readonly Func<T1, TResult> _func2;
public FuncAsync(Func<T1, TResult> action, T1 parameter1)
  : this(parameter1)
{
  this._func2 = action;
}
protected FuncAsync(T1 parameter1)
  : base()
{
  this.Parameter1 = parameter1;
}
public override IAsyncResult Invoke()
{
  var result = _func2.BeginInvoke(Parameter1, CompletedCallBack, null);
  SetAsyncResult(result);
  return result;
}
public override void CompletedCallBack(IAsyncResult ar)
{
  try
  {
    var result = _func2.EndInvoke(ar);
    SetResult(result);
  }
  catch (Exception exception)
  {
    CatchException(exception);
    SetResult(default(TResult));
  }
  ContinueAsync();
}
}

[b]Asynchronous 异步操作封装 [/b]ActionAsync和FuncAsync为异步操作打下了基础,接下来最重要的工作就是通过这两个类执行我们的异步操作,为此我封装了一个异步操作类主要封装了以下几个部分： [list] [*][b]WaitAll(IEnumerable<Operator> operations):等待所有操作执行完毕[/b][/*] [*][b]WaitAny(IEnumerable<Operator> operations):等待任意操作执行完毕[/b][/*] [*][b]ActionAsync[/b][/*] [*][b]FuncAsync[/b][/*] [*][b]ContinueWithAction[/b][/*] [*][b]ContinueWithFunc[/b][/*] [/list] 后面四个包含若干个重载,这里只是笼统的代表一个类型的方法 [b]WaitAll [/b]

public static void WaitAll(IEnumerable<Operator> operations)
{
foreach (var @operator in operations)
{
  @operator.Wait();
}
}

WaitAny

public static void WaitAny(IEnumerable<Operator> operations)
{
while (operations.All(o => !o.Middle.IsCompleted))
  Thread.Sleep(100);
}

等待时间可以自定义 ActionInvoke

public static Operator Invoke(Action action)
{
Operator operation = new ActionAsync(action);
operation.Invoke();
return operation;
}
public static Operator Invoke<T>(Action<T> action, T parameter)
{
Operator operation = new ActionAsync<T>(action, parameter);
operation.Invoke();
return operation;
}
public static Operator Invoke<T1, T2>(Action<T1, T2> action, T1 parameter1, T2 parameter2)
{
Operator operation = new ActionAsync<T1, T2>(action, parameter1, parameter2);
operation.Invoke();
return operation;
}

FuncInvoke

public static Operator Invoke<TResult>(Func<TResult> func)
{
Operator operation = new FuncAsync<TResult>(func);
operation.Invoke();
return operation;
}
public static Operator Invoke<TParameter, TResult>(Func<TParameter, TResult> func, TParameter parameter)
{
TParameter param = parameter;
Operator operation = new FuncAsync<TParameter, TResult>(func, param);
operation.Invoke();
return operation;
}
public static Operator Invoke<T1, T2, TResult>(Func<T1, T2, TResult> func, T1 parameter1, T2 parameter2)
{
Operator operation = new FuncAsync<T1, T2, TResult>(func, parameter1, parameter2);
operation.Invoke();
return operation;
}

ContinueWithAction

public static Operator ContinueWithAsync(IEnumerable<Operator>operators, Action action)
{
return Invoke(WaitAll, operators)
  .ContinueWithAsync(action);
}
public static Operator ContinueWithAsync<TParameter>(IEnumerable<Operator> operators, Action<TParameter> action, TParameter parameter)
{
return Invoke(WaitAll, operators)
  .ContinueWithAsync(action, parameter);
}

ContinueWithFunc

public static Operator ContinueWithAsync<TResult>(IEnumerable<Operator> operators,Func<TResult> func)
{
return Invoke(WaitAll, operators)
  .ContinueWithAsync(func);
}
public static Operator ContinueWithAsync<TParameter, TResult>(IEnumerable<Operator> operators, 
Func<TParameter, TResult> func, TParameter parameter)
{
return Invoke(WaitAll, operators)
  .ContinueWithAsync(func, parameter);
}

这里有个bug当调用ContinueWithAsync后无法调用Wait等待,本来Wait需要从前往后等待每个异步操作,但是测试了下不符合预期结果。不过理论上来说应该无需这样操作,ContinueWithAsync只是为了当上一个异步操作执行完毕时继续执行的异步操作,若要等待,那不如两个操作放到一起,最后再等待依然可以实现。前面的都是单步异步操作的调用,若需要对某集合进行某个方法的异步操作,可以foreach遍历

public void ForeachAsync(IEnumerbale<string> parameters)
{
  foreach(string p in parameters)
  {
    Asynchronous.Invoke(Tast,p);
  }
}
public void Test(string parameter)
{
  //TODO:做一些事
}

每次都需要去手写foreach,比较麻烦,因此实现类似于PLinq的并行计算方法实在有必要,不过有一点差别,PLinq是采用多核CPU进行并行计算,而我封装的仅仅遍历集合进行异步操作而已 ForeachAction

public static IEnumerable<Operator> Foreach<TParameter>(IEnumerable<TParameter> items, Action<TParameter> action)
{
  return items.Select(t => Invoke(action, t)).ToList();
}

ForeachFunc

public static IEnumerable<Operator> Foreach<TParameter, TResult>(IEnumerable<TParameter> items, Func<TParameter, TResult> func)
{
  return items.Select(parameter => Invoke(func, parameter)).ToList();
}

如何使用 [b]无返回值异步方法调用[/b]

public void DoSomeThing()
{
//TODO:
}

通过Asynchronous.Invoke(DoSomeThing) 执行

public void DoSomeThing(string parameter)
{
//TODO:
}

通过Asynchronous.Invoke(DoSomeThing,parameter) 执行 [b]有返回值异步方法调用[/b]

public string DoSomeThing()
{
//TODO:
}

通过Asynchronous.Invoke(()=>DoSomeThing())执行

public string DoSomeThing(string parameter)
{
//TODO:
}

通过Asynchronous.Invoke(()=>DoSomeThing(parameter))执行,或者也可以传入参数通过Asynchronous.Invoke(p=>DoSomeThing(p),parameter) [b]无返回值Foreach[/b]

public void Test
{
int[] parameters = {1,2,3,4,5};
Asynchronous.Foreach(parameters,Console.WriteLine);
}

[b]有返回值Foreach[/b]

public void Test
{
int[] parameters = {1,2,3,4,5};
var operators = Asynchronous.Foreach(parameters,p=> p*2);
Asynchrous.WaitAll(operators);
Asynchronous.Foreach(operators.Cast<FuncAsync<int,int>>(),
  p=> Console.WriteLine(p.Result));
}

首先将集合每个值扩大2倍,然后输出 [b]异步执行完再执行[/b]

public void Test
{
int[] parameters = {1,2,3,4,5};
var operators = Asynchronous.Foreach(parameters,p=> p*2);
Asynchrous.ContinueWithAsync(operators,Console.WriteLine,"执行完成");
}

[b]每次执行完继续执行 [/b]可能有时候我们需要遍历一个集合,每个元素处理完成后我们需要输出XX处理完成

public void Test
{
int[] parameters = {1,2,3,4,5};
var operators = Asynchronous.Foreach(parameters,p=> p*2);
Asynchronous.Foreach(operators,o=>{
  o.ContinueWithAsync(()={
    //每个元素执行完时执行
    if(o.Exception != null)
    {
      //之前执行时产生未处理的异常,这里可以捕获到 
    }
  });
});
}

[b]可以实现链式异步操作[/b]

public void Chain()
{
Asynchronous.Invoke(Console.WriteLine,1)
.ContinueWithAsync(Console.WriteLine,2)
.ContinueWithAsync(Console.WriteLine,3)
}

这样会按步骤输出1,2,3 结束语以上只是列出了部分重载方法，其他重载方法无非就是加参数，本质实际是一样的。希望对大家的学习有所帮助，在这祝大家新年快乐，新的一年大家一起努力。

全部评论(0)

上一篇：C#使用HttpWebRequest与HttpWebResponse模拟用户登录
下一篇：C#操作XML通用方法汇总

资讯排行榜
更多>>