Azure 服务 Bus/Service 结构消息未从队列中删除
Azure Service Bus/Service Fabric message not being removed from queue
我正在将一个计算引擎从 Azure 辅助角色移动到 Azure Service Fabric。它通过侦听服务总线上的消息来工作,然后根据消息内容进行处理。
目前,计算工作正常,但如果计算时间超过一分钟左右,则消息在完成后不会从队列中删除。在 worker 角色中,我们通过增加 "AutoRenewTimeout" 来解决这个问题。
var options = new OnMessageOptions { AutoComplete = true, AutoRenewTimeout = TimeSpan.FromMinutes(3) };
_queueClient.OnMessage(OnMessage, options);
但是,使用 "ServiceFabric.ServiceBus" nuget package, I cannot work out where you would set this. I have used the demo project 作为参考来设置实际运行计算的无状态服务。以下是 CalculateService.cs 中无状态服务初始化的摘录。
internal sealed class CalculateService : StatelessService
{
public CalculateService(StatelessServiceContext context)
: base(context)
{ }
/// <summary>
/// Optional override to create listeners (e.g., TCP, HTTP) for this service replica to handle client or user requests.
/// </summary>
/// <returns>A collection of listeners.</returns>
protected override IEnumerable<ServiceInstanceListener> CreateServiceInstanceListeners()
{
string serviceBusQueueName = CloudConfigurationManager.GetSetting("QueueName");
yield return new ServiceInstanceListener(context => new ServiceBusQueueCommunicationListener(new Handler(this), context, serviceBusQueueName), "StatelessService-ServiceBusQueueListener");
}
}
internal sealed class Handler : AutoCompleteServiceBusMessageReceiver
{
protected override Task ReceiveMessageImplAsync(BrokeredMessage message, CancellationToken cancellationToken)
{
ServiceEventSource.Current.ServiceMessage(_service, $"Handling queue message {message.MessageId}");
var computeRole = new ExcelCompute();
var rMessage = new RangeMessage();
rMessage = message.GetBody<RangeMessage>();
var result = computeRole.OnMessage(rMessage, message.MessageId); //returns true if the compute was successful (which it currently, always is)
return Task.FromResult(result);
}
}
我确实尝试使用 <BrokeredMessage> message.Complete();
,但那会引发消息锁定错误。
- 获取最新版本的包(>= v3.5.0)。将
CommunicationListener
上的 属性 'MessageLockRenewTimeSpan
' 设置为小于锁定持续时间的值。 (例如,锁定持续时间为 60 秒的 50 秒)这允许一些时钟偏差。注意:默认情况下这个属性是空的,这意味着没有自动更新锁。
当处理的批处理时间超过锁定持续时间所允许的时间时,此选项效果很好。
或
- 当处理时间超过锁定持续时间时,您可以使用 BrokeredMessage.RenewLock 定期延长消息锁定。如果需要,您可以在单独的线程中执行此操作。
May be useful info
当处理单个消息(批量大小 1)花费的时间超过锁定持续时间允许的时间时,此选项很有效。
我正在将一个计算引擎从 Azure 辅助角色移动到 Azure Service Fabric。它通过侦听服务总线上的消息来工作,然后根据消息内容进行处理。
目前,计算工作正常,但如果计算时间超过一分钟左右,则消息在完成后不会从队列中删除。在 worker 角色中,我们通过增加 "AutoRenewTimeout" 来解决这个问题。
var options = new OnMessageOptions { AutoComplete = true, AutoRenewTimeout = TimeSpan.FromMinutes(3) };
_queueClient.OnMessage(OnMessage, options);
但是,使用 "ServiceFabric.ServiceBus" nuget package, I cannot work out where you would set this. I have used the demo project 作为参考来设置实际运行计算的无状态服务。以下是 CalculateService.cs 中无状态服务初始化的摘录。
internal sealed class CalculateService : StatelessService
{
public CalculateService(StatelessServiceContext context)
: base(context)
{ }
/// <summary>
/// Optional override to create listeners (e.g., TCP, HTTP) for this service replica to handle client or user requests.
/// </summary>
/// <returns>A collection of listeners.</returns>
protected override IEnumerable<ServiceInstanceListener> CreateServiceInstanceListeners()
{
string serviceBusQueueName = CloudConfigurationManager.GetSetting("QueueName");
yield return new ServiceInstanceListener(context => new ServiceBusQueueCommunicationListener(new Handler(this), context, serviceBusQueueName), "StatelessService-ServiceBusQueueListener");
}
}
internal sealed class Handler : AutoCompleteServiceBusMessageReceiver
{
protected override Task ReceiveMessageImplAsync(BrokeredMessage message, CancellationToken cancellationToken)
{
ServiceEventSource.Current.ServiceMessage(_service, $"Handling queue message {message.MessageId}");
var computeRole = new ExcelCompute();
var rMessage = new RangeMessage();
rMessage = message.GetBody<RangeMessage>();
var result = computeRole.OnMessage(rMessage, message.MessageId); //returns true if the compute was successful (which it currently, always is)
return Task.FromResult(result);
}
}
我确实尝试使用 <BrokeredMessage> message.Complete();
,但那会引发消息锁定错误。
- 获取最新版本的包(>= v3.5.0)。将
CommunicationListener
上的 属性 'MessageLockRenewTimeSpan
' 设置为小于锁定持续时间的值。 (例如,锁定持续时间为 60 秒的 50 秒)这允许一些时钟偏差。注意:默认情况下这个属性是空的,这意味着没有自动更新锁。
当处理的批处理时间超过锁定持续时间所允许的时间时,此选项效果很好。
或
- 当处理时间超过锁定持续时间时,您可以使用 BrokeredMessage.RenewLock 定期延长消息锁定。如果需要,您可以在单独的线程中执行此操作。 May be useful info
当处理单个消息(批量大小 1)花费的时间超过锁定持续时间允许的时间时,此选项很有效。