如何编写长时间运行的活动以在 WF 4.0 中调用 Web 服务

Question

我创建了一个执行 Web 请求并将结果存储到数据库中的活动。我发现对于这些长时间运行的活动，我应该编写一些不同的代码，这样工作流引擎线程就不会被阻塞。

public sealed class WebSaveActivity : NativeActivity
{
    protected override void Execute(NativeActivityContext context)
    {
       GetAndSave(); // This takes 1 hour to accomplish.
    }
}

我应该如何重写这个活动以满足长期运行活动的要求

Answer 1

您可以在现有进程中生成一个线程，例如使用ThreadPool.QueueUserWorkItem() 因此，如果需要，您的工作流程的其余部分将继续运行。不过，请务必先了解多线程和线程同步的含义。 或者，您可以查看 Hangfire 或类似组件，将整个作业卸载到不同的进程中。

编辑：

根据您的评论，您可以查看基于任务的异步模式 (TAP)：Link 1、Link 2，这将为您提供一个很好的代码编写模型，可以继续处理可以在等待时完成的事情您长时间运行的操作的结果，直到它返回。但是，我不确定这是否满足您的所有需求。特别是在 Windows Workflow Foundation 中，您可能希望查看某种形式的 workflow hibernation/persistence。

Answer 2

这种情况是使用 WF 的持久性功能大放异彩的地方。它允许您将工作流实例持久化到数据库中，以允许完成一些长时间运行的操作。完成后，第二个线程或进程可以重新填充工作流实例并允许它恢复。

首先您为工作流应用程序指定一个工作流实例存储。 Microsoft 提供了您可以使用的SQL workflow instance store implementation，并提供了您可以在 SQL Server 上运行的 SQL 脚本。

namespace MySolution.MyWorkflowApp
{
    using System.Activities;
    using System.Activities.DurableInstancing;
    using System.Activities.Statements;
    using System.Threading;

    internal static class Program
    {
        internal static void Main(string[] args)
        {
            var autoResetEvent = new AutoResetEvent(false);
            var workflowApp = new WorkflowApplication(new Sequence());
            workflowApp.InstanceStore = new SqlWorkflowInstanceStore("server=mySqlServer;initial catalog=myWfDb;...");
            workflowApp.Completed += e => autoResetEvent.Set();
            workflowApp.Unloaded += e => autoResetEvent.Set();
            workflowApp.Aborted += e => autoResetEvent.Set();
            workflowApp.Run();
            autoResetEvent.WaitOne();
        }
    }
}

您的活动将启动一个实际执行保存操作的辅助进程/线程。有多种方法可以做到这一点：

• 在辅助线程上
• 通过异步调用 Web 方法，该方法实际上完成了执行保存操作的繁重工作

您的活动将如下所示：

public sealed class WebSaveActivity : NativeActivity
{
    public InArgument<MyBigObject> ObjectToSave { get; set; }

    protected override bool CanInduceIdle
    {
        get
        {
            // This notifies the WF engine that the activity can be unloaded / persisted to an instance store.
            return true;
        }
    }

    protected override void Execute(NativeActivityContext context)
    {
        var currentBigObject = this.ObjectToSave.Get(context);
        currentBigObject.WorkflowInstanceId = context.WorkflowInstanceId;
        StartSaveOperationAsync(this.ObjectToSave.Get(context)); // This method should offload the actual save process to a thread or even a web method, then return immediately.

        // This tells the WF engine that the workflow instance can be suspended and persisted to the instance store.
        context.CreateBookmark("MySaveOperation", AfterSaveCompletesCallback);
    }

    private void AfterSaveCompletesCallback(NativeActivityContext context, Bookmark bookmark, object value)
    {
        // Do more things after the save completes.
        var saved = (bool) value;
        if (saved)
        {
            // yay!
        }
        else
        {
            // boo!!!
        }
    }
}

书签创建向 WF 引擎发出信号，表明工作流实例可以从内存中卸载，直到有东西唤醒工作流实例。

在您的方案中，您希望在长时间保存操作完成后恢复工作流程。让我们假设StartSaveOperationAsync 方法将一条小消息写入某种队列，第二个线程或进程会轮询以执行保存操作：

public static void StartSaveOperationAsync(MyBigObject myObjectToSave)
{
    var targetQueue = new MessageQueue(".\private$\pendingSaveOperations");
    var message = new Message(myObjectToSave);
    targetQueue.Send(message);
}

在我的第二个过程中，我可以轮询队列以获取新的保存请求并重新填充持久化的工作流实例，以便它可以在保存操作完成后恢复。假设以下方法在不同的控制台应用程序中：

internal static void PollQueue()
{
    var targetQueue = new MessageQueue(@".\private$\pendingSaveOperations");
    while (true)
    {
        // This waits for a message to arrive on the queue.
        var message = targetQueue.Receive();
        var myObjectToSave = message.Body as MyBigObject;

        // Perform the long running save operation
        LongRunningSave(myObjectToSave);

        // Once the save operation finishes, you can resume the associated workflow.
        var autoResetEvent = new AutoResetEvent(false);
        var workflowApp = new WorkflowApplication(new Sequence());
        workflowApp.InstanceStore = new SqlWorkflowInstanceStore("server=mySqlServer;initial catalog=myWfDb;...");
        workflowApp.Completed += e => autoResetEvent.Set();
        workflowApp.Unloaded += e => autoResetEvent.Set();
        workflowApp.Aborted += e => autoResetEvent.Set();

        // I'm assuming the object to save has a field somewhere that refers the workflow instance that's running it.
        workflowApp.Load(myObjectToSave.WorkflowInstanceId);
        workflowApp.ResumeBookmark("LongSaveOperation", true); // The 'true' parameter is just our way of saying the save completed successfully. You can use any object type you desire here.
        autoResetEvent.WaitOne();
    }
}

private static void LongRunningSave(object myObjectToSave)
{
    throw new NotImplementedException();
}

public class MyBigObject 
{
    public Guid WorkflowInstanceId { get; set; } = Guid.NewGuid();
}

现在长时间运行的保存操作不会阻碍工作流引擎，并且不会将工作流实例长时间保存在内存中，从而更有效地利用系统资源。