C# · 12月 22, 2021

c# – Azure服务总线/服务结构消息未从队列中删除

我正在将计算引擎从Azure辅助角色移动到Azure Service Fabric中.它的工作原理是监听服务总线上的消息,然后根据消息内容进行处理.

目前,计算工作正常但如果计算花费的时间超过一分钟左右,则消息在完成后不会从队列中删除.在worker角色中,我们通过增加“AutoRenewTimeout”来解决这个问题.

var options = new OnMessageOptions { AutoComplete = true,AutoRenewTimeout = TimeSpan.FromMinutes(3) }; _queueClient.OnMessage(OnMessage,options);

但是,使用“ServiceFabric.ServiceBus” nuget包,我无法确定你设置它的位置.我使用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秒)这允许一些时钟偏差.注意:默认情况下,此属性为null,这意味着不会执行自动锁定续订.

处理批次所需的时间比锁定持续时间允许的时间长,此选项很有效.

要么

>当处理花费的时间超过锁定持续时间时,您可以使用BrokeredMessage.RenewLock定期扩展消息锁定.如果需要,您可以在单独的线程内执行此操作.
May be useful info

处理单个消息(批处理大小1)比锁定持续时间允许的时间更长时,此选项很有效.