NServiceBus - Come ottenere una coda separata per ogni tipo di messaggio destinatario del destinatario?

Question

Ho una situazione seguente:

Quindi, il ricevitore sottoscrive due tipi di eventi: Eventa e eventB. NServiceBus crea la coda per il destinatario (Ricevitore) e inserisce i messaggi di tipo eventA e eventB nella stessa coda. La domanda è: se posso configurare NServiceBus per utilizzare code separate (ReceiverEventA e ReceiverEventB) per ciascun tipo di evento per il destinatario? Oppure posso avere due ricevitori in un singolo processo (e ciascuna coda separata del ricevitore). Il fatto è che EventA richiede molto più tempo per l'elaborazione rispetto a EventB e sono indipendenti, quindi se fossero in code separate, potrebbero essere elaborati contemporaneamente.

Aggiornamento: Se io vado con l'approccio ingenuo come questo, il ricevitore non riesce ad avviarsi con l'eccezione nullo di riferimento:

private static IBus GetBus<THandler, TEvent>() 
    {     
     var bus = Configure.With(new List<Type> 
            { 
             typeof(THandler), 
             typeof(TEvent), 
             typeof(CompletionMessage) 
            }) 
      .Log4Net() 
      .DefaultBuilder() 
      .XmlSerializer() 
      .MsmqTransport() 
      .IsTransactional(true) 
      .PurgeOnStartup(false) 
      .UnicastBus() 
      .LoadMessageHandlers() 
      .ImpersonateSender(false); 

     bus.Configurer.ConfigureProperty<MsmqTransport>(x => x.InputQueue, "Queue" + typeof(THandler).Name); 

     return bus.CreateBus().Start(); 
    } 

    [STAThread] 
    static void Main() 
    { 
     Busses = new List<IBus> 
        { 
         GetBus<ItemEventHandlerA, ItemEventA>(), 
         GetBus<ItemEventHandlerB, ItemEventB>() 
        };   

     Application.EnableVisualStyles(); 
     Application.SetCompatibleTextRenderingDefault(false); 
     Application.Run(new TestForm()); 
    } 

eccezione dello stack trace è:

a NServiceBusTest2.WinFormsReceiver.Program .GetBusTHandler, TEvent in C: \ Users \ Utente \ Documenti \ Visual Studio 2010 \ Projects \ NServiceBusTest2 \ NServiceBusTest2.WinFormsReceiver \ Program.cs: riga 57
in NServiceBusTest2.WinFormsReceiver.Program.Main() in C: \ Users \ Utente \ Documenti \ Visual Studio 2 010 \ Projects \ NServiceBusTest2 \ NServiceBusTest2.WinFormsReceiver \ Program.cs: linea 26
a System.AppDomain._nExecuteAssembly (RuntimeAssembly assemblaggio, String [] args)
a System.AppDomain.ExecuteAssembly (String assemblyFile, Evidence assemblySecurity, String [ ] args) a Microsoft.VisualStudio.HostingProcess.HostProc.RunUsersAssembly()
a System.Threading.ThreadHelper.ThreadStart_Context (stato oggetto)
a System.Threading.ExecutionContext.Run (ExecutionContext ExecutionContext, ContextCallback callback, oggetto di stato, booleano ignoreSyncCtx)
a System.Threading.ExecutionContext.Run (ExecutionContext ExecutionContext, ContextCallback callback, Object state)
a System.Threading.ThreadHelper.ThreadStart()

Answer 1

Ho iniziato a scrivere una spiegazione più dettagliata del perché NON dovresti avere due processi separati, a supporto del mio commento alla risposta @stephenl pubblicata. NServiceBus applica fondamentalmente una coda di input per processo.

Così normalmente, avresti due processi separati. EventAService leggeva EventA da QForEventA, in un processo separato da EventBService che leggeva EventB da QForEventB.

Poi ho guardato più attentamente il codice di esempio e mi sono reso conto che si trovava in un'app Windows Form. Duh! Ora mi sento un po 'sciocco. Ovviamente puoi avere solo un processo. Immagina se dopo aver lanciato Outlook dovessi avviare MailService.exe per ricevere effettivamente la posta!

Quindi il problema è in realtà che avete tempi di elaborazione drasticamente diversi per EventA ed EventB all'interno della vostra app Windows Form. Non ho alcun modo di sapere quale sia quel lavoro, ma questo è un po 'strano per un'applicazione client.

Il più delle volte si tratta di un servizio che ha grande trasformazione a che fare, ed ogni messaggio ricevuto da un cliente è abbastanza leggero - qualcosa sulla falsariga di "Entity X è cambiata così la prossima volta che sarà necessario caricarlo direttamente dal database "e l'elaborazione implica semplicemente eliminare qualcosa dalla cache, certamente non un processo di lunga durata.

Ma sembra che per qualsiasi motivo l'elaborazione nel client richiede più tempo, che in un'applicazione WinForms sta interessando causa delle preoccupazioni circa thread UI di smistamento, bloccando il thread dell'interfaccia utente, ecc

vorrei suggerire un sistema dove non si esegue tutta l'elaborazione nel gestore NServiceBus nella propria app WinForms, ma si effettua il marshalling da qualche altra parte. Buttalo sul ThreadPool come oggetto di lavoro, o qualcosa del genere. Oppure metti gli oggetti di lunga durata in una coda e ottieni uno scricchiolio di thread in background su quelli alla sua stessa velocità. In questo modo tutto il gestore di messaggi di NServiceBus fa "Sì, ho ricevuto il messaggio. Grazie mille." Quindi non dovrebbe importare se i messaggi di NServiceBus vengono elaborati uno alla volta.

Aggiornamento:

Nei commenti, l'OP chiede cosa succede se si lancia il lavoro al ThreadPool dopo NServiceBus finsihes riceve. Questo è, naturalmente, il rovescio della medaglia di questo approccio - dopo che NServiceBus è terminato, sei da solo - se fallisce nel ThreadPool, poi spetta a te creare la tua logica dei tentativi, o solo prendere l'eccezione, avvisare l'utente dell'app WinForms e lasciarlo morire.

Ovviamente è ottimale, ma si pone la domanda: qual è esattamente il tipo di lavoro che stai cercando di realizzare nell'app WinForms? Se la robustezza offerta da NServiceBus (i tentativi automatici e la coda di errori per i messaggi velenosi) è un elemento fondamentale di questo puzzle, perché in una domanda di WinForms si sta verificando in primo luogo? Probabilmente questa logica deve essere scaricata su un servizio esterno all'applicazione WinForms, dove avere code separate per ogni tipo di messaggio (distribuendo servizi separati) diventa facile, e quindi solo i pezzi che interessano l'interfaccia utente vengono sempre restituiti al client WinForms. Quando i messaggi all'interfaccia utente interessano solo l'interfaccia utente, l'elaborazione di questi è quasi sempre banale e non sarà necessario scaricare il threadpool per tenere il passo.

Parlando direttamente alla situazione descritta nel GitHub Issue, questo sembra davvero il tipo di scenario in cui i processi separati per ciascun tipo di messaggio sono esattamente la soluzione prescritta. Ho sentito dire che suona schiacciante per l'implementazione e la gestione di molti processi, ma penso che scoprirai che non è così male come sembra. Ci sono anche vantaggi: se devi ridistribuire il tuo connettore con Amazon.com, ad esempio, devi solo ridistribuire quell'endpoint, senza tempi di inattività per nessuno degli altri, o qualsiasi preoccupazione che i bug potrebbero essere stati introdotti agli altri.

Per facilitare la distribuzione, si spera che si stia utilizzando un server di integrazione continua e quindi si esegua il check-in su strumenti come DropkicK che aiutano le distribuzioni a essere programmate. Personalmente, il mio strumento di distribuzione preferito è il buon vecchio Robocopy. Anche qualcosa di semplice come 1) NET STOP ServiceName, 2) ROBOCOPY, 3) NET START ServiceName è abbastanza efficace.

Answer 2

Non è necessario, è solo bisogno di creare un gestore per ogni evento nel ricevitore. Ad esempio

public class EventAHandler : IHandleMessages<EventA>{} 

public class EventBHandler : IHandleMessages<EventB>{} 

Se si desidera separare le code, è necessario inserire ciascun gestore in un endpoint separato. Ha senso?

Inoltre penso che il tuo schema ha bisogno di un po 'di lavoro. È una cosa dell'etichettatura, ne sono sicuro, ma il modo in cui lo vedo è che Host è l'editore effettivo, non gli endpoint client (che hai nominato Publisher A e Publisher B).

UPDATE: suo un semplice esempio, ma illustra quello che voglio dire - https://github.com/sliedig/sof9411638

ho esteso il campione Full Duplex che viene fornito con NServiceBus per includere tre punti finali supplementari, due dei quali sottoscrivono gli eventi in corso di pubblicazione da il server rispettivamente e un endpoint che gestisce entrambi. HTH.