Messaggistica Pub/Sub: Iscriviti a un argomento
Nel modello di messaggistica Pub/Sub, un abbonato (o applicazione client) si registra a un argomento tramite un abbonamento. NCache Fornisce diversi tipi di sottoscrizioni agli argomenti Pub/Sub, tra cui sottoscrizioni non durevoli, sottoscrizioni durevoli che sopravvivono alle disconnessioni del client e sottoscrizioni basate su pattern che utilizzano caratteri jolly per sottoscrivere più argomenti. Per i dettagli, vedere tipi di sottoscrizioni agli argomenti Pub/Sub.
Prerequisiti
Prima di usare il NCache Per le API lato client, assicurarsi che siano soddisfatti i seguenti prerequisiti:
- Installa i seguenti pacchetti NuGet nella tua applicazione client .NET:
- Includere i seguenti spazi dei nomi nell'applicazione:
- La cache deve essere in esecuzione.
- Assicurati che i dati aggiunti lo siano serializzabile.
- Per i dettagli dell'API, fare riferimento a: ICache, CacheItem, Argomento, IDurableTopicSubscription, Crea abbonamento, Creaabbonamento durevole,Nome argomento, NomeAbbonamento, IMessagingService, GetTopic, Politica di abbonamento, Annulla l'iscrizione, MessageEventArgs, Modalità di consegna, Argomenti Opzioni di ricerca, ITopicAbbonamento, Errore di consegna del messaggio, Carico utile, Servizio di messaggistica, Messaggio Ricevuto Richiamata.
- Aggiungi le seguenti dipendenze Maven per la tua applicazione client Java in
pom.xml file:
<dependency>
<groupId>com.alachisoft.ncache</groupId>
<!--for NCache Enterprise-->
<artifactId>ncache-client</artifactId>
<version>x.x.x</version>
</dependency>
- Importa i seguenti pacchetti nell'applicazione client Java:
- La cache deve essere in esecuzione.
- Assicurati che i dati aggiunti lo siano serializzabile.
- Per i dettagli dell'API, fare riferimento a: Cache, CacheItem, Argomento, Sottoscrizione all'argomento, getMessagingService, getTopic, createDurableSubscription, Politica di abbonamento, Intervallo di tempo, Annulla l'iscrizione, Ascoltatore del messaggio ricevuto, onMessaggio ricevuto, MessageEventArgs, getMessage, getPayload, createSubscription, Modalità di consegna, Argomenti Opzioni di ricerca, addMessageDeliveryFailureListener, getNome.
- Installa i seguenti pacchetti nella tua applicazione client Python:
- Importa i seguenti pacchetti nella tua applicazione:
- La cache deve essere in esecuzione.
- Assicurati che i dati aggiunti lo siano serializzabile.
- Per i dettagli dell'API, fare riferimento a: Cache, CacheItem, get_topic, get_messaging_service, create_durable_subscription, get_carico utile, Intervallo di tempo, MessageEventArgs, ricevi_messaggio, crea_abbonamento, Annulla l'iscrizione, add_message_delivery_failure_listener, Argomento, Sottoscrizione all'argomento, Politica di abbonamento, Modalità di consegna, Argomenti Opzioni di ricerca.
- Installa uno dei seguenti pacchetti NuGet nella tua applicazione client .NET:
- Enterprise:
Install-Package Alachisoft.NCache.SDK -Version 4.9.1.0
- Crea una nuova applicazione console.
- Assicurati che i dati aggiunti lo siano serializzabile.
- Aggiungi NCache Referenze individuando
%NCHOME%\NCache\bin\assembly\4.0 e aggiungendo Alachisoft.NCache.Web and Alachisoft.NCache.Runtime come appropriato.
- Includi il
Alachisoft.NCache.Web.Caching and Alachisoft.NCache.Runtime.Caching spazi dei nomi nella tua applicazione.
- Per saperne di più sul NCache API legacy, scaricala NCache 4.9 documenti disponibili come file .zip file sul Alachisoft Sito web.
Metodi per creare un abbonamento
Qui descriviamo come creare abbonamenti non durevoli, abbonamenti durevoli, abbonamenti multipli e abbonamenti asincroni nel modello di messaggistica Pub/Sub.
Abbonamenti non durevoli
Migliori ITopic/Topic l'interfaccia facilita la creazione di un Abbonamento non durevole e pubblicazione di messaggi relativi all'argomento. Il metodo di creazione di sottoscrizione registra una sottoscrizione non durevole relativa a un argomento, se l'argomento esiste. Consente all'abbonato di registrarsi MessageReceivedCallback sull'argomento affinché possa ricevere i messaggi pubblicati.
L'esempio di codice seguente esegue le seguenti operazioni:
- Ottieni l'argomento di interesse esistente, ad esempio,
NewBeverages.
- Crea un abbonamento per ogni argomento.
- Registra eventi affinché gli abbonati ricevano i messaggi una volta pubblicati sull'argomento.
try
{
// Precondition: Cache is already connected
// NewBeverages is the name of the Topic created beforehand
string topicName = "NewBeverages";
// Get the Topic
ITopic topic = cache.MessagingService.GetTopic(topicName);
// If Topic exists, Create subscription
if (topic != null)
{
// Create and register subscribers for order Topic
// Message received callback is specified
ITopicSubscription subscription = topic.CreateSubscription(MessageReceived);
}
}
catch (OperationFailedException ex)
{
if (ex.ErrorCode == NCacheErrorCodes.TOPIC_DISPOSED)
{
// Specified topic has been disposed
}
if (ex.ErrorCode == NCacheErrorCodes.DEFAULT_TOPICS)
{
// Operation cannot be performed on default topics,
// Get user-defined topics instead
}
else
{
// Exception can occur due to:
// Connection Failures
// Operation Timeout
// Operation performed during state transfer
}
}
catch (Exception ex)
{
// Any other generic exception like ArgumentNullException or ArgumentException
// Topic name is null/empty
}
try
{
// Precondition: Cache is already connected
// NewBeverages is the name of the Topic created beforehand
String topicName = "NewBeverages";
// Get the Topic
Topic ordertopic = cache.getMessagingService().getTopic(topicName);
// If Topic exists, Create subscription
if (ordertopic != null) {
// Create and register subscribers for Topic
ordertopic.createSubscription(new PubSubMessageReceivedListener());
System.out.println("Subscriber registered for Topic: " + ordertopic.getName());
} else {
System.out.println("Topic does not exist.");
}
}
catch (OperationFailedException ex)
{
if (ex.ErrorCode == NCacheErrorCodes.TOPIC_DISPOSED) {
// Specified topic has been disposed
}
if (ex.ErrorCode == NCacheErrorCodes.DEFAULT_TOPICS) {
// Operation cannot be performed on default topics,
// Get user-defined topics instead
} else {
// Exception can occur due to:
// Connection Failures
// Operation Timeout
// Operation performed during state transfer
}
}
catch (Exception ex)
{
// Any other generic exception like IllegalArgumentException or NullPointerException
}
try:
# Precondition: Cache is already connected
# NewBeverages is the name of the Topic created beforehand
topic_name = "NewBeverages"
# Get the Topic
order_topic = cache.get_messaging_service().get_topic(topic_name)
# If Topic exists, Create subscription
if order_topic is not None:
# Create and register subscribers for the Topic
# Message received callback is specified
order_subscriber = order_topic.create_subscription(on_message_received)
except Exception as ex:
# Exception can occur due to:
# - Connection failures
# - Operation timeout
# - Topic does not exist / invalid arguments
print("Operation failed: " + str(ex))
try
{
// This is an async method
// Precondition: Cache is already connected
// NewBeverages is the name of the Topic created beforehand
let topicName = "NewBeverages";
// NewBeverages is the name of the Topic created beforehand
let messagingService = await cache.getMessagingService();
let ordertopic = await messagingService.getTopic(topicName);
// If Topic exists, Create subscription
if (ordertopic != null)
{
// Create and register subscribers for Topic
// Message received callback is specified
let messageListener = new ncache.MessageReceivedListener(onMessageReceived);
let ordersubscriber = await ordertopic.createSubscription(messageListener);
}
}
catch (error) {
// Handle any errors
}
try
{
// Using NCache Enterprise 4.9.1
// Precondition: Cache is already connected
// NewBeverages is the name of the Topic created beforehand
string topicName = "NewBeverages";
// Get Order Topic
ITopic topic = cache.MessagingService.GetTopic(topicName);
// If Topic exists, Create subscription
if (topic != null)
{
// Create and register subscribers for Topic
// Message received callback is specified
ITopicSubscription ordSubscriber = topic.CreateSubscription(MessageReceived);
}
}
catch (OperationFailedException ex)
{
if (ex.ErrorCode == NCacheErrorCodes.TOPIC_DISPOSED)
{
// Specified topic has been disposed
}
if (ex.ErrorCode == NCacheErrorCodes.DEFAULT_TOPICS)
{
// Operation cannot be performed on default topics,
// Get user-defined topics instead
}
else
{
// Exception can occur due to:
// Connection Failures
// Operation Timeout
// Operation performed during state transfer
}
}
catch (Exception ex)
{
// Any other generic exception like ArgumentNullException or ArgumentException
// Topic name is null/empty
}
Note:
Per garantire che l'operazione sia a prova di errore, si consiglia di gestire eventuali potenziali eccezioni all'interno dell'applicazione, come spiegato in Gestione dei guasti.
Abbonamenti durevoli
Migliori IDurableTopicSubscription l'interfaccia facilita la creazione Abbonamenti durevoli e pubblicare messaggi contro l'argomento se esiste. Consente all'abbonato di registrare un MessageReceivedCallback rispetto all'argomento, in modo che possa ricevere i messaggi pubblicati.
Abbonamenti durevoli condivisi
Il codice seguente illustra la politica di abbonamento condiviso.
// Precondition: Cache is already connected
// DiscountedBeverages is the name of the Topic created beforehand
string topicName = "DiscountedBeverages";
string subscriptionName = "DiscountedBeveragesSubscription";
// Get the Topic
ITopic ordertopic = cache.MessagingService.GetTopic(topicName);
if (ordertopic != null)
{
// Create and register subscribers for Topic
// Message received callback is specified
// The subscription policy is shared which means that the subscription can have more than one subscriber
IDurableTopicSubscription ordersubscription = ordertopic.CreateDurableSubscription (subscriptionName, SubscriptionPolicy.Shared, MessageReceived, TimeSpan.FromMinutes(20));
}
// Precondition: Cache is already connected
// DiscountedBeverages is the name of the Topic created beforehand
String topicName = "DiscountedBeverages";
Topic ordertopic = cache.getMessagingService().getTopic(topicName);
// Get the Topic
if (ordertopic != null) {
// Create and register subscribers for Topic
// The subscription policy is shared which means that the subscription can have more than one subscriber
System.out.println("Topic retrieved: " + ordertopic.getName());
ordertopic.createDurableSubscription(subscriptionName, SubscriptionPolicy.Shared, new PubSubMessageReceivedListener(), TimeSpan.FromMinutes(20));
System.out.println("Durable subscriber registered for Topic: " + ordertopic.getName());
} else {
System.out.println("Topic does not exist.");
}
# Precondition: Cache is already connected
# DiscountedBeverages is the name of the Topic created beforehand
topic_name = "DiscountedBeverages";
subscription_name = "DiscountedBeveragesSubscription";
# Get the Topic
order_topic = cache.get_messaging_service().get_topic(topic_name);
if order_topic is not None:
# Create and register subscribers for Topic
# MessageReceived callback is specified below
# The subscription policy is Shared which means that the subscription can have more than one subscriber
order_Subscriber = order_topic.create_durable_subscription(subscription_name, SubscriptionPolicy.SHARED, on_message_received, TimeSpan.from_minutes(20));
// Precondition: Cache is already connected
// DiscountedBeverages is the name of the Topic created beforehand
let topicName = "DiscountedBeverages";
let subscriptionName = "DiscountedBeveragesSubscription";
// Get the Topic
let messagingService = await cache.getMessagingService();
let orderTopic = await messagingService.getTopic(topicName);
let messageListener = new ncache.MessageReceivedListener(onMessageReceived);
if (orderTopic !=null)
{
// Create and register subscribers for Topic
// Message received callback is specified below
// The subscription policy is Shared which means that the subscription can have more than one subscriber
let timeSpan = new TimeSpan(0, 20, 0);
let orderSubscriber = orderTopic.createDurableSubscription(subscriptionName, SubscriptionPolicy.Shared, messageListener, timeSpan, DeliveryMode.Reliable);
}
Abbonamenti durevoli esclusivi
Il codice seguente illustra la politica di Abbonamento Esclusivo.
// Precondition: Cache is already connected
// orderTopic is the name of the Topic created beforehand
string topicName = "orderTopic";
string subscriptionName = "orderTopicName";
// Get the Topic
ITopic orderTopic = cache.MessagingService.GetTopic(topicName);
// Create and register subscribers for Topic
// The subscription policy is exclusive which means that the subscription can have only one subscriber
IDurableTopicSubscription orderSubscriber = orderTopic.CreateDurableSubscription(subscriptionName, SubscriptionPolicy.Exclusive, MessageReceived, TimeSpan.FromMinutes(20));
// Precondition: Cache is already connected
// Get the Topic
Topic orderTopic = cache.getMessagingService().getTopic(topicName);
if (orderTopic != null) {
System.out.println("Topic retrieved: " + orderTopic.getName());
// Create and register subscribers for Topic
// The subscription policy is exclusive which means that the subscription can have only one subscriber
orderTopic.createDurableSubscription(subscriptionName,SubscriptionPolicy.Exclusive, new PubSubMessageReceivedListener(),TimeSpan.FromMinutes(20));
System.out.println("Durable subscriber registered for Topic: " + orderTopic.getName());
} else {
System.out.println("Topic does not exist.");
}
# Precondition: Cache is already connected
# orderTopic is the name of the Topic created beforehand
topic_name = "orderTopic"
subscription_name = "orderTopicName"
# Get the Topic
order_topic = cache.get_messaging_service().get_topic(topic_name)
# Create and register subscribers for orderTopic
# The subscription policy is exclusive which means that the subscription can have more than one subscriber
order_subscriber = order_topic.create_durable_subscription(subscription_name, SubscriptionPolicy.EXCLUSIVE, on_message_received, TimeSpan.from_minutes(20))
// Precondition: Cache is already connected
// orderTopic is the name of the Topic created beforehand
let topicName = "orderTopic";
let subscriptionName = "orderTopicName";
// Get the Topic
let messagingService = await cache.getMessagingService();
let orderTopic = await messagingService.getTopic(topicName);
let messageListener = new ncache.MessageReceivedListener(onMessageReceived);
// Create and register subscribers for order Topic
// Message received callback is specified below
let timeSpan = new TimeSpan(0, 20, 0);
let orderSubscriber = orderTopic.createDurableSubscription(subscriptionName, SubscriptionPolicy.Exclusive, messageListener, timeSpan, DeliveryMode.Reliable);
Abbonamenti multipli
Utilizzando questo metodo di sottoscrizione, gli utenti possono fornire modelli per iscriversi a più argomenti (Topic) con una singola chiamata. A tal fine, è importante che gli argomenti corrispondenti al modello esistano già sul server. Una volta create correttamente le sottoscrizioni sull'argomento basato sul modello, gli utenti riceveranno i messaggi pubblicati sugli argomenti che corrispondono al modello. Inoltre, se un argomento viene creato dopo che la sottoscrizione basata sul modello è stata registrata, registrerà l'utente anche su tale argomento. Analogamente, per quanto riguarda la disiscrizione da un argomento, questo metodo annulla l'iscrizione dell'utente a tutti gli argomenti corrispondenti al modello fornito, senza influire su altre sottoscrizioni che utilizzano la stessa chiamata.
Note:
- Un abbonato può ottenere solo argomenti basati su pattern e non gli è consentito crearli.
- Un modello può essere utilizzato dall'editore solo per ricevere notifiche di errore.
Caratteri jolly supportati
Il metodo di sottoscrizione basato su pattern supporta i seguenti tre caratteri jolly:
* : Zero o molti caratteri. Ad esempio, bl* si abbona a nero, blu e sfocato, ecc.? : Un carattere qualsiasi. Ad esempio, h?t si abbona a hit, hot and hat, ecc.[] : Gamma di caratteri. Ad esempio, b[ae]t si iscrive a bet e bat, ma non bit.
Creazione di un abbonamento con caratteri jolly
L'esempio seguente crea una sottoscrizione fornendo un modello in base al quale viene sottoscritto l'argomento corrispondente al modello.
// Precondition: Cache is already connected
// Create Topic name for all Topics with suffix Beverages
// Only ? * [] wildcards supported
string topicName = "*Beverages";
// Get the Topic
ITopic topic = cache.MessagingService.GetTopic(topicName, TopicSearchOptions.ByPattern);
// If Topic exists, Create subscription
if (topic != null)
{
// Create and register subscribers for Topic
ITopicSubscription subscription = topic.CreateSubscription(MessageReceived);
}
// Precondition: Cache is already connected
// Define Topic name pattern
String topicNamePattern = "*ages";
// Get all Topics that fulfill the pattern
Topic topic = cache.getMessagingService().getTopic(topicNamePattern, TopicSearchOptions.ByPattern);
if (topic != null) {
System.out.println("Matching Topic retrieved: " + topic.getName());
// Create and register notifications for Topic
topic.createSubscription(new PubSubMessageReceivedListener());
System.out.println("Subscription created for Topic: " + topic.getName());
} else {
System.out.println("No matching Topic found.");
}
# Precondition: Cache is already connected
# Create Topic name for all Topics with suffix Beverages
# Only ? * [] wildcards supported
topic_name = "*Beverages*"
# Get the Topic
order_topic = cache.get_messaging_service().get_topic(topic_name, TopicSearchOptions.BY_PATTERN)
if order_topic is not None:
# Create and register subscribers for Topic
order_subscriber = order_topic.create_subscription(on_message_received)
// Precondition: Cache is already connected
// Create Topic name for all Topics with suffix Beverages
// Only ? * [] wildcards supported
let topicName = "*Beverages";
// Get the Topic
let messagingService = await cache.getMessagingService();
let orderTopic = await messagingService.getTopic(topicName, TopicSearchOptions.ByPattern);
let messageListener = new ncache.MessageReceivedListener(onMessageReceived);
// If Topic exists, Create subscription
if (orderTopic!=null)
{
// Create and register subscribers for Topic
let orderSubscriber = orderTopic.createSubscription(messageListener);
}
Abbonamenti asincroni
La modalità di consegna può essere specificata durante la creazione di abbonamenti per i messaggi ordinati e può esserlo sync or async. L'esempio seguente crea la sottoscrizione utilizzando CreateSubscription metodo. Si consiglia di utilizzare sync modalità per i messaggi ordinati e async modalità altrimenti per ottenere prestazioni elevate.
Durante la creazione di una sottoscrizione, è possibile specificare la modalità di consegna che può essere sincrona o asincrona. In particolare, è possibile utilizzare la modalità di consegna sincrona messaggi ordinati, mentre se non si utilizzano messaggi ordinati, la modalità asincrona migliora le prestazioni.
Note:
Per impostazione predefinita, la modalità di consegna dei messaggi è impostata su sincrona.
// Precondition: Cache is already connected
// NewBeverages is the name of the Topic created beforehand
string topicName = "NewBeverages";
// Get the Topic
ITopic topic = cache.MessagingService.GetTopic(topicName);
// If Topic exists, Create subscription
if (topic != null)
{
// Create and register subscribers for Topic
// Message received callback is specified
// DeliveryMode is set to async
ITopicSubscription subscription = topic.CreateSubscription(MessageReceived, DeliveryMode.Async);
}
// Precondition: Cache is already connected
// NewBeverages is the name of the Topic created beforehand
String topicName = "NewBeverages";
// Get the Topic
Topic topic = cache.getMessagingService().getTopic(topicName);
if (topic != null)
{
// Create and register subscribers for Topic
// Message received callback is specified
// DeliveryMode is set to async
topic.createSubscription(new PubSubMessageReceivedListener(),DeliveryMode.Async);
} else
{
System.out.println("Topic does not exist.");
}
# Precondition: Cache is already connected
# NewBeverages is the name of the Topic created beforehand
topic_name = "NewBeverages"
# Get the Topic
order_topic = cache.get_messaging_service().get_topic(topic_name)
if order_topic is not None:
# Create and register subscribers for Topic
# Message received callback is specified
# DeliveryMode is set to async
order_subscriber = order_topic.create_subscription(on_message_received, DeliveryMode.SYNC)
// Precondition: Cache is already connected
// This is an async method
// NewBeverages is the name of the Topic created beforehand
let topicName = "NewBeverages";
// Get the Topic
let messagingService = await cache.getMessagingService();
let orderTopic = await messagingService.getTopic(topicName);
let messageListener = new ncache.MessageReceivedListener(onMessageReceived);
if (orderTopic != null)
{
// Create and register subscribers for Topic
// Message received callback is specified
let subscription = await orderTopic.createSubscription(messageListener, DeliveryMode.Sync);
}
Note:
Puoi anche annullare l'iscrizione a un argomento.
Risorse addizionali
NCache fornisce un'applicazione di esempio per Pub/Sub su GitHub.
Vedere anche
.NETTO: Alachisoft.NCache.Memorizzazione.della.cache spazio dei nomi.
Giava: com.alachisoft.ncache.runtime.caching spazio dei nomi.
Pitone: servizi.ncache.client modulo.
Node.js: Classe TopicSubscription classe.
