Tạo một nhiệm vụ với nhịp tim

Question

Tôi muốn chạy một Task có "heartbeat" tiếp tục chạy tại một khoảng thời gian cụ thể cho đến khi tác vụ hoàn tất.

Tôi đang nghĩ đến một phương pháp mở rộng như thế này sẽ làm việc tốt:

public static async Task WithHeartbeat(this Task primaryTask, TimeSpan heartbeatInterval, Action<CancellationToken> heartbeatAction, CancellationToken cancellationToken) 

Ví dụ:

public class Program { 
    public static void Main() { 
     var cancelTokenSource = new CancellationTokenSource(); 
     var cancelToken = cancelTokenSource.Token; 
     var longRunningTask = Task.Factory.StartNew(SomeLongRunningTask, cancelToken, TaskCreationOptions.LongRunning, TaskScheduler.Current); 
     var withHeartbeatTask = longRunningTask.WithHeartbeat(TimeSpan.FromSeconds(1), PerformHeartbeat, cancelToken); 
     withHeartbeatTask.Wait(); 
     Console.WriteLine("Long running task completed!"); 
     Console.ReadLine() 
    } 

    private static void SomeLongRunningTask() { 
     Console.WriteLine("Starting long task"); 
     Thread.Sleep(TimeSpan.FromSeconds(9.5)); 
    } 
    private static int _heartbeatCount = 0; 
    private static void PerformHeartbeat(CancellationToken cancellationToken) { 
     Console.WriteLine("Heartbeat {0}", ++_heartbeatCount); 
    } 
} 

Chương trình này nên đầu ra:

Starting long task 
Heartbeat 1 
Heartbeat 2 
Heartbeat 3 
Heartbeat 4 
Heartbeat 5 
Heartbeat 6 
Heartbeat 7 
Heartbeat 8 
Heartbeat 9 
Long running task completed! 

Lưu ý rằng nó không nên (trong trường hợp bình thường) đầu ra "Heartbeat 10" vì nhịp tim bắt đầu sau thời gian chờ ban đầu (tức là 1 giây). Tương tự, nếu tác vụ mất ít thời gian hơn khoảng nhịp tim, thì nhịp tim sẽ không xảy ra chút nào.

Cách tốt để thực hiện điều này là gì?

Thông tin cơ bản: Tôi có dịch vụ đang nghe hàng đợi Azure Service Bus. Tôi không muốn Complete thông báo (sẽ xóa vĩnh viễn thư đó khỏi hàng đợi) cho đến khi tôi xử lý xong, có thể mất nhiều thời gian hơn thông báo tối đa LockDuration trong 5 phút. Vì vậy, tôi cần sử dụng cách tiếp cận nhịp tim này để gọi RenewLockAsync trước khi thời hạn khóa hết hạn để thông báo không hết thời gian chờ trong khi xử lý kéo dài.

Answer 1

Đây là nỗ lực của tôi:

public static class TaskExtensions { 
    /// <summary> 
    /// Issues the <paramref name="heartbeatAction"/> once every <paramref name="heartbeatInterval"/> while <paramref name="primaryTask"/> is running. 
    /// </summary> 
    public static async Task WithHeartbeat(this Task primaryTask, TimeSpan heartbeatInterval, Action<CancellationToken> heartbeatAction, CancellationToken cancellationToken) { 
     if (cancellationToken.IsCancellationRequested) { 
      return; 
     } 

     var stopHeartbeatSource = new CancellationTokenSource(); 
     cancellationToken.Register(stopHeartbeatSource.Cancel); 

     await Task.WhenAny(primaryTask, PerformHeartbeats(heartbeatInterval, heartbeatAction, stopHeartbeatSource.Token)); 
     stopHeartbeatSource.Cancel(); 
    } 

    private static async Task PerformHeartbeats(TimeSpan interval, Action<CancellationToken> heartbeatAction, CancellationToken cancellationToken) { 
     while (!cancellationToken.IsCancellationRequested) { 
      try { 
       await Task.Delay(interval, cancellationToken); 
       if (!cancellationToken.IsCancellationRequested) { 
        heartbeatAction(cancellationToken); 
       } 
      } 
      catch (TaskCanceledException tce) { 
       if (tce.CancellationToken == cancellationToken) { 
        // Totally expected 
        break; 
       } 
       throw; 
      } 
     } 
    } 
} 

hoặc với một tinh chỉnh nhẹ, bạn thậm chí có thể làm cho nhịp tim async như trong:

/// <summary> 
    /// Awaits a fresh Task created by the <paramref name="heartbeatTaskFactory"/> once every <paramref name="heartbeatInterval"/> while <paramref name="primaryTask"/> is running. 
    /// </summary> 
    public static async Task WithHeartbeat(this Task primaryTask, TimeSpan heartbeatInterval, Func<CancellationToken, Task> heartbeatTaskFactory, CancellationToken cancellationToken) { 
     if (cancellationToken.IsCancellationRequested) { 
      return; 
     } 

     var stopHeartbeatSource = new CancellationTokenSource(); 
     cancellationToken.Register(stopHeartbeatSource.Cancel); 

     await Task.WhenAll(primaryTask, PerformHeartbeats(heartbeatInterval, heartbeatTaskFactory, stopHeartbeatSource.Token)); 

     if (!stopHeartbeatSource.IsCancellationRequested) { 
      stopHeartbeatSource.Cancel(); 
     } 
    } 

    public static Task WithHeartbeat(this Task primaryTask, TimeSpan heartbeatInterval, Func<CancellationToken, Task> heartbeatTaskFactory) { 
     return WithHeartbeat(primaryTask, heartbeatInterval, heartbeatTaskFactory, CancellationToken.None); 
    } 

    private static async Task PerformHeartbeats(TimeSpan interval, Func<CancellationToken, Task> heartbeatTaskFactory, CancellationToken cancellationToken) { 
     while (!cancellationToken.IsCancellationRequested) { 
      try { 
       await Task.Delay(interval, cancellationToken); 
       if (!cancellationToken.IsCancellationRequested) { 
        await heartbeatTaskFactory(cancellationToken); 
       } 
      } 
      catch (TaskCanceledException tce) { 
       if (tce.CancellationToken == cancellationToken) { 
        // Totally expected 
        break; 
       } 
       throw; 
      } 
     } 
    } 

mà sẽ cho phép bạn thay đổi mẫu mã để một cái gì đó như này:

private static async Task PerformHeartbeat(CancellationToken cancellationToken) { 
    Console.WriteLine("Starting heartbeat {0}", ++_heartbeatCount); 
    await Task.Delay(1000, cancellationToken); 
    Console.WriteLine("Finishing heartbeat {0}", _heartbeatCount); 
} 

PerformHeartbeat có thể được thay thế bằng một cuộc gọi không đồng bộ như RenewLockAsync để bạn không phải lãng phí thời gian của luồng bằng cách sử dụng lệnh gọi chặn như RenewLock mà cách tiếp cận Hành động sẽ yêu cầu.

Tôi là answering my own question per SO guidelines, nhưng tôi cũng mở ra các cách tiếp cận trang nhã hơn cho vấn đề này.

Answer 2

Dưới đây là cách tiếp cận của tôi

using System; 
using System.Threading; 
using System.Threading.Tasks; 

namespace ConsoleApplication3 
{ 
class Program 
{ 
    static void Main(string[] args) 
    { 
     Console.WriteLine("Start Main"); 
     StartTest().Wait(); 
     Console.ReadLine(); 
     Console.WriteLine("Complete Main"); 
    } 

    static async Task StartTest() 
    { 
     var cts = new CancellationTokenSource(); 

     // ***Use ToArray to execute the query and start the download tasks. 
     Task<bool>[] tasks = new Task<bool>[2]; 
     tasks[0] = LongRunningTask("", 20, cts.Token); 
     tasks[1] = Heartbeat("", 1, cts.Token); 

     // ***Call WhenAny and then await the result. The task that finishes 
     // first is assigned to firstFinishedTask. 
     Task<bool> firstFinishedTask = await Task.WhenAny(tasks); 

     Console.WriteLine("first task Finished."); 
     // ***Cancel the rest of the downloads. You just want the first one. 
     cts.Cancel(); 

     // ***Await the first completed task and display the results. 
     // Run the program several times to demonstrate that different 
     // websites can finish first. 
     var isCompleted = await firstFinishedTask; 
     Console.WriteLine("isCompleted: {0}", isCompleted); 
    } 

    private static async Task<bool> LongRunningTask(string id, int sleep, CancellationToken ct) 
    { 
     Console.WriteLine("Starting long task"); 


     await Task.Delay(TimeSpan.FromSeconds(sleep)); 

     Console.WriteLine("Completed long task"); 
     return true; 
    } 

    private static async Task<bool> Heartbeat(string id, int sleep, CancellationToken ct) 
    { 
     while(!ct.IsCancellationRequested) 
     { 
      await Task.Delay(TimeSpan.FromSeconds(sleep)); 
      Console.WriteLine("Heartbeat Task Sleep: {0} Second", sleep); 
     } 

     return true; 
    } 

} 

}