RabbitMQ - 确保每条消息单线程

Question

我有一个程序将 NServiceBus 用于排队消息机制。 我在日志中看到有时一条消息被不同的线程处理了两次，尽管这些消息只发送一次。

更重要的是，如果我强制我的服务只使用 1 个线程，我就没有这个问题。

如何确保消息只处理一次？

谢谢。

缩短的日志示例（您可以看到我收到带有 chunkId 48 的消息，我完成了所有处理，包括删除回调，然后我再次收到带有 chunkId 48 的相同消息，此时我无法找到回调。然后是 NserviceBus 异常）：

Risco.Rsp.RACServiceImpl.RACOutputMessageHandler [(null)] - RACOutputMessageHandler.Handle start for ProcessType:GetClientCacheEntity with ProcessId:44eb600b-87e6-4eab-a54e-0a12480c9784 and ChunkId: 48 and TotalChunks: 57 and主题：23b041a3-4a62-48d1-a199-20a20e5de72a …… …… ... Risco.Rsp.RACServiceImpl.RACOutputMessageHandler [(null)] - 已成功发送进程 ID：44eb600b-87e6-4eab-a54e-0a12480c9784 的结果，进程类型：GetClientCacheEntity，客户端 44eb600b-87e6-4eab- a54e-0a12480c9784（0 块中的 0 块） 2017-01-02 11:13:13,882 [64] 信息 Risco.Rsp.RACServiceImpl.RACOutputMessageHandler [(null)] - RACOutputMessageHandler.Handle start for ProcessType:GetClientCacheEntity with ProcessId:44eb600b-87e6-4eab-a54e -0a12480c9784 和 ChunkId：48 和 TotalChunks：57 和主题：23b041a3-4a62-48d1-a199-20a20e5de72a 2017-01-02 11:13:13,888 [64] 信息 Risco.Rsp.RACServiceImpl.RACOutputMessageHandler [(null)] - 句柄 (RACOutputMessage) - 找不到键 44eb600b-87e6-4eab-a54e 的回调-0a12480c9784，进程 ID：44eb600b-87e6-4eab-a54e-0a12480c9784，进程类型：GetClientCacheEntity 2017-01-02 11:13:13,923 [158] 错误 NServiceBus.Timeout.TimeoutManagerDeferrer [(null)] - 延迟消息时出现问题。确保没有为您的端点调用 DisableTimeoutManager。 NServiceBus.Unicast.Queuing.QueueNotFoundException: Exchange for the recipient不存在---> RabbitMQ.Client.Exceptions.AlreadyClosedException: 已经关闭: AMQP操作被中断: AMQP关闭原因, 由Peer发起, code=404, text ="NOT_FOUND - 在 vhost 'AxesPlus' 中没有交换 'Risco.Rsp.Ac.AMAC.Service.Timeouts'"，classId=60，methodId=40，cause= 在 RabbitMQ.Client.Impl.ModelBase.WaitForConfirms（时间跨度超时，布尔和超时） 在 RabbitMQ.Client.Impl.ModelBase.WaitForConfirmsOrDie（时间跨度超时） 在 C:\BuildAgent\work\ef98ad7376e3379a\src\NServiceBus.RabbitMQ\ConfirmsAwareChannel.cs:line 31 中的 NServiceBus.Transports.RabbitMQ.ConfirmsAwareChannel.Dispose() --- 内部异常堆栈跟踪结束 --- 在 C:\BuildAgent\work\ef98ad7376e3379a\src\NServiceBus.RabbitMQ\ConfirmsAwareChannel.cs:line 47 中的 NServiceBus.Transports.RabbitMQ.ConfirmsAwareChannel.Dispose() 在 NServiceBus.Transports.RabbitMQ.RabbitMqMessageSender.Send(TransportMessage message, SendOptions sendOptions) 在 c:\BuildAgent\work\ef98ad7376e3379a\src\NServiceBus.RabbitMQ\RabbitMqMessageSender.cs:line 27 在 NServiceBus.Timeout.TimeoutManagerDeferrer.Defer(TransportMessage message, SendOptions sendOptions) 在 C:\BuildAgent\work\3206e2123f54fce4\src\NServiceBus.Core\Timeout\Core\TimeoutManagerDeferrer.cs:line 42 2017-01-02 11:13:13,936 [174] 错误 NServiceBus.Timeout.TimeoutManagerDeferrer [(null)] - 延迟消息时出现问题。确保没有为您的端点调用 DisableTimeoutManager。 NServiceBus.Unicast.Queuing.QueueNotFoundException: Exchange for the recipient不存在---> RabbitMQ.Client.Exceptions.AlreadyClosedException: 已经关闭: AMQP操作被中断: AMQP关闭原因, 由Peer发起, code=404, text ="NOT_FOUND - 在 vhost 'AxesPlus' 中没有交换 'Risco.Rsp.Ac.AMAC.Service.Timeouts'"，classId=60，methodId=40，cause= 在 RabbitMQ.Client.Impl.ModelBase.WaitForConfirms（时间跨度超时，布尔和超时） 在 RabbitMQ.Client.Impl.ModelBase.WaitForConfirmsOrDie（时间跨度超时） 在 C:\BuildAgent\work\ef98ad7376e3379a\src\NServiceBus.RabbitMQ\ConfirmsAwareChannel.cs:line 31 中的 NServiceBus.Transports.RabbitMQ.ConfirmsAwareChannel.Dispose() --- 内部异常堆栈跟踪结束 --- 在 C:\BuildAgent\work\ef98ad7376e3379a\src\NServiceBus.RabbitMQ\ConfirmsAwareChannel.cs:line 47 中的 NServiceBus.Transports.RabbitMQ.ConfirmsAwareChannel.Dispose() 在 NServiceBus.Transports.RabbitMQ.RabbitMqMessageSender.Send(TransportMessage message, SendOptions sendOptions) 在 c:\BuildAgent\work\ef98ad7376e3379a\src\NServiceBus.RabbitMQ\RabbitMqMessageSender.cs:line 27 在 NServiceBus.Timeout.TimeoutManagerDeferrer.Defer(TransportMessage message, SendOptions sendOptions) 在 C:\BuildAgent\work\3206e2123f54fce4\src\NServiceBus.Core\Timeout\Core\TimeoutManagerDeferrer.cs:line 42

Answer 1

问题在于超时配置。 MessageHandler 不够快来处理消息并离开处理程序，所以我通过创建一个新任务来修复它，该任务将“替换”处理程序（意思是，在处理程序开始时调用一个替换原始处理程序的函数）