重新启动消息队列服务后，服务未收到消息 [英] Service not receiving messages after Message Queuing service restarted

问题描述

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

We have a service that receives messages from n message queues. However, if the Message Queuing service is restarted, the message retrieval service stops receiving messages even after the Message Queuing service has restarted successfully.

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

I have tried to specifically catch the MessageQueueException that is thrown in the message retrieval service and invoke the queue's BeginReceive method again. However, in the 2 seconds or so that it takes the Message Queuing service to restart, I get about 1875 instances of the exception and then the service stops functioning when another MessageQueueException is thrown in our StartListening method.

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

Is there an elegant way to recover from a Message Queuing service restart?

    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();
    }

我需要处理由此引起的无限循环问题，并对其进行一些清理，但是您明白了.

I need to deal with the infinite loop issue this causes and clean it up a bit but you get the idea.

发生MessageQueueException时，调用RecoverQueue方法.

When the MessageQueueException occurs, invoke the RecoverQueue method.

    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事件.

Upon receiving the exception that is the result of the service restarting, you have to release the old MessageQueue, i.e. unwiring your ReceiveCompleted event, disposing the MessageQueue, etc. Then create a new instance of the MessageQueue and hook up to the ReceiveCompleted event again on the new MessageQueue instance.

或者，您可以使用一种轮询方法，该方法以一定间隔创建一个新实例，调用MessageQueue.Receive(TimeSpan)，将等待传入消息或直到发生超时.在这种情况下，您可以处理消息并销毁MessageQueue实例，然后重新开始迭代.

Alternatively, you can use a polling method that creates a new instance on a certain interval, calls MessageQueue.Receive(TimeSpan), will wait for an incoming message or until the timeout occurs. In which case you handle the message and destroy the MessageQueue instance and start the iteration again.

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

By recreating the MessageQueue each time, you ensure a built in recovery. Also, the overhead of creating the MessageQueue is minimal due to internal caching of the underlying queue.

伪代码...

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
    }
}

这篇关于重新启动消息队列服务后，服务未收到消息的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！