Messaggistica Pub/Sub con NCache Utilizzo di un'applicazione .NET
In questo video esamineremo Messaggistica Pub/Sub con NCache Utilizzando un'applicazione .NET, il modello Pub/Sub è uno schema di messaggistica che consente a diverse parti del sistema di comunicare tra loro senza essere direttamente dipendenti l'una dall'altra. Si ha quindi un publisher che invia messaggi e un subscriber che li riceve. Nel mezzo si trova un broker, utilizzato per instradare i messaggi da un'estremità all'altra.
Qui è dove NCache entra. Agisce come un broker qui usando l'argomento. Come puoi vedere qui. Ora questo argomento si trova all'interno del NCache Il cluster di cache memorizza i tuoi messaggi. Memorizza le informazioni del tuo editore, le informazioni del tuo abbonato e viene utilizzato anche per inoltrare eventi al tuo editore o al tuo abbonato.
Tutto ciò, ecco come, permette di inviare i messaggi da un punto all'altro senza che le applicazioni dipendano l'una dall'altra.
Che cos'è la messaggistica Pub/Sub?
- Publisher invia messaggi
- abbonato aspetta un messaggio
- Broker si trova al centro, instradando i messaggi dagli editori agli abbonati
- Indiretto e disaccoppiato Editore e abbonato. Facile da scalare, gestire ed estendere.
Argomento
Iniziamo quindi esaminando come creare un argomento. Come potete vedere, chiamiamo l'interfaccia del servizio di messaggistica all'interno della cache. Chiamiamo il metodo createTopic. L'unica cosa che passiamo, ovviamente, è il nome dell'argomento. Una volta creato un argomento, questo rimane nella cache. Esiste sempre finché non lo si elimina. E naturalmente, la riga qui sopra mostra come recuperare l'argomento. Una volta creato, è sufficiente utilizzare questa chiamata al metodo.
Crea/Ottieni argomento
String topicName = "ExampleTopic";
ITopic topic = _cache.MessagingService.GetTopic(topicName);
if (topic == null)
{
topic = _cache.MessagingService.CreateTopic(topicName);
}Pubblica messaggio
Panoramica dell'editore
Opzione di consegna
- Tutti - Invia il messaggio a tutti gli abbonati.
- Qualunque - Invia il messaggio a un solo abbonato.
Modalità di consegna
- Sincronizzazione - Attende la conferma prima di inviare il messaggio successivo.
- Asincrono - Invia messaggi in background senza attendere.
- Massa - Invia più messaggi contemporaneamente in un'unica operazione.
Abbiamo quindi un argomento. La prossima cosa che vogliamo fare è pubblicare un messaggio all'interno di questo argomento. NCache Offre diverse funzionalità per la pubblicazione dei messaggi. Puoi quindi selezionare un'opzione di invio differente. Puoi selezionare "Tutti" per inviare un messaggio a tutti gli iscritti registrati, oppure selezionare "Qualsiasi" per inviare un messaggio solo a uno degli iscritti.
Inoltre, è possibile selezionare la modalità di invio per pubblicare i messaggi in modo sincrono. È possibile inviarli in modo asincrono, il che ovviamente offre prestazioni migliori. Infine, è possibile anche inviarli in blocco, ovvero in batch, anziché uno alla volta. Anche questa opzione, ovviamente, offre prestazioni migliori. La modalità sincrona è consigliata quando si desidera che l'applicazione del destinatario riceva i messaggi nello stesso ordine in cui sono stati inviati dal mittente. Ecco perché si sceglie la modalità sincrona.
Pubblicazione di un messaggio
Ora diamo un'occhiata al codice per la pubblicazione di un messaggio. Ovviamente, prima dobbiamo ottenere il nostro oggetto ITopic, il nostro oggetto topic. Creiamo un oggetto messaggio e i parametri che passiamo sono il payload. Quindi sto passando un oggetto ordine come payload. È possibile anche passare un tempo di scadenza per il messaggio. Infine, chiamiamo il metodo publish. Forniamo questi parametri: il messaggio stesso, l'opzione di consegna ed è anche possibile pubblicare il messaggio in modo asincrono. Questo terzo parametro qui, come potete vedere, serve per la notifica di errore di consegna del messaggio. Ve lo spiegherò tra un attimo.
ITopic topic = cache.MessagingService.GetTopic(topicName);
Message message = new Message(new Order(), TimeSpan.FromSeconds(15));
// deliver message to all subscribers
topic.Publish(message, DeliveryOption.All, true);
// deliver message to any one subscriber
topic.Publish(message, DeliveryOption.Any, true);
// send message asynchronously
topic.PublishAsync(message, deliveryOption, true);Errore di consegna principale
Quindi, abiliti quell'opzione lì, impostandola su true, e crei anche questo metodo di callback. In questo caso, puoi vedere che si limita a visualizzare un messaggio di log che indica che la consegna del messaggio non è andata a buon fine. E questa riga qui serve per registrare il gestore eventi. Quindi, devi abilitare quel parametro nella diapositiva precedente, come puoi vedere qui. E registri il gestore eventi tramite questo frammento di codice. In questo modo, ogni volta che un messaggio non viene ricevuto da un sottoscrittore, il mittente viene notificato. E poi, utilizzando questo callback, puoi fare in modo che la tua applicazione reagisca in un certo modo o registri un messaggio.
private static void MessageDeliveryFailureNotification(object sender, MessageFailedEventArgs args) {
Console.WriteLine("Failed to deliver message. " + args.MessageFailureReason);
}
// register event handler
ITopic topic = cache.MessagingService.GetTopic(topicName);
topic.MessageDeliveryFailure += MessageDeliveryFailedNotification;Sottoscrivi
Tipo di abbonamento
- Durevole - Memorizza i messaggi quando è offline e li consegna al momento della riconnessione.
- Non durevole - Elimina i messaggi quando l'abbonato è offline.
Politica
- Exclusive - Consente un solo abbonato attivo.
- Durevole - Consente la presenza di più abbonati attivi.
Modalità di consegna
- Sincronizza - Trasmette i messaggi in sequenza, preservandone l'ordine.
- Asincrono - Consegna i messaggi simultaneamente, senza garanzia di ordine.
Come potete vedere, ci possono essere più editori che pubblicano un messaggio su un determinato argomento. All'interno dell'argomento, ci possono essere più abbonamenti, che a loro volta hanno più abbonati. Come potete notare, esistono diversi tipi di abbonamenti. Ad esempio, ci sono abbonamenti durevoli e abbonamenti non durevoli.
Abbonamenti durevoli
Il vantaggio di avere un abbonamento durevole è che, come in un abbonamento permanente, i tuoi messaggi vengono memorizzati anche se sei abbonato e la tua applicazione è offline o si disconnette. Il messaggio verrà conservato fino alla riconnessione o fino alla scadenza del messaggio stesso.
Abbonamenti non durevoli
Al contrario, con gli abbonamenti non permanenti, i messaggi vengono persi se l'applicazione dell'abbonato si disconnette. Quindi, considerateli come abbonamenti temporanei.
Politica di Abbonamento
Dopodiché, puoi anche selezionare la politica di abbonamento. L'opzione Esclusiva ti consente di avere un solo abbonato al massimo. L'opzione Condivisa ti consente di avere più abbonati.
Quindi gli abbonamenti non permanenti sono esclusivi per impostazione predefinita. Puoi avere al massimo un solo abbonato, mentre per gli abbonamenti permanenti puoi ovviamente selezionare la politica di abbonamento. Ok.
Modalità di consegna
Infine, analogamente a come funziona il publisher, anche per l'elaborazione dei messaggi da parte dell'abbonamento è possibile scegliere tra modalità di consegna sincrona o asincrona. Naturalmente, una modalità offre messaggi di ordine superiore, mentre l'altra garantisce prestazioni migliori.
Creazione di abbonamenti non durevoli
Innanzitutto, vediamo come creare una sottoscrizione non durevole. Come potete vedere, prima creiamo un argomento e poi chiamiamo il metodo createSubscription. Questo creerà una sottoscrizione non durevole. L'unico parametro del metodo che passiamo è la callback di ricezione del messaggio. Ecco a cosa serve la callback di ricezione del messaggio: notifica al sottoscrittore quando riceve il messaggio. In questo caso, si tratta semplicemente di visualizzare un messaggio.
ITopic topic = cache.MessagingService.GetTopic(topicName);
// Create and register subscribers for the given topic
// Message received callback has to be passed
ITopicSubscription subscription = topic.CreateSubscription(MessageReceivedCallback);
// Used to notify the Subscriber when it receives a message
private static void MessageReceivedCallback(object sender, MessageEventArgs args) {
Console.WriteLine("Message received for topic " + args.TopicName);
}Creazione di abbonamenti duraturi
Per le sottoscrizioni durevoli la situazione è leggermente diversa, ovviamente si devono passare più parametri. È importante sottolineare che le sottoscrizioni durevoli sono sottoscrizioni con nome. È possibile assegnare loro un nome. Alle sottoscrizioni non durevoli, invece, non è possibile assegnare un nome. Il primo parametro che passiamo per creare una sottoscrizione durevole è il nome, poi specifichiamo la policy, se è condivisa o esclusiva, il callback per la ricezione dei messaggi e il tempo di scadenza della sottoscrizione. Bene.
// Multiple Subscribers can subscribe to this Subscription
SubscriptionPolicy sharedPolicy = SubscriptionPolicy.Shared;
// Only one Subscriber allowed on this Subscription
SubscriptionPolicy exclusivePolicy = SubscriptionPolicy.Exclusive;
IDurableTopicSubscription subscription =
topic.CreateDurableSubscription("ExampleSubscription",
sharedPolicy,
MessageReceivedCallback,
TimeSpan.FromHours(1));Pubblica messaggi su un argomento
Quindi, prima di passare alla diapositiva successiva, voglio mostrarvi l'applicazione di esempio che ho. Si tratta ovviamente di un'applicazione .NET. Come potete vedere, ho già scritto il codice. Ci sono quattro soluzioni. Ci sono i dati di esempio, ovviamente. Si tratta di un ordine. Abbiamo il publisher dell'ordine. Abbiamo un sottoscrittore dell'ordine primario e un sottoscrittore dell'ordine secondario. Bene.
Come potete vedere, nel mio editor di ordini ho un paio di metodi. Questo pubblica l'ordine in modo asincrono. Ovviamente creiamo il nostro oggetto argomento. Registriamo l'errore di consegna del messaggio e poi iteriamo su tutti i nostri ordini e li pubblichiamo come messaggi. La scadenza di questo messaggio è impostata a 15 secondi.
E le mie opzioni di consegna, in questo caso, sono impostate su tutti gli abbonati. Ok. E una volta che ogni messaggio viene consegnato, come puoi vedere ho aggiunto un ritardo di cinque secondi solo a scopo dimostrativo. E dovrebbe visualizzare ogni ordine dopo averlo inviato. Quindi c'è anche l'opzione asincrona qui. Chiama semplicemente il metodo di pubblicazione asincrono. E poi abbiamo anche la pubblicazione in blocco degli ordini. Ok. Quindi ecco come appare la mia notifica di errore di consegna del messaggio. Puoi vedere che visualizza semplicemente questo messaggio di log "Impossibile inviare l'ID dell'ordine" e poi invia anche i dettagli dell'ordine, ok. Quindi quello che farò ora è eseguire questa applicazione senza eseguire un abbonato.
Ma prima di farlo, diamo un'occhiata al nostro NCache Cluster di cache. Ho già creato una cache qui con il nome demoCache. Si tratta di un singolo nodo server. Ad esso è collegato un solo server ed è distribuito su un'istanza Docker. Voglio solo controllare le statistiche e assicurarmi che non ci siano dati. Perfetto. Ok. La cache è attiva e funzionante. Ora posso eseguire la mia applicazione. Voglio solo provare a ridurre a icona questa finestra per poter visualizzare le statistiche affiancate.
Avviamo quindi il publisher. Vi spiego cosa mi aspetto di vedere ora, dato che non abbiamo un'applicazione subscriber in esecuzione, verrà creato un topic nella nostra cache. Quindi, vediamo innanzitutto che il client è connesso. Ha creato un topic e ora sta pubblicando gli ordini uno alla volta in modo sincrono. E potete vedere che anche la dimensione della cache sta aumentando perché il topic è memorizzato al suo interno.
E poiché non esiste un'applicazione di sottoscrizione in ascolto per questi ordini, mi aspetto che l'editore venga informato dell'errore di invio del messaggio. Come potete vedere, l'invio dell'ordine con ID numero uno non è riuscito, semplicemente perché i messaggi sono scaduti. E mi aspetto che accada lo stesso per tutti e cinque gli ordini. Quindi, sì, l'invio degli ordini non è riuscito. Chiudo questa segnalazione per risparmiare tempo.
Crea un abbonamento per un argomento
La prossima demo che voglio mostrarvi è di avviare prima l'applicazione del sottoscrittore e poi quella del publisher. Bene. Prima di farlo, vi mostro come appare il codice qui. Come potete vedere, per prima cosa inizializzo la cache. Recuperiamo il nostro topic degli ordini. Creiamo una sottoscrizione durevole con il nome di "order subscription". Ma questo è solo un mio metodo privato, non la chiamata API vera e propria.
Ecco come appare la chiamata API. Ok. Stiamo passando il nome dell'abbonamento. La policy, che credo in questo caso sia "shared". Quindi è possibile avere più abbonati. Stiamo passando la callback per la ricezione del messaggio e un intervallo di tempo di un'ora. La callback è semplicemente così: "ordine ricevuto" e stampa i dettagli dell'ordine, come potete vedere qui. Ora eseguirò questa applicazione, che dovrebbe creare il mio abbonamento. Ok. Una volta creato, eseguirò la mia applicazione publisher e vedremo come si comporta.
Ecco il mio abbonato. Ecco il mio editore. Il tuo ordine con ID numero uno è stato pubblicato e puoi vedere che il nostro abbonato sta ricevendo tutti gli ordini. E quello che noti è che, poiché stavamo usando la pubblicazione sincrona, anche la nostra applicazione di abbonato elabora in modo sincrono perché è quello che fa di default. Puoi vedere che sta ricevendo tutto in ordine. Credo che questo errore di invio dell'ordine con ID numero cinque provenga dal messaggio precedente. Dato che ho chiuso l'applicazione prima di poterlo effettivamente inviare, lo ignoreremo per ora. Ma puoi vedere che tutto è in ordine.
Ora proviamo a pubblicare i nostri messaggi in modo asincrono e vediamo come cambia il comportamento. Modificherò la chiamata a questo metodo in asincrono. Credo che i parametri debbano essere gli stessi. Modificherò il tempo di scadenza a 60 secondi in modo da poter ignorare il ritardo del buffer. Ora vi mostrerò la differenza tra asincrono e sincrono. Eseguiremo l'applicazione del sottoscrittore. Assicuratevi che tutto sia chiuso. Sì. Una volta avviata, eseguiremo il publisher. Mi aspetto di vedere che l'applicazione del sottoscrittore riceve tutti gli ordini, ma non in sequenza. Come potete vedere, tutti i nostri ordini sono stati pubblicati e li abbiamo ricevuti, ma prima abbiamo ricevuto l'articolo numero tre, poi il numero due, poi il numero cinque e infine il numero uno. Quindi non sono in sequenza, ed è proprio questa la differenza tra asincrono e sincrono.
Quindi, per quanto riguarda il lato abbonato, lasciatemi mostrarvelo velocemente. Abbiamo selezionato la modalità sincrona per l'elaborazione degli ordini, ma per assicurarci che funzioni al 100%, è necessario che sia sincronizzata sia nel sottoscrittore che nel publisher. Come potete vedere, in questo caso la modalità di consegna predefinita è sincrona. È necessario passarla come parametro aggiuntivo per cambiarla da sincrona ad asincrona. Ok.
Abbonamenti durevoli
Proviamo quindi qualcos'altro. Vedremo la differenza tra sottoscrizioni durevoli e non durevoli. Vi mostrerò prima come le sottoscrizioni durevoli conservano i messaggi anche se l'applicazione va offline. Ok. Prima di tutto, assicuriamoci che tutto sia chiuso. Ora creeremo una sottoscrizione durevole. Eseguiremo prima questa applicazione e vedremo. E ora, per la pubblicazione, la riporterò alla modalità sincrona per la demo. Scusate gli errori di ortografia. Pubblicheremo quindi il nostro ordine in modo sincrono e imposterò il tempo di scadenza a 60 secondi per evitare il ritardo del buffer. E ora eseguiamolo. Mentre il mio publisher sta pubblicando i messaggi, chiuderò a metà l'applicazione subscriber, la riavvierò e vedremo come si comporta.
Quindi entrambi sono in esecuzione ora. L'ordine numero uno è stato ricevuto, lo chiuderò. Continuerà a pubblicare tutti gli ordini sull'argomento. E ora riattiverò il sottoscrittore. Quindi tenete presente che non creerà effettivamente un nuovo abbonamento. Dato che è durevole, ripristinerà semplicemente quello attualmente esistente. Quindi potete vedere che un paio di ordini sono stati pubblicati mentre la mia applicazione era offline, ed è ancora in grado di riceverli. Abbiamo già ricevuto il numero cinque, e mi aspetto che riceva anche il tre e il quattro e forse il due. Quindi anche l'ordine numero quattro è stato ricevuto. Apriamo rapidamente anche le statistiche della cache affiancate. Potete vedere che due client sono connessi. Uno è l'applicazione sottoscrittrice, l'altro è il publisher. E speriamo di riuscire a ricevere i messaggi prima che scadano, ecco perché ho impostato un tempo di 60 secondi. Quindi, dovrebbe essere da un momento all'altro. Ovviamente, speriamo che i messaggi non scadano davvero. E questo è l'ordine numero tre. Ok. Perfetto.
Abbonamenti non durevoli
Quindi ora chiuderò questa finestra e vedremo come si comportano le sottoscrizioni non durevoli nella stessa identica situazione. Permettetemi quindi di mostrarvi il codice per la sottoscrizione non durevole. Come potete vedere, otteniamo il nostro topic "create our orders". Quindi chiamiamo il metodo "create subscription" e passiamo semplicemente la funzione di callback. In questo modo verrà creata una sottoscrizione non durevole.
ITopic topic = cache.MessagingService.GetTopic(topicName);
// Create and register subscribers for the given topic
// Message received callback has to be passed
ITopicSubscription subscription = topic.CreateSubscription(MessageReceivedCallback);
// Used to notify the Subscriber when it receives a message
private static void MessageReceivedCallback(object sender, MessageEventArgs args) {
Console.WriteLine("Message received for topic " + args.TopicName);
}E ora lo eseguirò. Una volta avviato, avvierò l'applicazione publisher. Quindi è attiva e funzionante. Riceverà gli ordini. Ora la chiuderò, aspetterò che vengano pubblicati un paio di ordini e poi rieseguiremo il subscriber. Quindi rieseguiamolo ora. Come potete vedere, credo che gli ordini numero tre e quattro siano stati pubblicati mentre l'applicazione era offline. E non sarà in grado di riceverli. L'ordine numero cinque è stato ricevuto mentre era attivo e funzionante. Ecco perché lo vedete visualizzato. Ma chiaramente non riceverà nessuno degli altri ordini perché quei messaggi sono andati persi. Quindi, a scopo dimostrativo, aspetterò forse altri dieci secondi per mostrarvi che non conserva il messaggio.
Messaggi ordinati (lato editore)
Quindi, mentre ciò accade, lasciatemi anche mostrarvi come potete garantire l'ordine dei messaggi. Come ho spiegato in precedenza, è necessario assicurarsi che il publisher invii i messaggi in modo sincrono e quindi è possibile utilizzare un subscriber per elaborare i messaggi in modo sincrono. NCache Offre un'ulteriore funzionalità. È possibile specificare un nome di sequenza insieme alla chiamata al metodo publish, garantendo così che un determinato gruppo di messaggi venga consegnato in sequenza. Bene. Quindi, in questo frammento di codice, come potete vedere, creiamo un messaggio e lo pubblichiamo con questo nome di sequenza. E ripetiamo questa operazione per l'intero gruppo di messaggi. In questo modo, ovviamente, i messaggi vengono recapitati in ordine.
ITopic topic = cache.MessagingService.GetTopic(topicName);
for (int i = 0; i < 30; i++) {
Order order = FetchAnyOrderFromDB();
Message message = new Message(order);
// Specify a unique sequence name for the messages
string sequenceName = "OrderMessages";
// Publish message with the sequence name
topic.Publish(message, DeliveryOption.All, sequenceName, true);
}Messaggi ordinati (lato abbonato)
E per quanto riguarda il lato dell'abbonato, come vi ho già mostrato, è necessario specificare la modalità di consegna come parametro aggiuntivo per selezionare la modalità di consegna sincrona o asincrona. Bene.
ITopic topic = cache.MessagingService.GetTopic(topicName);
// Create and register subscribers for Topic
// Message received callback is specified
// DeliveryMode is set to async to ensure ordered messages
ITopicSubscription subscription = topic.CreateSubscription(MessageReceived, DeliveryMode.Sync);Tornando all'output degli ordini, si può notare che i messaggi sono scaduti. Il sistema non è in grado di riceverli. Quindi, tutto funziona praticamente come previsto.
Pianificare un demo
Ragazzi, questa è la demo. Spero che questo video vi sia piaciuto. Se volete saperne di più su NCache, suo Caratteristichee su come puoi integrarlo nella tua applicazione, contattaci per programma una demoGrazie mille per aver guardato questo video.