程序師世界是廣大編程愛好者互助、分享、學習的平台,程序師世界有你更精彩!
首頁
編程語言
C語言|JAVA編程
Python編程
網頁編程
ASP編程|PHP編程
JSP編程
數據庫知識
MYSQL數據庫|SqlServer數據庫
Oracle數據庫|DB2數據庫
 程式師世界 >> 編程語言 >> C語言 >> 關於C語言 >> 簡單實現C#異步操作

簡單實現C#異步操作

編輯:關於C語言

在.net4.0以後異步操作,並行計算變得異常簡單,但是由於公司項目開發基於.Net3.5所以無法用到4.0的並行計算以及Task等異步編程。因此,為了以後更方便的進行異步方式的開發,我封裝實現了異步編程框架,通過BeginInvoke、EndInvoke的方式實現異步編程。

一、框架結構

整個框架包括四個部分

1、基類抽象Opeartor
我把每個異步執行過程稱為一個Operate,因此需要一個Opeartor去執行
2、FuncAsync
異步的Func
3、ActionAsync
異步的Action
4、Asynchorus
對ActionAsync和FuncAsync的封裝

Operator
Operator是一個抽象類,實現了IOperationAsync和IContinueWithAsync兩個接口。
IOperationAsync實現了異步操作,IContinueWithAsync實現了類似於Task的ContinueWith方法,在當前異步操作完成後繼續進行的操作

IOperationAsync接口詳解

? 1 2 3 4 5 6 7 public interface IOperationAsync { IAsyncResult Invoke(); void Wait(); void CompletedCallBack(IAsyncResult ar); void CatchException(Exception exception); }
  • Invoke():異步方法的調用
  • Wait():等待異步操作執行
  • CompletedCallBack():操作完成回調
  • CatchException():抓取異常

IContinueWithAsync接口詳情

? 1 2 3 4 5 6 7 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():異步繼續操作

? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 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; } }

無返回異步操作
ActionAsync

? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 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(); } }

有返回異步
FuncAsync實現了IFuncOperationAsync接口

IFuncOperationAsync

? 1 2 3 4 5 public interface IFuncOperationAsync<T> { void SetResult(T result); T GetResult(); }
  • SetResult(T result):異步操作完成設置返回值
  • GetResult():獲取返回值

1)、FuncAsync

? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 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(); } }

Asynchronous 異步操作封裝
ActionAsync和FuncAsync為異步操作打下了基礎,接下來最重要的工作就是通過這兩個類執行我們的異步操作,為此我封裝了一個異步操作類
主要封裝了以下幾個部分:

  • WaitAll(IEnumerable<Operator> Operations):等待所有操作執行完畢
  • WaitAny(IEnumerable<Operator> Operations):等待任意操作執行完畢
  • ActionAsync
  • FuncAsync
  • ContinueWithAction
  • ContinueWithFunc

後面四個包含若干個重載,這裡只是籠統的代表一個類型的方法
WaitAll

? 1 2 3 4 5 6 7 public static void WaitAll(IEnumerable<Operator> Operations) { foreach (var @Operator in Operations) { @Operator.Wait(); } }

WaitAny

? 1 2 3 4 5 public static void WaitAny(IEnumerable<Operator> Operations) { while (Operations.All(o => !o.Middle.IsCompleted)) Thread.Sleep(100); }

等待時間可以自定義
ActionInvoke

? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 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

? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 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

? 1 2 3 4 5 6 7 8 9 10 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

? 1 2 3 4 5 6 7 8 9 10 11 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遍歷

? 1 2 3 4 5 6 7 8 9 10 11 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

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

ForeachFunc

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

如何使用
無返回值異步方法調用

? 1 2 3 4 public void DOSomeThing() { //TODO: }

通過Asynchronous.Invoke(DOSomeThing) 執行

? 1 2 3 4 public void DOSomeThing(string parameter) { //TODO: }

通過Asynchronous.Invoke(DOSomeThing,parameter) 執行

有返回值異步方法調用

? 1 2 3 4 public string DOSomeThing() { //TODO: }

通過Asynchronous.Invoke(()=>DOSomeThing())執行

? 1 2 3 4 public string DOSomeThing(string parameter) { //TODO: }

通過Asynchronous.Invoke(()=>DoSomeThing(parameter))執行,或者也可以傳入參數通過Asynchronous.Invoke(p=>DOSomeThing(p),parameter)

無返回值Foreach

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

有返回值Foreach

? 1 2 3 4 5 6 7 8 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倍,然後輸出
異步執行完再執行

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

每次執行完繼續執行
可能有時候我們需要遍歷一個集合,每個元素處理完成後我們需要輸出XX處理完成

? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 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) { //之前執行時產生未處理的異常,這裡可以捕獲到 } }); }); }

可以實現鏈式異步操作

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

這樣會按步驟輸出1,2,3
結束語

以上只是列出了部分重載方法,其他重載方法無非就是加參數,本質實際是一樣的。

希望對大家的學習有所幫助,在這祝大家新年快樂,新的一年大家一起努力。

  1. 上一頁:
  2. 下一頁:
Copyright © 程式師世界 All Rights Reserved