MSMQ Receive() metoda timeout

Question

Moje pierwotne pytanie od pewnego czasu jest MSMQ Slow Queue Reading, jednak mam od tego zaawansowane i teraz myślę, że wiem, że problem jest nieco bardziej wyraźny.

Mój kod (no faktycznie częścią otwartej biblioteki źródłowej używam) wygląda następująco:

queue.Receive(TimeSpan.FromSeconds(10), MessageQueueTransactionType.Automatic); 

Który jest za pomocą funkcji Messaging.MessageQueue.Receive i kolejka to kolejka komunikatów. Problem jest następujący.

Powyższy wiersz kodu zostanie wywołany z określonym czasem oczekiwania (10 sekund). Funkcja Receive(...) jest funkcją blokującą i powinna blokować, dopóki wiadomość nie dotrze do kolejki, w której to nastąpi. Jeśli żadna wiadomość nie zostanie odebrana przed upływem limitu czasu, nastąpi powrót po przekroczeniu limitu czasu. Jeśli wiadomość jest w kolejce, gdy funkcja zostanie wywołana, natychmiast zwróci tę wiadomość.

Jednak dzieje się tak, że wywoływana jest funkcja Receive(...), ponieważ w kolejce nie ma żadnej wiadomości, a zatem czeka na pojawienie się nowej wiadomości. Gdy pojawi się nowa wiadomość (przed upływem limitu czasu), nie wykrywa tej nowej wiadomości i nadal czeka. Przekroczono limit czasu, w którym to momencie kod nadal się uruchamia i ponownie wywołuje Receive(...), gdzie odbiera wiadomość i przetwarza ją.

Ten problem występuje tylko po kilku dniach/tygodniach. Mogę sprawić, by działał normalnie ponownie przez usunięcie & odtworzenia kolejki. Zdarza się na różnych komputerach i różnych kolejkach. Wygląda więc na to, że coś się buduje, aż do momentu, w którym przełamie zdolność wyzwalania/powiadamiania, z której korzysta funkcja Receive(...).

Sprawdziłem wiele różnych rzeczy i wszystko wydaje się być normalne. & nie różni się od normalnej pracy kolejki. Jest dużo wolnego miejsca na dysku (13gig) i pamięci RAM (około 350MB wolnego od 1GB z tego co mogę powiedzieć). Sprawdziłem wpisy rejestru, które wyglądają tak samo jak inne kolejki, a monitor wydajności nie pokazuje niczego poza normalnymi. Uruchomiłem też narzędzie TMQ i nie widzę niczego, co mogłoby być nieprawdziwe.

Używam systemu Windows XP na wszystkich komputerach i wszystkie mają zainstalowany dodatek Service Pack 3. Nie wysyłam dużej ilości wiadomości do kolejek, co najwyżej 1 raz na 2 sekundy, ale generalnie o wiele rzadziej. Wiadomości są również niewielkie i nie zbliżają się do limitu 4 MB.

Jedyne co zauważyłem to to, że pliki p0000001.mq i r0000067.mq w C: \ WINDOWS \ system32 \ msmq \ storage mają rozmiar 4 096 KB, jednak mają taki rozmiar również na innych komputerach, które obecnie nie występują. problem. Problem nie dotyczy każdej kolejki na komputerze naraz, ponieważ mogę odtworzyć jedną kolejkę problemów na komputerze, a pozostałe koleje nadal występują.

Nie mam dużego doświadczenia z MSMQ, więc jeśli publikujesz możliwe rzeczy do sprawdzenia, możesz wyjaśnić, jak je sprawdzić lub gdzie mogę znaleźć więcej informacji na temat tego, o czym mówisz.

Obecnie sytuacja:

• ComputerA - 4 kolejki normalne
• ComputerB - 2 kolejki wystąpią problemy, 1 kolejka normalnego
• ComputerC - 2 kolejki wystąpią problemy
• ComputerD - 1 kolejka normalne
• Komputer - 2 kolejki normalne

Mam więc dużą liczbę komputerów/kolejek do porównania i testów.

Answer 1

Spróbuj

publicznej wiadomości Otrzymuj (timeout TimeSpan, Kursor Kursor)

przeciążony funkcji.

Aby uzyskać kursor dla obiektu MessageQueue, należy wywołać metodę CreateCursor dla tej kolejki.

Kursor jest używany z takimi metodami, jak Peek (TimeSpan, Cursor, PeekAction) i Receive (TimeSpan, Cursor), gdy trzeba czytać wiadomości, które nie znajdują się z przodu kolejki. Obejmuje to odczytywanie wiadomości synchronicznie lub asynchronicznie. Kursory nie muszą być używane do odczytu tylko pierwszej wiadomości w kolejce.

Podczas czytania wiadomości w ramach transakcji Usługa kolejkowania wiadomości nie przesuwa ruchu kursora, jeśli transakcja została przerwana. Załóżmy na przykład, że istnieje kolejka z dwoma komunikatami, A1 i A2. Jeśli usuniesz wiadomość A1 podczas transakcji, Usługa kolejkowania wiadomości przesunie kursor do wiadomości A2. Jeśli jednak transakcja zostanie przerwana z dowolnego powodu, wiadomość A1 zostanie wstawiona z powrotem do kolejki, ale kursor nadal wskazuje komunikat A2.

Aby zamknąć kursor, wywołaj Zamknij.

Answer 2

Jakiś szczególny powód, dla którego nie korzystasz z funkcji obsługi zdarzeń do słuchania kolejek? Biblioteka System.Messaging umożliwia dołączenie programu obsługi do kolejki zamiast, jeśli dobrze rozumiem, co robisz poprawnie, zapętlając Odbieranie co 10 sekund. Wypróbuj coś takiego:

class MSMQListener 
{ 
    public void StartListening(string queuePath) 
    { 
     MessageQueue msQueue = new MessageQueue(queuePath); 
     msQueue.ReceiveCompleted += QueueMessageReceived; 
     msQueue.BeginReceive(); 
    } 

    private void QueueMessageReceived(object source, ReceiveCompletedEventArgs args) 
    { 
     MessageQueue msQueue = (MessageQueue)source; 

     //once a message is received, stop receiving 
     Message msMessage = null; 
     msMessage = msQueue.EndReceive(args.AsyncResult); 

     //do something with the message 

     //begin receiving again 
     msQueue.BeginReceive(); 
    } 
} 

Answer 3

Używamy również NServiceBus i mieliśmy podobny problem w naszej sieci.

Zasadniczo MSMQ używa UDP z zatwierdzeniami dwufazowymi. Po otrzymaniu wiadomości musi zostać potwierdzona. Dopóki nie zostanie potwierdzony, nie może zostać odebrany po stronie klienta, ponieważ transakcja odbioru nie została sfinalizowana.

Zostało to spowodowane przez różne rzeczy w różnych czasach dla nas: - kiedyś było to spowodowane Distributed Transaction Coordinator w stanie komunikować się między komputerami za błędną konfigurację zapory - kolejny raz używaliśmy sklonowane maszyny wirtualne bez sysprep co czyniło wewnętrzny MSMQ ids nieunikalne i spowodowało otrzymanie wiadomości do jednego komputera i potwierdzenia do innego. Ostatecznie MSMQ wymyśla pewne rzeczy, ale zajmuje sporo czasu.

Mam nadzieję, że to pomoże.

Answer 4

Jeśli chcesz użyć coś całkowicie synchroniczny i bez wypadku można przetestować tę metodę

public object Receive(string path, int millisecondsTimeout) 
{ 
    var mq = new System.Messaging.MessageQueue(path); 
    var asyncResult = mq.BeginReceive(); 
    var handles = new System.Threading.WaitHandle[] { asyncResult.AsyncWaitHandle }; 
    var index = System.Threading.WaitHandle.WaitAny(handles, millisecondsTimeout); 
    if (index == 258) // Timeout 
    { 
     mq.Close(); 
     return null; 
    } 
    var result = mq.EndReceive(asyncResult); 
    return result; 
}