Używanie utraconych kolejek na potrzeby obsługi transferów komunikatów zakończonych niepowodzeniem
Dostarczanie komunikatów w kolejce może zakończyć się niepowodzeniem. Te komunikaty, które zakończyły się niepowodzeniem, są rejestrowane w kolejce utraconych komunikatów. Niepowodzenie dostarczania może być spowodowane przyczynami, takimi jak awarie sieci, usunięta kolejka, pełna kolejka, niepowodzenie uwierzytelniania lub niepowodzenie dostarczania na czas.
Komunikaty w kolejce mogą pozostawać w kolejce przez długi czas, jeśli aplikacja odbierający nie odczytuje ich z kolejki w odpowiednim czasie. To zachowanie może nie być odpowiednie w przypadku komunikatów z uwzględnieniem czasu. Komunikaty wrażliwe na czas mają właściwość Time to Live (TTL) ustawioną w powiązaniu w kolejce, która wskazuje, jak długo komunikaty mogą znajdować się w kolejce przed ich wygaśnięciem. Wygasłe komunikaty są wysyłane do specjalnej kolejki nazywanej kolejką utraconych komunikatów. Komunikaty mogą być również umieszczane w kolejce utraconych wiadomości z innych powodów, takich jak przekroczenie limitu przydziału kolejki lub z powodu niepowodzenia uwierzytelniania.
Ogólnie rzecz biorąc, aplikacje zapisują logikę kompensacji w celu odczytywania komunikatów z kolejki utraconych komunikatów i przyczyn awarii. Logika kompensacji zależy od przyczyny awarii. Na przykład w przypadku niepowodzenia uwierzytelniania można poprawić certyfikat dołączony do komunikatu i wysłać ponownie komunikat. Jeśli dostarczanie nie powiodło się, ponieważ osiągnięto docelowy limit przydziału kolejki, możesz ponownie wykonać dostarczanie w nadziei, że problem z limitem przydziału został rozwiązany.
Większość systemów kolejkowania ma kolejkę utraconych komunikatów dla całego systemu, w której są przechowywane wszystkie komunikaty, które zakończyły się niepowodzeniem. Kolejkowanie komunikatów (MSMQ) udostępnia dwie kolejki utraconych komunikatów dla całego systemu: transakcyjną kolejkę utraconych komunikatów, która przechowuje komunikaty, które zakończyły się niepowodzeniem dostarczania do kolejki transakcyjnej, oraz kolejkę utraconych komunikatów obejmujących cały system bez transakcji, która przechowuje komunikaty, które zakończyły się niepowodzeniem dostarczania do kolejki nie transakcyjnej. Jeśli dwóch klientów wysyła komunikaty do dwóch różnych usług, a w związku z tym różne kolejki w programie WCF współużytkują tę samą usługę MSMQ do wysłania, możliwe jest połączenie komunikatów w kolejce komunikatów utraconych w systemie. Nie zawsze jest to optymalne. W kilku przypadkach (na przykład zabezpieczenia) jeden klient może nie chcieć odczytać komunikatów innego klienta z kolejki utraconych komunikatów. Udostępniona kolejka utraconych komunikatów wymaga również od klientów przeglądania kolejki w celu znalezienia wysłanego komunikatu, który może być zbyt kosztowny w oparciu o liczbę komunikatów w kolejce utraconych komunikatów. W związku z tym w usługach WCFNetMsmqBindingMsmqIntegrationBinding, i MSMQ w systemie Windows Vista udostępniają niestandardową kolejkę utraconych komunikatów (czasami nazywaną kolejką utraconych komunikatów specyficznych dla aplikacji).
Niestandardowa kolejka utraconych komunikatów zapewnia izolację między klientami, którzy korzystają z tej samej usługi MSMQ w celu wysyłania komunikatów.
W systemach Windows Server 2003 i Windows XP program Windows Communication Foundation (WCF) udostępnia kolejkę utraconych komunikatów na całym systemie dla wszystkich aplikacji klienckich w kolejce. W systemie Windows Vista program WCF udostępnia kolejkę utraconych komunikatów dla każdej aplikacji klienckiej w kolejce.
Określanie użycia kolejki utraconych komunikatów
Kolejka utraconych komunikatów znajduje się w menedżerze kolejek aplikacji wysyłającej. Przechowuje komunikaty, które wygasły lub które zakończyły się niepowodzeniem, transferem lub dostarczaniem.
Powiązanie ma następujące właściwości kolejki utraconych komunikatów:
Odczytywanie komunikatów z kolejki utraconych komunikatów
Aplikacja, która odczytuje komunikaty z kolejki utraconych komunikatów, jest podobna do usługi WCF, która odczytuje z kolejki aplikacji, z wyjątkiem następujących drobnych różnic:
Aby odczytać komunikaty z kolejki komunikatów transakcyjnych utraconych wiadomości, identyfikator URI (Uniform Resource Identifier) musi mieć postać: net.msmq://localhost/system$; DeadXact.
Aby odczytać komunikaty z kolejki komunikatów nieu transakcyjnych utraconych, identyfikator URI musi mieć postać: net.msmq://localhost/system$; DeadLetter.
Aby odczytać komunikaty z niestandardowej kolejki utraconych komunikatów, identyfikator URI musi mieć postać:net.msmq://localhost/private/<custom-dlq-name>, gdzie nazwa custom-dlq-name jest nazwą niestandardowej kolejki utraconych komunikatów.
Aby uzyskać więcej informacji o sposobie rozwiązywania problemów z kolejkami, zobacz Service Endpoints and Queue Address (Punkty końcowe usługi i adresowanie kolejek).
Stos WCF na odbiorniku pasuje do adresów, na których usługa nasłuchuje z adresem w komunikacie. Jeśli adresy są zgodne, wiadomość jest wysyłana; jeśli nie, komunikat nie jest wysyłany. Może to powodować problemy podczas odczytywania z kolejki utraconych komunikatów, ponieważ komunikaty w kolejce utraconych komunikatów są zwykle adresowane do usługi, a nie do usługi kolejki utraconych komunikatów. W związku z tym odczyt usługi z kolejki utraconych wiadomości musi zainstalować filtr ServiceBehavior adresu, który nakazuje stosowi dopasowanie wszystkich komunikatów w kolejce niezależnie od adresu. W szczególności należy dodać parametr ServiceBehavior z parametrem Any do usługi odczytu komunikatów z kolejki utraconych komunikatów.
Obsługa komunikatów zatrutych z kolejki utraconych komunikatów
Obsługa zatrutych komunikatów jest dostępna w kolejkach utraconych komunikatów z pewnymi warunkami. Ponieważ nie można tworzyć kolejek podrzędnych z kolejek systemowych, podczas odczytywania z kolejki ReceiveErrorHandling utraconych komunikatów systemowych nie można ustawić Movewartości . Należy pamiętać, że jeśli odczytujesz z niestandardowej kolejki utraconych komunikatów, możesz mieć kolejki podrzędne i dlatego Move jest prawidłowym dyspozycją dla komunikatu trucizny.
Gdy ReceiveErrorHandling jest ustawiona wartość Reject, podczas odczytywania z niestandardowej kolejki utraconych komunikatów jest umieszczany w kolejce utraconych komunikatów systemowych. W przypadku odczytywania z kolejki utraconych komunikatów systemowych jest usuwany (przeczyszczony). Odrzucenie z kolejki utraconych komunikatów systemu w MSMQ spada (czyści).
Przykład
W poniższym przykładzie pokazano, jak utworzyć kolejkę utraconych komunikatów i jak używać jej do przetwarzania wygasłych komunikatów. Przykład jest oparty na przykładzie w temacie Instrukcje: Wymiana komunikatów w kolejce z punktami końcowymi programu WCF. W poniższym przykładzie pokazano, jak napisać kod klienta w usłudze przetwarzania zamówień, która używa kolejki utraconych komunikatów dla każdej aplikacji. W przykładzie pokazano również, jak przetwarzać komunikaty z kolejki utraconych komunikatów.
Poniżej znajduje się kod klienta, który określa kolejkę utraconych komunikatów dla każdej aplikacji.
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
Poniżej przedstawiono kod pliku konfiguracji klienta.
Poniżej przedstawiono kod komunikatów przetwarzania komunikatów usługi z kolejki utraconych komunikatów.
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 ", po);
MsmqMessageProperty mqProp = OperationContext.Current.IncomingMessageProperties[MsmqMessageProperty.Name] as MsmqMessageProperty;
Console.WriteLine("Message Delivery Status: {0} ", mqProp.DeliveryStatus);
Console.WriteLine("Message Delivery Failure: {0}", mqProp.DeliveryFailure);
Console.WriteLine();
}
[OperationBehavior(TransactionScopeRequired = true, TransactionAutoComplete = true)]
public void SubmitPurchaseOrder(PurchaseOrder po)
{
Console.WriteLine("Submitting purchase order did not succeed ", po);
MsmqMessageProperty mqProp = OperationContext.Current.IncomingMessageProperties[MsmqMessageProperty.Name] as 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 ||
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
Poniżej przedstawiono kod pliku konfiguracji usługi kolejki utraconych komunikatów.