分享一个异步任务在遇到IO异常时支持递归回调的辅助方法

简介:
复制代码
public void TryAsyncActionRecursively<TAsyncResult>(
    string asyncActionName,
    Func<Task<TAsyncResult>> asyncAction,
    Action<int> mainAction,
    Action<TAsyncResult> successAction,
    Func<string> getContextInfoFunc,
    Action<Exception> failedAction,
    int retryTimes) where TAsyncResult : AsyncOperationResult
{
    var retryAction = new Action<int>(currentRetryTimes =>
    {
        if (currentRetryTimes >= _immediatelyRetryTimes)
        {
            Task.Factory.StartDelayedTask(_retryIntervalForIOException, () => mainAction(currentRetryTimes + 1));
        }
        else
        {
            mainAction(currentRetryTimes + 1);
        }
    });
    var executeFailedAction = new Action<Exception>(ex =>
    {
        try
        {
            if (failedAction != null)
            {
                failedAction(ex);
            }
        }
        catch (Exception unknownEx)
        {
            _logger.Error(string.Format("Failed to execute the failedCallbackAction of asyncAction:{0}, contextInfo:{1}",
                asyncActionName, getContextInfoFunc()), unknownEx);
        }
    });
    var processTaskException = new Action<Exception, int>((ex, currentRetryTimes) =>
    {
        if (ex is IOException)
        {
            _logger.Error(string.Format("Async task '{0}' has io exception, contextInfo:{1}, current retryTimes:{2}",
                asyncActionName, getContextInfoFunc(), currentRetryTimes), ex);
            retryAction(retryTimes);
        }
        else
        {
            _logger.Error(string.Format("Async task '{0}' has unknown exception, contextInfo:{1}, current retryTimes:{2}",
                asyncActionName, getContextInfoFunc(), currentRetryTimes), ex);
            executeFailedAction(ex);
        }
    });
    var completeAction = new Action<Task<TAsyncResult>>(t =>
    {
        if (t.Exception != null)
        {
            processTaskException(t.Exception.InnerException, retryTimes);
            return;
        }
        if (t.IsCanceled)
        {
            _logger.ErrorFormat("Async task '{0}' was cancelled, contextInfo:{1}, current retryTimes:{2}",
                asyncActionName, getContextInfoFunc(), retryTimes);
            retryAction(retryTimes);
            return;
        }
        var result = t.Result;
        if (result.Status == AsyncOperationResultStatus.IOException)
        {
            _logger.ErrorFormat("Async task '{0}' has io exception, contextInfo:{1}, current retryTimes:{2}, errorMsg:{3}",
                asyncActionName, getContextInfoFunc(), retryTimes, result.ErrorMessage);
            retryAction(retryTimes);
            return;
        }
        if (successAction != null)
        {
            successAction(result);
        }
    });

    try
    {
        asyncAction().ContinueWith(completeAction);
    }
    catch (IOException ex)
    {
        _logger.Error(string.Format("Execute async action '{0}' failed, contextInfo:{1}, current retryTimes:{2}",
            asyncActionName, getContextInfoFunc(), retryTimes), ex);
        retryAction(retryTimes);
    }
    catch (Exception ex)
    {
        _logger.Error(string.Format("Execute async action '{0}' failed, contextInfo:{1}, current retryTimes:{2}",
            asyncActionName, getContextInfoFunc(), retryTimes), ex);
        executeFailedAction(ex);
    }
}
复制代码

该函数的功能是:执行一个异步任务(返回Task的方法),如果执行出现IO异常,则重试当前主函数(mainAction);用户的mainAction中会再次调用TryAsyncActionRecursively方法。从而实现当遇到IO异常时,能做到不断重试。另外,重试只立即重试指定的次数,超过指定次数,则不立即重试,而是暂停一定间隔后再次执行。该函数还提供当acyncAction执行成功或失败后的回调函数,以及允许传入当前上下文的一些说明信息,以便记录有意义的错误日志信息。

下面是使用示例:

复制代码
private void PublishEventAsync(ProcessingCommand processingCommand, EventStream eventStream, int retryTimes)
{
    TryAsyncActionRecursively<AsyncOperationResult>("PublishEventAsync",
    () => _eventPublisher.PublishAsync(eventStream),
    currentRetryTimes => PublishEventAsync(processingCommand, eventStream, currentRetryTimes),
    result =>
    {
        _logger.DebugFormat("Publish events success, {0}", eventStream);
        processingCommand.Complete(new CommandResult(CommandStatus.Success, processingCommand.Command.Id));
    },
    () => string.Format("[eventStream:{0}]", eventStream),
    ex => processingCommand.Complete(new CommandResult(CommandStatus.Failed, processingCommand.Command.Id)),
    retryTimes);
}
复制代码
PublishEventAsync(processingCommand, eventStream, 0);

目录
相关文章
|
2月前
|
并行计算 数据处理 Python
Python并发编程迷雾:IO密集型为何偏爱异步?CPU密集型又该如何应对?
在Python的并发编程世界中,没有万能的解决方案,只有最适合特定场景的方法。希望本文能够为你拨开迷雾,找到那条通往高效并发编程的光明大道。
43 2
|
3月前
|
开发框架 并行计算 算法
揭秘Python并发神器:IO密集型与CPU密集型任务的异步革命,你竟还傻傻分不清?
揭秘Python并发神器:IO密集型与CPU密集型任务的异步革命,你竟还傻傻分不清?
47 4
|
2月前
|
监控 并行计算 数据处理
构建高效Python应用:并发与异步编程的实战秘籍,IO与CPU密集型任务一网打尽!
在Python编程的征途中,面对日益增长的性能需求,如何构建高效的应用成为了每位开发者必须面对的课题。并发与异步编程作为提升程序性能的两大法宝,在处理IO密集型与CPU密集型任务时展现出了巨大的潜力。今天,我们将深入探讨这些技术的最佳实践,助你打造高效Python应用。
40 0
|
2月前
|
搜索推荐 索引
【文件IO】实现:查找文件并删除、文件复制、递归遍历目录查找文件
【文件IO】实现:查找文件并删除、文件复制、递归遍历目录查找文件
36 2
|
3月前
|
安全 Java API
【Java面试题汇总】Java基础篇——String+集合+泛型+IO+异常+反射(2023版)
String常量池、String、StringBuffer、Stringbuilder有什么区别、List与Set的区别、ArrayList和LinkedList的区别、HashMap底层原理、ConcurrentHashMap、HashMap和Hashtable的区别、泛型擦除、ABA问题、IO多路复用、BIO、NIO、O、异常处理机制、反射
【Java面试题汇总】Java基础篇——String+集合+泛型+IO+异常+反射(2023版)
|
2月前
|
开发框架 并行计算 .NET
脑洞大开!Python并发与异步编程的哲学思考:IO密集型与CPU密集型任务的智慧选择!
脑洞大开!Python并发与异步编程的哲学思考:IO密集型与CPU密集型任务的智慧选择!
32 1
|
3月前
|
开发框架 并行计算 .NET
燃烧吧,Python!异步编程如何点燃IO密集型任务,让CPU密集型任务也加速狂奔?
燃烧吧,Python!异步编程如何点燃IO密集型任务,让CPU密集型任务也加速狂奔?
25 2
|
3月前
|
算法 Java 程序员
解锁Python高效之道:并发与异步在IO与CPU密集型任务中的精准打击策略!
在数据驱动时代,高效处理大规模数据和高并发请求至关重要。Python凭借其优雅的语法和强大的库支持,成为开发者首选。本文将介绍Python中的并发与异步编程,涵盖并发与异步的基本概念、IO密集型任务的并发策略、CPU密集型任务的并发策略以及异步IO的应用。通过具体示例,展示如何使用`concurrent.futures`、`asyncio`和`multiprocessing`等库提升程序性能,帮助开发者构建高效、可扩展的应用程序。
127 0
|
3月前
|
UED 开发者 Python
Python并发编程新纪元:异步编程如何重塑IO与CPU密集型任务的处理方式?
在Python编程中,异步编程作为一种非阻塞模式,通过允许程序在等待IO操作时继续执行其他任务,提高了程序的响应性和吞吐量。与传统同步编程相比,它减少了线程等待时间,尤其在处理IO密集型任务时表现出色,如使用`asyncio`库进行异步HTTP请求。尽管对CPU密集型任务的直接提升有限,但结合多进程或多线程可间接提高效率。异步编程虽强大,但也带来了代码复杂度增加和调试难度提升等挑战,需要开发者掌握最佳实践来克服这些问题。随着其技术的成熟,异步编程正在逐步改变我们处理IO与CPU密集型任务的方式,成为提升性能和优化用户体验的重要工具。
26 0
|
4月前
|
NoSQL Java Redis
【Azure Spring Cloud】Java Spring Cloud 应用部署到Azure上后,发现大量的 java.lang.NullPointerException: null at io.lettuce.core.protocol.CommandHandler.writeSingleCommand(CommandHandler.java:426) at ... 异常
【Azure Spring Cloud】Java Spring Cloud 应用部署到Azure上后,发现大量的 java.lang.NullPointerException: null at io.lettuce.core.protocol.CommandHandler.writeSingleCommand(CommandHandler.java:426) at ... 异常