Azure队列工作器角色多线程示例
本文关键字：多线程 角色 队列 工作 Azure | 更新日期: 2023-09-27 18:06:45
我们有4个Azure队列，通过直接REST API或我们提供的WCF服务填充。
- 我们希望有一个worker角色来监控这4个队列。
- 我正在考虑使用多线程，从配置中读取队列名称等，并旋转进程方法(从队列中读取消息并进行处理)
有人能给我提供一个例子或指导如何在一个工人的角色，请实现这一点?
不太确定以上是否可以实现没有多线程，因为我是一个很新的多线程。
谢谢
您可以为不同的任务启动不同的线程，但也可以考虑非线程方法(根据您对消息的处理，其性能可能更好或更差):
while (true)
{
var msg = queue1.GetMessage();
if (msg != null)
{
didSomething = true;
// do something with it
queue1.DeleteMessage(msg);
}
msg = queue2.GetMessage();
if (msg != null)
{
didSomething = true;
// do something with it
queue2.DeleteMessage(msg);
}
// ...
if (!didSomething) Thread.Sleep(TimeSpan.FromSeconds(1)); // so I don't enter a tight loop with nothing to do
}
这是我们目前的实现，以更好的方式(或我们认为)做你所要求的。也就是说，这段代码仍然需要一些繁重的清理。不过，这是它的功能版0.1。
public class WorkerRole : RoleEntryPoint
{
public override void Run()
{
var logic = new WorkerAgent();
logic.Go(false);
}
public override bool OnStart()
{
// Initialize our Cloud Storage Configuration.
AzureStorageObject.Initialize(AzureConfigurationLocation.AzureProjectConfiguration);
return base.OnStart();
}
}
public class WorkerAgent
{
private const int _resistance_to_scaling_larger_queues = 9;
private Dictionary<Type, int> _queueWeights = new Dictionary<Type, int>
{
{typeof (Queue1.Processor), 1},
{typeof (Queue2.Processor), 1},
{typeof (Queue3.Processor), 1},
{typeof (Queue4.Processor), 1},
};
private readonly TimeSpan _minDelay = TimeSpan.FromMinutes(Convert.ToDouble(RoleEnvironment.GetConfigurationSettingValue("MinDelay")));
private readonly TimeSpan _maxDelay = TimeSpan.FromMinutes(Convert.ToDouble(RoleEnvironment.GetConfigurationSettingValue("MaxDelay")));
protected TimeSpan CurrentDelay { get; set; }
public Func<string> GetSpecificQueueTypeToProcess { get; set; }
/// <summary>
/// This is a superset collection of all Queues that this WorkerAgent knows how to process, and the weight of focus it should receive.
/// </summary>
public Dictionary<Type, int> QueueWeights
{
get
{
return _queueWeights;
}
set
{
_queueWeights = value;
}
}
public static TimeSpan QueueWeightCalibrationDelay
{
get { return TimeSpan.FromMinutes(15); }
}
protected Dictionary<Type, DateTime> QueueDelays = new Dictionary<Type, DateTime>();
protected Dictionary<Type, AzureQueueMetaData> QueueMetaData { get; set; }
public WorkerAgent(Func<string> getSpecificQueueTypeToProcess = null)
{
CurrentDelay = _minDelay;
GetSpecificQueueTypeToProcess = getSpecificQueueTypeToProcess;
}
protected IProcessQueues CurrentProcessor { get; set; }
/// <summary>
/// Processes queue request(s).
/// </summary>
/// <param name="onlyProcessOnce">True to only process one time. False to process infinitely.</param>
public void Go(bool onlyProcessOnce)
{
if (onlyProcessOnce)
{
ProcessOnce(false);
}
else
{
ProcessContinuously();
}
}
public void ProcessContinuously()
{
while (true)
{
// temporary hack to get this started.
ProcessOnce(true);
}
}
/// <summary>
/// Attempts to fetch and process a single queued request.
/// </summary>
public void ProcessOnce(bool shouldDelay)
{
PopulateQueueMetaData(QueueWeightCalibrationDelay);
if (shouldDelay)
{
Thread.Sleep(CurrentDelay);
}
var typesToPickFrom = new List<Type>();
foreach(var item in QueueWeights)
{
for (var i = 0; i < item.Value; i++)
{
typesToPickFrom.Add(item.Key);
}
}
var randomIndex = (new Random()).Next()%typesToPickFrom.Count;
var typeToTryAndProcess = typesToPickFrom[randomIndex];
CurrentProcessor = ObjectFactory.GetInstance(typeToTryAndProcess) as IProcessQueues;
CleanQueueDelays();
if (CurrentProcessor != null && !QueueDelays.ContainsKey(typeToTryAndProcess))
{
var errors = CurrentProcessor.Go();
var amountToDelay = CurrentProcessor.NumberProcessed == 0 && !errors.Any()
? _maxDelay // the queue was empty
: _minDelay; // else
QueueDelays[CurrentProcessor.GetType()] = DateTime.Now + amountToDelay;
}
else
{
ProcessOnce(true);
}
}
/// <summary>
/// This method populates/refreshes the QueueMetaData collection.
/// </summary>
/// <param name="queueMetaDataCacheLimit">Specifies the length of time to cache the MetaData before refreshing it.</param>
private void PopulateQueueMetaData(TimeSpan queueMetaDataCacheLimit)
{
if (QueueMetaData == null)
{
QueueMetaData = new Dictionary<Type, AzureQueueMetaData>();
}
var queuesWithoutMetaData = QueueWeights.Keys.Except(QueueMetaData.Keys).ToList();
var expiredQueueMetaData = QueueMetaData.Where(qmd => qmd.Value.TimeMetaDataWasPopulated < (DateTime.Now - queueMetaDataCacheLimit)).Select(qmd => qmd.Key).ToList();
var validQueueData = QueueMetaData.Where(x => !expiredQueueMetaData.Contains(x.Key)).ToList();
var results = new Dictionary<Type, AzureQueueMetaData>();
foreach (var queueProcessorType in queuesWithoutMetaData)
{
if (!results.ContainsKey(queueProcessorType))
{
var queueProcessor = ObjectFactory.GetInstance(queueProcessorType) as IProcessQueues;
if (queueProcessor != null)
{
var queue = new AzureQueue(queueProcessor.PrimaryQueueName);
var metaData = queue.GetMetaData();
results.Add(queueProcessorType, metaData);
QueueWeights[queueProcessorType] = (metaData.ApproximateMessageCount) == 0
? 1
: (int)Math.Log(metaData.ApproximateMessageCount, _resistance_to_scaling_larger_queues) + 1;
}
}
}
foreach (var queueProcessorType in expiredQueueMetaData)
{
if (!results.ContainsKey(queueProcessorType))
{
var queueProcessor = ObjectFactory.GetInstance(queueProcessorType) as IProcessQueues;
if (queueProcessor != null)
{
var queue = new AzureQueue(queueProcessor.PrimaryQueueName);
var metaData = queue.GetMetaData();
results.Add(queueProcessorType, metaData);
}
}
}
QueueMetaData = results.Union(validQueueData).ToDictionary(data => data.Key, data => data.Value);
}
private void CleanQueueDelays()
{
QueueDelays = QueueDelays.Except(QueueDelays.Where(x => x.Value < DateTime.Now)).ToDictionary(x => x.Key, x => x.Value);
}
}
_queueWeights集合。我们设置_resistance_to_scaling_larger_queues常数来控制我们想要的比例。请注意，这是以对数方式缩放的(参见PopulateQueueMetaData方法)。没有队列的权重小于1，即使它有0个条目。如果将PopulateQueueMetaData设置为10，则每增加10个数量级，该类型的"权重"就增加1。例如，如果您有包含0项的QueueA、包含0项的QueueB和包含10项的QueueC，那么您各自的权重分别为1、1和2。这意味着QueueC有50%的机会被处理，而QueueA和QueueB各自只有25%的机会被处理。如果QueueC有100个条目，那么权重为1,1,3，被处理的几率为20%，20%，60%。这可以确保您的空队列不会被遗忘。
它做的另一件事是它有_minDelay和_maxDelay。如果这段代码认为队列中至少有一个项目，那么它将继续以_minDelay的速度处理它。但是，如果它最后有0个条目，那么它将不允许它的处理速度比_maxDelay更快。因此，这意味着如果随机数生成器拉出具有0项的队列(无论权重如何)，它将直接跳过尝试处理它并继续进行下一次迭代。(为了提高存储事务效率，可以对这部分进行一些额外的优化，但这只是一个简单的补充。)
我们在这里有几个自定义类(比如AzureQueue和AzureQueueMetaData)——一个本质上是CloudQueue的包装器，另一个存储一些信息，比如队列的大概计数——没有什么有趣的(只是一种简化代码的方法)。
最后，像这样写这段代码允许我们有一个可以处理更多队列的项目。如果我们发现这根本跟不上，我们可以轻松地将其扩展到更大数量的实例，并扩展到所有队列。在最小的场景中，您可以部署一个实例来监视3个队列。但是，如果第4个队列开始影响性能(或者您需要更高的可用性)，则最多增加到2个实例。一旦你排了15个队，再加第三个。25个队列，添加第4个实例。获得一个新客户，并且需要跨系统处理许多队列请求，这很好。将这个角色旋转到20个实例，直到它完成，然后再将它们旋转回来。排队的队伍特别烦人?在_queueWeights集合中注释队列，部署以管理您的队列的其余部分，然后再次与所有其他队列重新部署它，除了这个在_queueWeights集合中注释的队列，然后再次将其部署到一组不同的实例并进行调试，而无需a)让其他QueueProcessors干扰您的调试和b)您的调试干扰其他QueueProcessors。最终，这提供了很大的灵活性和效率。
在worker角色的while循环中，启动4个线程，就像你正在编写多线程c#应用程序一样。当然，您需要定义四个不同的线程函数，并且这些函数应该有单独的while循环来轮询队列。在worker的while循环结束时，只需等待线程完成。