在 C# 中为同步/异步任务添加重试/回滚机制的最佳方法是啥？

Posted 2023-03-31

【中文标题】在 C# 中为同步/异步任务添加重试/回滚机制的最佳方法是啥？

【英文标题】：Which is the best way to add a retry/rollback mechanism for sync/async tasks in C#?在 C# 中为同步/异步任务添加重试/回滚机制的最佳方法是什么？

【发布时间】：2016-09-20 12:02:11

【问题描述】：

想象一个 WebForms 应用程序，其中有一个名为 CreateAll() 的主方法。我可以将方法任务的过程逐步描述如下：

1) 存储到数据库（更新/创建 Db 项 3-4 次）

2) 启动一个新线程

3) Result1 = 调用soap服务，并通过使用超时阈值检查状态并在x分钟后继续（现在状态正常，并不意味着失败）

4) 存储到数据库（更新/创建 Db 项 3-4 次）

5) result2 = 调用一个肥皂服务（以一种火而忘记的方式）

6) 更新配置文件（实际上取自 result1）

7) 通过使用回调请求，它每 x 秒在前面检查一次结果的状态2，并且 UI 显示一个进度条。如果过程完成 (100%)，则表示成功

我认为所有这些都是可以按类型分组的任务。基本上这几种类型的操作是：

类型1：数据库事务 Type2：服务通信/事务 Type3：配置文件 I/O 事务

我想在现有实现中添加回滚/重试机制，并使用面向任务的架构并重构现有的遗留代码。

我发现 C# 中的 Memento Design Pattern 或 Command Pattern 可以帮助实现此目的。我还发现了 msdn Retry Pattern描述有趣。我真的不知道，我希望有人引导我做出最安全、最好的决定...

您能否建议我在这种情况下保持现有实现和流程但将其包装在通用和抽象的重试/回滚/任务列表实现中的最佳方法？

最终实现必须能够在每种情况下重试（无论任务或一般故障，例如在整个一般 createAll 过程中的超时等），并且还有一个回滚决策列表，其中应用必须能够回滚所有任务完成了。

我想要一些如何破解这个耦合代码的例子。

---

可能有用的伪代码：

class something
  
    static result CreateAll(object1 obj1, object2 obj2 ...)
    
        //Save to database obj1
        //...
        //Update to database obj1 
        //
        //NEW THREAD
       //Start a new thread with obj1, obj2 ...CreateAll
       //...          
      

    void CreateAllAsync()
    
        //Type1 Save to database obj1
        //...
        //Type1 Update to database obj2

        //Type2 Call Web Service to create obj1 on the service (not async)

        while (state != null && now < times)
        
            if (status == "OK")
            break;      
            else
            //Wait for X seconds
        

        //Check status continue or general failure
        //Type1 Update to database obj2 and obj1

        //Type2 Call Web Service to create obj2 on the service (fire and forget)

        //Type3 Update Configuration File
        //Type1 Update to database obj2 and obj1
        //..   

    return;


//Then the UI takes the responsibility to check the status of result2

【问题讨论】：

您可以查看 MS 模式和实践中的瞬态故障处理 msdn.microsoft.com/en-us/library/hh675232.aspx 它可以扩展以解决其他异常，但同时处理异步和非异步。

重试委托的通用重试机制是否适合您？

你推荐哪一个？我正在尝试将命令模式（将现有代码分解为特定任务）与重试策略的 Polly 框架相结合

@CharlesNRice 你能举一个瞬态故障处理应用框架的例子吗？

【参考方案1】：

查看使用Polly 的重试场景，这似乎与您的伪代码非常吻合。此答案的末尾是文档中的示例。您可以执行各种重试方案，重试和等待等。例如，您可以多次重试一个完整的事务，或者多次重试一组幂等操作，然后在重试时编写补偿逻辑政策最终失败。

备忘录模式更适合您在文字处理器（Ctrl-Z 和 Ctrl-Y）中找到的撤消重做逻辑。

其他有用的模式是简单队列、持久队列甚至是服务总线，它们可以为您提供最终的一致性，而无需让用户等待一切成功完成。

// Retry three times, calling an action on each retry 
// with the current exception and retry count
Policy
    .Handle<DivideByZeroException>()
    .Retry(3, (exception, retryCount) =>
    
        // do something 
    );

基于您的伪代码的示例可能如下所示：

static bool CreateAll(object1 obj1, object2 obj2)

     // Policy to retry 3 times, waiting 5 seconds between retries.
     var policy =
         Policy
              .Handle<SqlException>()
              .WaitAndRetry(3, count =>
              
                 return TimeSpan.FromSeconds(5); 
              );

       policy.Execute(() => UpdateDatabase1(obj1));
       policy.Execute(() => UpdateDatabase2(obj2));
  

【讨论】：

看来 OP 没有重试，他 polling 该服务。

@Evk - 请参阅问题中的第 3 点和第 7 点。需要“守卫”来不断重试以查看该过程，但是整个过程可以位于更大的轮询类型场景中以按计划运行。为此，我会考虑使用 Quartz.Net quartz-scheduler.net

3 和 7 在我看来都像是在轮询，我认为这与“重试失败”模式不同。只需给 OP 做一点说明，以防这真的是轮询（问题不是很清楚）。

你能帮我把代码分解成任务或其他东西（我符合命令模式）吗？示例代码会很棒。Poly 确实很棒。其他人也提出了重构技术。你也可以帮助我吗？谢谢

我已根据您的 CreateAll 伪代码在我的答案中添加了一个示例。当您说任务时，您是指单个调用（例如，我已将策略重新用于多个操作），还是您正在考虑并行工作（您可以使用 Parallel.ForEach( ...)) 或异步操作，您可以在其中创建委托/回调以在后台线程上的工作完成时通知您？

【参考方案2】：

您可以选择命令模式，其中每个命令都包含所有必要的信息，例如连接字符串、服务 url、重试次数等。在此之上，您可以考虑使用rx，数据流块来做管道。

高层观点:

更新：意图是分离关注点。重试逻辑仅限于一个类，该类是现有命令的包装器。 您可以进行更多分析并提出适当的命令、调用者和接收者对象并添加回滚功能。

public abstract class BaseCommand

    public abstract RxObservables Execute();


public class DBCommand : BaseCommand

    public override RxObservables Execute()
    
        return new RxObservables();
    


public class WebServiceCommand : BaseCommand

    public override RxObservables Execute()
    
        return new RxObservables();
    


public class ReTryCommand : BaseCommand // Decorator to existing db/web command

    private readonly BaseCommand _baseCommand;
    public RetryCommand(BaseCommand baseCommand)
    
         _baseCommand = baseCommand
    
    public override RxObservables Execute()
    
        try
        
            //retry using Polly or Custom
            return _baseCommand.Execute();
        
        catch (Exception)
        
            throw;
        
    


public class TaskDispatcher

    private readonly BaseCommand _baseCommand;
    public TaskDispatcher(BaseCommand baseCommand)
    
        _baseCommand = baseCommand;
    

    public RxObservables ExecuteTask()
    
        return _baseCommand.Execute();
    


public class Orchestrator

    public void Orchestrate()
    
        var taskDispatcherForDb = new TaskDispatcher(new ReTryCommand(new DBCommand));
        var taskDispatcherForWeb = new TaskDispatcher(new ReTryCommand(new WebCommand));
        var dbResultStream = taskDispatcherForDb.ExecuteTask();
        var WebResultStream = taskDispatcherForDb.ExecuteTask();
    

【讨论】：

您能更具体地介绍一下代码示例吗？特别是如何包装一般的重试机制。

@谢谢你的更新。最后一个是给我一个与其他人一样的重试实现的例子。你有替代方案吗？那太好了。提前致谢！

重试将或多或少与其他人解释的在同一行，可能一直是异步的。

重试为DB和WebService？ 2 RxObservables（dbResultStream 和 WebResultStream）。您对示例代码中的 2 个 流 什么都不做。

【参考方案3】：

对我来说，这听起来像是“分布式事务”，因为您拥有不同的资源（数据库、服务通信、文件 i/o）并且想要进行可能涉及所有这些的事务。

在 C# 中，您可以使用 Microsoft Distributed Transaction Coordinator 解决此问题。对于每个资源，您都需要一个资源管理器。据我所知，对于数据库，如 sql server 和文件 i/o，它已经可用。对于其他人，您可以开发自己的。

例如，要执行这些事务，您可以像这样使用TransactionScope 类：

using (TransactionScope ts = new TransactionScope())

    //all db code here

    // if an error occurs jump out of the using block and it will dispose and rollback

    ts.Complete();

（示例取自here）

要开发您自己的资源管理器，您必须实现IEnlistmentNotification，这可能是一项相当复杂的任务。这是一个简短的example。

【讨论】：

也许这对 I/O 操作有好处，但所有其他回滚情况也会有一些逻辑。所以它只涵盖了我的范式的一部分。你能扩展一下吗？谢谢

特别是对于服务部分，您首先可能需要检查 protokoll，以便它“启用事务”。当您调用肥皂服务时，如果发生超时，您可以实现回滚。由于您只是在阅读信息，因此这应该有效。所以一个相对简单的资源管理器就可以完成这项工作。

【参考方案4】：

一些可以帮助您实现目标的代码。

public static class Retry

   public static void Do(
       Action action,
       TimeSpan retryInterval,
       int retryCount = 3)
   
       Do<object>(() => 
       
           action();
           return null;
       , retryInterval, retryCount);
   

   public static T Do<T>(
       Func<T> action, 
       TimeSpan retryInterval,
       int retryCount = 3)
   
       var exceptions = new List<Exception>();

       for (int retry = 0; retry < retryCount; retry++)
       
          try
           
              if (retry > 0)
                  Thread.Sleep(retryInterval);
              return action();
          
          catch (Exception ex)
           
              exceptions.Add(ex);
          
       

       throw new AggregateException(exceptions);
   

如下调用并重试：

int result = Retry.Do(SomeFunctionWhichReturnsInt, TimeSpan.FromSeconds(1), 4);

参考：http://gist.github.com/KennyBu/ac56371b1666a949daf8

【讨论】：

那么，例如，您是否更喜欢您的代码而不是 Polly？我会试试你的例子。

刚看了Polly，它的实现更复杂，但我觉得有点开销。我会说如果一段简单的代码可以实现我们的目标，为什么我们需要引入额外的代码。但是，这完全取决于您的决定。

这不是抄袭。我在 GitHub 和 *** 中都阅读了这段代码。 gist.github.com/KennyBu/ac56371b1666a949daf8我不知道谁是原始代码所有者，但它是在 GitHub 的 Gnu 许可下。虽然我的帖子中应该有参考，但我认为这不是强制性的。

如果从网上复制代码是抄袭，我们都要坐牢很久很久。

如果从另一个 SO 帖子中复制整个答案不是抄袭，那么什么是抄袭？

【参考方案5】：

嗯...听起来是一个非常非常糟糕的情况。你不能打开一个事务，写一些东西到数据库，然后去公园遛狗。因为事务有这种为每个人锁定资源的讨厌习惯。这消除了您的最佳选择：分布式事务。

我会执行所有操作并准备一个反向脚本。如果操作成功，我会清除脚本。否则我会运行它。但这对潜在的陷阱是开放的，脚本必须准备好处理它们。例如 - 如果在中途有人已经更新了您添加的记录怎么办；还是根据您的值计算汇总？

仍然：构建逆向脚本是简单的解决方案，没有火箭科学。只是

List<Command> reverseScript;

然后，如果您需要回滚：

using (TransactionScope tx= new TransactionScope()) 
    foreach(Command cmd in reverseScript) cmd.Execute();
    tx.Complete();