消息队列服务重新启动后，服务未收到消息

本文关键字：消息 服务 队列 重新启动 | 更新日期: 2023-09-27 18:32:17

我们有一个从n个消息队列接收消息的服务。但是，如果重新启动消息队列服务，则即使在消息队列服务成功重新启动后，消息检索服务也会停止接收消息。

我尝试专门捕获消息检索服务中抛出的 MessageQueueException，并再次调用队列的 BeginReceive 方法。但是，在消息队列服务重新启动的 2 秒左右，我得到了大约 1875 个异常实例，然后当在我们的 StartListen 方法中抛出另一个 MessageQueueException 时，该服务停止运行。

是否有一种优雅的方法可以从消息队列服务重新启动中恢复？

    private void OnReceiveCompleted(object sender, ReceiveCompletedEventArgs e)
    {
        MessageQueue queue = (MessageQueue)sender;
        try
        {
            Message message = queue.EndReceive(e.AsyncResult);
            this.StartListening(queue);
            if (this.MessageReceived != null)
                this.MessageReceived(this, new MessageReceivedEventArgs(message));
        }
        catch (MessageQueueException)
        {
            LogUtility.LogError(String.Format(CultureInfo.InvariantCulture, StringResource.LogMessage_QueueManager_MessageQueueException, queue.MachineName, queue.QueueName, queue.Path));
            this.StartListening(queue);
        }            
    }
    public void StartListening(MessageQueue queue)
    {
        queue.BeginReceive();
    }

我需要处理这导致的无限循环问题并清理一下，但你明白了。

当发生 MessageQueueException 时，调用 RecoverQueue 方法。

    private void RecoverQueue(MessageQueue queue)
    {            
        string queuePath      = queue.Path;
        bool   queueRecovered = false;
        while (!queueRecovered)
        {
            try
            {
                this.StopListening(queue);
                queue.Close();
                queue.Dispose();
                Thread.Sleep(2000);
                MessageQueue newQueue = this.CreateQueue(queuePath);
                newQueue.ReceiveCompleted += new ReceiveCompletedEventHandler(this.OnReceiveCompleted);
                this.StartListening(newQueue);
                LogUtility.LogInformation(String.Format(CultureInfo.InvariantCulture, "Message queue {0} recovered successfully.", newQueue.QueueName));
                queueRecovered = true;
            }
            catch (Exception ex)
            {
                LogUtility.LogError(String.Format(CultureInfo.InvariantCulture, "The following error occurred while trying to recover queue: {0} error: {1}", queue.QueueName, ex.Message));                
            }
        }           
    }
    public void StopListening(MessageQueue queue)
    {
        queue.ReceiveCompleted -= new ReceiveCompletedEventHandler(this.OnReceiveCompleted);            
    }

消息队列服务重新启动后，服务未收到消息

收到服务重新启动导致的异常后，您必须释放旧MessageQueue，即取消ReceiveCompleted事件的连线，处置MessageQueue等。然后创建MessageQueue的新实例，并在新的MessageQueue实例上再次挂接到ReceiveCompleted事件。

或者，您可以使用轮询方法，该方法按特定时间间隔创建新实例，调用MessageQueue.Receive(TimeSpan) ，将等待传入消息或直到超时发生。在这种情况下，您可以处理消息并销毁MessageQueue实例并再次启动迭代。

通过每次重新创建MessageQueue，您可以确保内置恢复。此外，由于基础队列的内部缓存，创建MessageQueue的开销最小。

伪代码...

while (!notDone)// or use a timer or periodic task of some sort...
{
    try
    {
        using (MessageQueue queue = new MessageQueue(queuePath))
        {
            Message message = queue.Receive(TimeSpan.FromMilliseconds(500));
            // process message
        }
    }
    catch (MessageQueueException ex)
    {
        // handle exceptions
    }
}
相关文章：