Azure 服务结构消息队列

Question

我正在尝试使用 Azure Service Fabric 对一系列任务进行排队，然后运行它们异步进行。我目前正在使用具有辅助角色的 CloudMessageQueue。我正在尝试迁移到 Service Fabric。从工作者角色，这是我的代码：

    private void ExecuteTask()
    {
        CloudQueueMessage message = null;

        if (queue == null)
        {
            jmaLogProvider.WriteToLog(new ErrorMessage(MessageSeverity.Error, String.Format("Queue for WorkerRole2 is null. Exiting.")));
            return;
        }

        try
        {
            message = queue.GetMessage();
            if (message != null)
            {
                JMATask task = GetTask(message.AsString);
                string msg = (message == null) ? string.Empty : message.AsString;
                //jmaLogProvider.WriteToLog(new ErrorMessage(MessageSeverity.JMA, String.Format("Executing task {0}", msg)));
                queue.DeleteMessage(message);
                PerformTask(task);
            }
        }
        catch (Exception ex)
        {
            string msg = (message == null) ? string.Empty : message.AsString;
            jmaLogProvider.WriteToLog(new ErrorMessage(MessageSeverity.Error, String.Format("Message {0} Error removing message from queue {1}", msg, ex.ToString())));
        }
    }

我有一些问题：

如何运行异步执行任务方法？我想同时运行大约 30 - 40 个任务。
我有一个 JMATask 列表。如何将列表添加到队列？

列表是否需要加入队列？

namespace Stateful1
{
   public class JMATask
   {
     public string Name { get; set; }
   }

/// <summary>
/// An instance of this class is created for each service replica by the Service Fabric runtime.
/// </summary>
internal sealed class Stateful1 : StatefulService
{
public Stateful1(StatefulServiceContext context)
    : base(context)
{ }

/// <summary>
/// Optional override to create listeners (e.g., HTTP, Service Remoting, WCF, etc.) for this service replica to handle client or user requests.
/// </summary>
/// <remarks>
/// For more information on service communication, see http://aka.ms/servicefabricservicecommunication
/// </remarks>
/// <returns>A collection of listeners.</returns>
protected override IEnumerable<ServiceReplicaListener> CreateServiceReplicaListeners()
{
    return new ServiceReplicaListener[0];
}

/// <summary>
/// This is the main entry point for your service replica.
/// This method executes when this replica of your service becomes primary and has write status.
/// </summary>
/// <param name="cancellationToken">Canceled when Service Fabric needs to shut down this service replica.</param>
protected override async Task RunAsync(CancellationToken cancellationToken)
{
    // TODO: Replace the following sample code with your own logic 
    //       or remove this RunAsync override if it's not needed in your service.

    IReliableQueue<JMATask> tasks = await this.StateManager.GetOrAddAsync<IReliableQueue<JMATask>>("JMATasks");
    //var myDictionary = await this.StateManager.GetOrAddAsync<IReliableDictionary<string, long>>("myDictionary");

    while (true)
    {
        cancellationToken.ThrowIfCancellationRequested();

        using (var tx = this.StateManager.CreateTransaction())
        {
            var result = await tasks.TryDequeueAsync(tx);

            //how do I execute this method async?
            PerformTask(result.Value);

            //Create list of JMA Tasks to queue up
            await tasks.EnqueueAsync(tx, new JMATask());

            //var result = await myDictionary.TryGetValueAsync(tx, "Counter");

            //ServiceEventSource.Current.ServiceMessage(this, "Current Counter Value: {0}",
            //    result.HasValue ? result.Value.ToString() : "Value does not exist.");

            //await myDictionary.AddOrUpdateAsync(tx, "Counter", 0, (key, value) => ++value);

            // If an exception is thrown before calling CommitAsync, the transaction aborts, all changes are 
            // discarded, and nothing is saved to the secondary replicas.
            await tx.CommitAsync();
        }

        await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken);
    }
}

private async void PerformTask(JMATask task)
{
    //execute task
}

}

Answer 1

您可以创建一个任务列表，然后执行 await Task.WhenAll(taskList);

这可能是最简单的直接答案。

但是 - 如果每个任务都略有不同，您是否考虑过为每个任务创建单独的微服务？

Answer 2

RunAsync 方法不应有此行代码：await tasks.EnqueueAsync(tx, new JMATask());

创建要排队的 JMA 任务列表应该是有状态服务中的另一种方法，如下所示：

    public async Task AddJMATaskAsync(JMATask task)
    {
        var tasksQueue = await StateManager.GetOrAddAsync<IReliableQueue<JMATask>>("JMATasks");
        using (var tx = StateManager.CreateTransaction())
        {
            try
            {
                await tasksQueue.EnqueueAsync(tx, request);
                await tx.CommitAsync();
            }
            catch (Exception ex)
            {
                tx.Abort();
            }
        }
    }

然后您的 PerformTask 方法可以包含对无状态微服务的调用：

    public async Task PerformTask (JMATask task)
    {
        //1. resolve stateless microservice URI
        // statelessSvc

        //2. call method of the stateless microservice
        // statelessSvc.PerformTask(task);
    }

所以基本上，有状态服务只会对任务进行排队和出队。执行实际任务可以通过微服务来完成，该微服务将对集群中的所有节点可用。

Azure 服务结构消息队列

Azure Service Fabric Message Queue

c#

azure

azure-service-fabric