zoukankan      html  css  js  c++  java
  • 实现更简单的异步操作

    前言

    在.net4.0以后异步操作,并行计算变得异常简单,但是由于公司项目开发基于.net3.5所以无法用到4.0的并行计算以及Task等异步编程。因此,为了以后更方便的进行异步方式的开发,我封装实现了异步编程框架,通过BeginInvoke、EndInvoke的方式实现异步编程。

    框架结构

    整个框架包括四个部分

    1. 基类抽象Opeartor
      异步操作的基类,实现了异步操作接口
    2. FuncAsync
      异步的Func
    3. ActionAsync
      异步的Action
    4. Asynchorus
      对ActionAsync和FuncAsync的封装

    Operator

    Operator是一个抽象类,实现了IOperationAsyncIContinueWithAsync两个接口。
    IOperationAsync实现了异步操作,IContinueWithAsync实现了类似于Task的ContinueWith方法,在当前异步操作完成后继续进行的操作

    IOperationAsync接口详解

        public interface IOperationAsync
        {
            IAsyncResult Invoke();
            void Wait();
            void CompletedCallBack(IAsyncResult ar);
            void CatchException(Exception exception);
        }
    
    • Invoke():异步方法的调用
    • Wait():等待异步操作执行
    • CompletedCallBack():操作完成回调
    • CatchException():抓取异常

    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;
        }
    }
    

    无返回异步操作

    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();
        }
    }
    

    有返回异步

    FuncAsync实现了IFuncOperationAsync接口

    IFuncOperationAsync

    public interface IFuncOperationAsync<T>
    {
        void SetResult(T result);
        T GetResult();
    }
    
    • SetResult(T result):异步操作完成设置返回值
    • GetResult():获取返回值

    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();
        }
    }
    

    Asynchronous 异步操作封装

    ActionAsync和FuncAsync为异步操作打下了基础,接下来最重要的工作就是通过这两个类执行我们的异步操作,为此我封装了一个异步操作类
    主要封装了以下几个部分:

    1. WaitAll(IEnumerable<Operator> operations):等待所有操作执行完毕
    2. WaitAny(IEnumerable<Operator> operations):等待任意操作执行完毕
    3. ActionAsync
    4. FuncAsync
    5. ContinueWithAction
    6. ContinueWithFunc

    后面四个包含若干个重载,这里只是笼统的代表一个类型的方法

    WaitAll

    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();
    }
    

    如何使用

    • 无返回值异步方法调用
    public void DoSomeThing()
    {
        //TODO:
    }
    

    通过Asynchronous.Invoke(DoSomeThing) 执行

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

    通过Asynchronous.Invoke(DoSomeThing,parameter) 执行

    • 有返回值异步方法调用
    public string DoSomeThing()
    {
        //TODO:
    }
    

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

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

    通过Asynchronous.Invoke(()=>DoSomeThing(parameter))执行,或者也可以传入参数通过Asynchronous.Invoke(p=>DoSomeThing(p),parameter)

    • 无返回值Foreach
    public void Test
    {
        int[] parameters = {1,2,3,4,5};
        Asynchronous.Foreach(parameters,Console.WriteLine);
    }
    
    • 有返回值Foreach
    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倍,然后输出

    • 异步执行完再执行
    public void Test
    {
        int[] parameters = {1,2,3,4,5};
        var operators = Asynchronous.Foreach(parameters,p=> p*2);
        Asynchrous.ContinueWithAsync(operators,Console.WriteLine,"执行完成");
    }
    
    • 每次执行完继续执行
      可能有时候我们需要遍历一个集合,每个元素处理完成后我们需要输出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)
                {
                    //之前执行时产生未处理的异常,这里可以捕获到  
                }
            });
        });
    }
    
    • 可以实现链式异步操作
    public void Chain()
    {
        Asynchronous.Invoke(Console.WriteLine,1)
        .ContinueWithAsync(Console.WriteLine,2)
        .ContinueWithAsync(Console.WriteLine,3)
    }
    

    这样会按步骤输出1,2,3

    结束语

    以上只是列出了部分重载方法,其他重载方法无非就是加参数,本质实际是一样的,具体可以已提交至github。我还封装了for方法,但是感觉没什么用,而且也从没有用到过,这里不再提
    通过以上的封装,已经能完成日常大部分的操作,调用还是比较方便的。

    本文发布至作业部落

  • 相关阅读:
    Android Studio环境搭建
    JavaScript、Ajax与jQuery的关系
    Creational模式之Builder模式
    数据结构--二叉查找树的java实现
    ShopEx文章页添加上一篇下一篇功能
    WinCE的C#中使用StreamReader 来读取TXT文档,读取文本文档。
    关于对FLASH开发,starling、starling feathers、starling MVC框架的理解
    C语言实现牛顿迭代法解方程
    ios的notification机制是同步的还是异步的
    XCL-Chart柱形图的期望线/分界线
  • 原文地址:https://www.cnblogs.com/Jack-Blog/p/5182310.html
Copyright © 2011-2022 走看看