Использование очередей Dead-Letter для обработки сбоев передачи сообщений

Статья
2025-03-25

Сообщения в очереди могут не доставляться. Эти неудачные сообщения записываются в очередь недоставленных писем. Сбой доставки может быть вызван такими причинами, как сбои сети, удаленная очередь, полная очередь, сбой проверки подлинности или сбой доставки вовремя.

Сообщения в очереди могут оставаться в очереди в течение длительного времени, если принимающее приложение не считывает их из очереди своевременно. Это поведение может быть неуместным для времязависимых сообщений. У сообщений с учетом времени есть свойство Time to Live (TTL) в привязке с очередью, указывающее, сколько времени сообщения могут находиться в очереди до истечения срока их действия. Просроченные сообщения отправляются в специальную очередь, называемую очередью недоставленных писем. Сообщения также могут быть помещены в очередь недоставленных писем по другим причинам, например превышение квоты очереди или из-за сбоя проверки подлинности.

Как правило, приложения записывают логику компенсации для чтения сообщений из очереди недоставленных писем и причин сбоя. Логика компенсации зависит от причины сбоя. Например, в случае сбоя проверки подлинности можно исправить сертификат, подключенный с сообщением, и повторно отправить сообщение. Если доставка завершилась ошибкой, так как достигнута квота целевой очереди, вы можете попробовать повторно осуществить доставку в надежде, что проблема квоты устранена.

Большинство систем очередей имеют общесистемную очередь недоставленных писем, где хранятся все неудачные сообщения из этой системы. Очередь сообщений (MSMQ) предоставляет две системные очереди недоставленных сообщений: транзакционную системную очередь недоставленных сообщений, в которой хранятся сообщения, не доставленные в транзакционную очередь, и нетранзакционную системную очередь недоставленных сообщений, в которой хранятся сообщения, не доставленные в нетранзакционную очередь. Если два клиента отправляют сообщения в две разные службы, и поэтому разные очереди в WCF совместно используют одну и ту же службу MSMQ для отправки, то можно использовать сочетание сообщений в системной очереди недоставленных писем. Это не всегда оптимально. В нескольких случаях (например, безопасность) может не потребоваться, чтобы один клиент считывал сообщения другого клиента из очереди недоставленных писем. Общая очередь недоставленных писем также требует от клиентов просматривать очередь, чтобы найти сообщение, которое они отправили, что может быть чрезвычайно дорого в зависимости от количества сообщений в этой очереди. Таким образом, в WCFNetMsmqBindingMsmqIntegrationBinding, и MSMQ в Windows Vista предоставляют настраиваемую очередь недоставленных писем (иногда называемую очередью недоставленных сообщений для конкретного приложения).

Настраиваемая очередь недоставленных писем обеспечивает изоляцию между клиентами, которые совместно используют ту же службу MSMQ для отправки сообщений.

В Windows Server 2003 и Windows XP Windows Communication Foundation (WCF) предоставляет общесистемную очередь недоставленных писем для всех клиентских приложений в очереди. В Windows Vista WCF предоставляет очередь недоставленных писем для каждого клиентского приложения в очереди.

Указание использования очереди Dead-Letter

Очередь недоставленных сообщений находится в менеджере очередей отправляющего приложения. Он сохраняет сообщения, истекшие или которые завершились сбоем передачи или доставки.

Привязка имеет следующие свойства очереди недоставленных писем:

Чтение сообщений из очереди Dead-Letter

Приложение, которое считывает сообщения из очереди недоставленных писем, похоже на службу WCF, которая считывает из очереди приложений, за исключением следующих незначительных различий:

Чтобы считывать сообщения из системной транзакционной очереди недоставленных сообщений, универсальный идентификатор ресурса (URI) должен иметь формат: net.msmq://localhost/system$;DeadXact.
Для чтения сообщений из системной очереди недоставленных сообщений URI должен иметь форму: net.msmq://localhost/system$;DeadLetter.
Чтобы считывать сообщения из пользовательской очереди недоставленных сообщений, URI должен иметь вид: net.msmq://localhost/private/<custom-dlq-name>, где custom-dlq-name — это имя пользовательской очереди недоставленных сообщений.

Более подробную информацию об адресации очередей см. в разделе «Конечные точки службы и адресация очередей».

Стек WCF на приемнике сравнивает адреса, на которых служба прослушивает, с адресом в сообщении. Если адреса совпадают, сообщение отправляется; Если нет, сообщение не отправляется. Это может привести к проблемам при чтении из очереди недоставленных писем, так как сообщения в очереди недоставленных писем обычно адресуются к службе, а не к службе хранения недоставленных писем. Поэтому служба, читающая из очереди недоставленных писем, должна установить фильтр адресов ServiceBehavior, который указывает системе сопоставлять все сообщения в очереди независимо от адресата. В частности, необходимо добавить ServiceBehavior с параметром Any в службу чтения сообщений из очереди недоставленных писем.

Обработка отравленных сообщений из очереди Dead-Letter

Обработка подозрительных сообщений доступна в очередях недоставленных писем с некоторыми условиями. Так как вы не можете создавать вложенные очереди из системных очередей, при чтении из системной очереди недоставленных писем ReceiveErrorHandling нельзя задать для Move. Обратите внимание, что если вы читаете из настраиваемой очереди неполученных писем, вы можете иметь подочереди и, следовательно, Move является допустимым расположением для плохого сообщения.

Если для ReceiveErrorHandling задано значение Reject, при чтении из настраиваемой очереди недоставленных писем проблемное сообщение помещается в системную очередь недоставленных писем. При чтении из очереди недоставленных писем системы сообщение удаляется (очищается). Сообщение из системной очереди недоставленных сообщений в MSMQ отклоняется и удаляется (очищается).

Пример

В следующем примере показано, как создать очередь недоставленных писем и как использовать ее для обработки просроченных сообщений. Этот пример основан на примере из раздела Как обмениваться очередями сообщений с конечными точками WCF. В следующем примере показано, как написать клиентский код в службу обработки заказов, которая использует очередь недоставленных писем для каждого приложения. В примере также показано, как обрабатывать сообщения из очереди недоставленных писем.

Ниже приведен код для клиента, указывающего очередь недоставленных писем для каждого приложения.

using System;
using System.ServiceModel.Channels;
using System.Configuration;
//using System.Messaging;
using System.ServiceModel;
using System.Transactions;

namespace Microsoft.ServiceModel.Samples
{
    
    //The service contract is defined in generatedProxy.cs, generated from the service by the svcutil tool.

    //Client implementation code.
    class Client
    {
        static void Main()
        {
            // Get MSMQ queue name from appsettings in configuration.
            string deadLetterQueueName = ConfigurationManager.AppSettings["deadLetterQueueName"];

            // Create the transacted MSMQ queue for storing dead message if necessary.
            if (!System.Messaging.MessageQueue.Exists(deadLetterQueueName))
                System.Messaging.MessageQueue.Create(deadLetterQueueName, true);

            OrderProcessorClient client = new OrderProcessorClient("OrderProcessorEndpoint");
        try
            {	

                // Create the purchase order.
                PurchaseOrder po = new PurchaseOrder();
                po.CustomerId = "somecustomer.com";
                po.PONumber = Guid.NewGuid().ToString();

                PurchaseOrderLineItem lineItem1 = new PurchaseOrderLineItem();
                lineItem1.ProductId = "Blue Widget";
                lineItem1.Quantity = 54;
                lineItem1.UnitCost = 29.99F;

                PurchaseOrderLineItem lineItem2 = new PurchaseOrderLineItem();
                lineItem2.ProductId = "Red Widget";
                lineItem2.Quantity = 890;
                lineItem2.UnitCost = 45.89F;

                po.orderLineItems = new PurchaseOrderLineItem[2];
                po.orderLineItems[0] = lineItem1;
                po.orderLineItems[1] = lineItem2;

                //Create a transaction scope.
                using (TransactionScope scope = new TransactionScope(TransactionScopeOption.Required))
                {
                    // Make a queued call to submit the purchase order.
                    client.SubmitPurchaseOrder(po);
                    // Complete the transaction.
                    scope.Complete();
                }

                client.Close();
            }
            catch(TimeoutException timeout)
            {
        Console.WriteLine(timeout.Message);
                client.Abort();
        }
            catch(CommunicationException conexcp)
            {
        Console.WriteLine(conexcp.Message);
                client.Abort();
        }

            Console.WriteLine();
            Console.WriteLine("Press <ENTER> to terminate client.");
            Console.ReadLine();
        }
    }
}

Imports System.ServiceModel.Channels
Imports System.Configuration
'using System.Messaging;
Imports System.ServiceModel
Imports System.Transactions

Namespace Microsoft.ServiceModel.Samples

    'The service contract is defined in generatedProxy.cs, generated from the service by the svcutil tool.

    'Client implementation code.
    Friend Class Client
        Shared Sub Main()
            ' Get MSMQ queue name from appsettings in configuration.
            Dim deadLetterQueueName As String = ConfigurationManager.AppSettings("deadLetterQueueName")

            ' Create the transacted MSMQ queue for storing dead message if necessary.
            If (Not System.Messaging.MessageQueue.Exists(deadLetterQueueName)) Then
                System.Messaging.MessageQueue.Create(deadLetterQueueName, True)
            End If


            Dim client As New OrderProcessorClient("OrderProcessorEndpoint")
            Try


                ' Create the purchase order.
                Dim po As New PurchaseOrder()
                po.CustomerId = "somecustomer.com"
                po.PONumber = Guid.NewGuid().ToString()

                Dim lineItem1 As New PurchaseOrderLineItem()
                lineItem1.ProductId = "Blue Widget"
                lineItem1.Quantity = 54
                lineItem1.UnitCost = 29.99F

                Dim lineItem2 As New PurchaseOrderLineItem()
                lineItem2.ProductId = "Red Widget"
                lineItem2.Quantity = 890
                lineItem2.UnitCost = 45.89F

                po.orderLineItems = New PurchaseOrderLineItem(1) {}
                po.orderLineItems(0) = lineItem1
                po.orderLineItems(1) = lineItem2

                'Create a transaction scope.
                Using scope As New TransactionScope(TransactionScopeOption.Required)
                    ' Make a queued call to submit the purchase order.
                    client.SubmitPurchaseOrder(po)
                    ' Complete the transaction.
                    scope.Complete()
                End Using


                client.Close()
            Catch timeout As TimeoutException
                Console.WriteLine(timeout.Message)
                client.Abort()
            Catch conexcp As CommunicationException
                Console.WriteLine(conexcp.Message)
                client.Abort()
            End Try

            Console.WriteLine()
            Console.WriteLine("Press <ENTER> to terminate client.")
            Console.ReadLine()
        End Sub
    End Class
End Namespace

Ниже приведен код для файла конфигурации клиента.

Ниже приведен код для обработки сообщений службы из очереди недоставленных писем.

using System;
using System.ServiceModel.Description;
using System.Configuration;
using System.Messaging;
using System.ServiceModel;
using System.ServiceModel.Channels;
using System.Transactions;

namespace Microsoft.ServiceModel.Samples
{
    // Define a service contract.
    [ServiceContract(Namespace = "http://Microsoft.ServiceModel.Samples")]
    public interface IOrderProcessor
    {
        [OperationContract(IsOneWay = true)]
        void SubmitPurchaseOrder(PurchaseOrder po);
    }

    // Service class that implements the service contract.
    // Added code to write output to the console window
    [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single, ConcurrencyMode = ConcurrencyMode.Single, AddressFilterMode = AddressFilterMode.Any)]
    public class PurchaseOrderDLQService : IOrderProcessor
    {
        OrderProcessorClient orderProcessorService;
        public PurchaseOrderDLQService()
        {
            orderProcessorService = new OrderProcessorClient("OrderProcessorEndpoint");
        }

        [OperationBehavior(TransactionScopeRequired = true, TransactionAutoComplete = true)]
        public void SimpleSubmitPurchaseOrder(PurchaseOrder po)
        {
            Console.WriteLine($"Submitting purchase order did not succeed ");
            MsmqMessageProperty mqProp = OperationContext.Current.IncomingMessageProperties[MsmqMessageProperty.Name] as MsmqMessageProperty;

            Console.WriteLine($"Message Delivery Status: {mqProp.DeliveryStatus} ");
            Console.WriteLine($"Message Delivery Failure: {mqProp.DeliveryFailure}");
            Console.WriteLine();
        }

        [OperationBehavior(TransactionScopeRequired = true, TransactionAutoComplete = true)]
        public void SubmitPurchaseOrder(PurchaseOrder po)
        {
            Console.WriteLine($"Submitting purchase order did not succeed ");
            MsmqMessageProperty mqProp = OperationContext.Current.IncomingMessageProperties[MsmqMessageProperty.Name] as MsmqMessageProperty;

            Console.WriteLine($"Message Delivery Status: {mqProp.DeliveryStatus} ");
            Console.WriteLine($"Message Delivery Failure: {mqProp.DeliveryFailure}");
            Console.WriteLine();

            // Resend the message if timed out.
            if (mqProp.DeliveryFailure == DeliveryFailure.ReachQueueTimeout ||
                mqProp.DeliveryFailure == DeliveryFailure.ReceiveTimeout)
            {
                // Re-send.
                Console.WriteLine("Purchase order Time To Live expired");
                Console.WriteLine("Trying to resend the message");

                // Reuse the same transaction used to read the message from dlq to enqueue the message to the application queue.
                orderProcessorService.SubmitPurchaseOrder(po);
                Console.WriteLine("Purchase order resent");
            }
        }

        // Host the service within this EXE console application.
        public static void Main()
        {
            // Create a ServiceHost for the PurchaseOrderDLQService type.
            using (ServiceHost serviceHost = new ServiceHost(typeof(PurchaseOrderDLQService)))
            {
                // Open the ServiceHostBase to create listeners and start listening for messages.
                serviceHost.Open();

                // The service can now be accessed.
                Console.WriteLine("The dead letter service is ready.");
                Console.WriteLine("Press <ENTER> to terminate service.");
                Console.WriteLine();
                Console.ReadLine();

                // Close the ServiceHostBase to shutdown the service.
                serviceHost.Close();
            }
        }
    }
}


Imports System.ServiceModel.Description
Imports System.Configuration
Imports System.Messaging
Imports System.ServiceModel
Imports System.ServiceModel.Channels
Imports System.Transactions

Namespace Microsoft.ServiceModel.Samples
    ' Define a service contract. 
    <ServiceContract(Namespace:="http://Microsoft.ServiceModel.Samples")> _
    Public Interface IOrderProcessor
        <OperationContract(IsOneWay:=True)> _
        Sub SubmitPurchaseOrder(ByVal po As PurchaseOrder)
    End Interface

    ' Service class that implements the service contract.
    ' Added code to write output to the console window
    <ServiceBehavior(InstanceContextMode:=InstanceContextMode.Single, ConcurrencyMode:=ConcurrencyMode.Single, AddressFilterMode:=AddressFilterMode.Any)> _
    Public Class PurchaseOrderDLQService
        Implements IOrderProcessor
        Private orderProcessorService As OrderProcessorClient
        Public Sub New()
            orderProcessorService = New OrderProcessorClient("OrderProcessorEndpoint")
        End Sub

        <OperationBehavior(TransactionScopeRequired:=True, TransactionAutoComplete:=True)> _
        Public Sub SimpleSubmitPurchaseOrder(ByVal po As PurchaseOrder)
            Console.WriteLine("Submitting purchase order did not succeed ", po)
            Dim mqProp As MsmqMessageProperty = TryCast(OperationContext.Current.IncomingMessageProperties(MsmqMessageProperty.Name), MsmqMessageProperty)

            Console.WriteLine("Message Delivery Status: {0} ", mqProp.DeliveryStatus)
            Console.WriteLine("Message Delivery Failure: {0}", mqProp.DeliveryFailure)
            Console.WriteLine()
        End Sub

        <OperationBehavior(TransactionScopeRequired:=True, TransactionAutoComplete:=True)> _
        Public Sub SubmitPurchaseOrder(ByVal po As PurchaseOrder) Implements IOrderProcessor.SubmitPurchaseOrder
            Console.WriteLine("Submitting purchase order did not succeed ", po)
            Dim mqProp As MsmqMessageProperty = TryCast(OperationContext.Current.IncomingMessageProperties(MsmqMessageProperty.Name), MsmqMessageProperty)

            Console.WriteLine("Message Delivery Status: {0} ", mqProp.DeliveryStatus)
            Console.WriteLine("Message Delivery Failure: {0}", mqProp.DeliveryFailure)
            Console.WriteLine()

            ' Resend the message if timed out.
            If mqProp.DeliveryFailure = DeliveryFailure.ReachQueueTimeout OrElse mqProp.DeliveryFailure = DeliveryFailure.ReceiveTimeout Then
                ' Re-send.
                Console.WriteLine("Purchase order Time To Live expired")
                Console.WriteLine("Trying to resend the message")

                ' Reuse the same transaction used to read the message from dlq to enqueue the message to the application queue.
                orderProcessorService.SubmitPurchaseOrder(po)
                Console.WriteLine("Purchase order resent")
            End If
        End Sub

        ' Host the service within this EXE console application.
        Public Shared Sub Main()
            ' Create a ServiceHost for the PurchaseOrderDLQService type.
            Using serviceHost As New ServiceHost(GetType(PurchaseOrderDLQService))
                ' Open the ServiceHostBase to create listeners and start listening for messages.
                serviceHost.Open()

                ' The service can now be accessed.
                Console.WriteLine("The dead letter service is ready.")
                Console.WriteLine("Press <ENTER> to terminate service.")
                Console.WriteLine()
                Console.ReadLine()

                ' Close the ServiceHostBase to shutdown the service.
                serviceHost.Close()
            End Using
        End Sub
    End Class
End Namespace

Ниже приведен код для файла конфигурации службы очереди недоставленных писем.

Поделиться через

Использование очередей Dead-Letter для обработки сбоев передачи сообщений

Указание использования очереди Dead-Letter

Чтение сообщений из очереди Dead-Letter

Обработка отравленных сообщений из очереди Dead-Letter

Пример

См. также

Дополнительные ресурсы