首页 文章

Azure辅助角色多线程队列处理

提问于
浏览
3

我有一个具有经典配置WebRole Worker角色的azure Cloud服务 . Worker从队列中获取消息,处理它而不是一次删除一次 .

我的代码是这样的:

public override void Run()
    {
        Trace.TraceInformation("Worker is running");
            try
            {
                this.RunAsync(this.cancellationTokenSource.Token).Wait();
            }
            finally
            {
                this.runCompleteEvent.Set();
            }
    }

public override bool OnStart()
        {
            ServicePointManager.DefaultConnectionLimit = 500;
            bool result = base.OnStart();
            Trace.TraceInformation("WorkerAnalytics has been started");
            return result;
        }



private async Task RunAsync(CancellationToken cancellationToken)
        {
            var queue = ....//omitted info for brevity
            CloudQueueMessage retrievedMessage = null;

            while (!cancellationToken.IsCancellationRequested)
            {
                 try
                    {
                        retrievedMessage = await queue.GetMessageAsync();
                        if (retrievedMessage != null)
                        {
                            await ProcessMessage(retrievedMessage);
                        }
                        else
                        {
                            System.Threading.Thread.Sleep(500);
                        }
                    }
                    catch (Exception e)
                    {
                        System.Threading.Thread.Sleep(500);
                    }
                }
            }

        }

现在这个工作完美但CPU很低,为3%,它一次只处理一个元素(每个大约1秒),但队列每秒大约有1000个新元素,这还不够 .

How can process more queue messages at a time using all the CPU power the machine has and without complicating too much this code?

Also what the ServicePointManager.DefaultConnectionLimit is for?

我搜索了几个小时的工作者角色的有效多线程解决方案,但现在所有的WebJobs或旧的框架使事情变得复杂 .

谢谢

1 回答

  • 5

    您可以尝试运行多次 RunAsync() .

    var tasks = new List<Task>();
    tasks.Add(this.RunAsync(this.cancellationTokenSource.Token));
    tasks.Add(this.RunAsync(this.cancellationTokenSource.Token));
    tasks.Add(this.RunAsync(this.cancellationTokenSource.Token));
    Task.WaitAll(tasks.ToArray());
    

相关问题