Pub/Sub-Messaging und Streams

Viele Serveranwendungen müssen heutzutage die Arbeit mit anderen Anwendungen koordinieren und Nachrichten und Daten asynchron austauschen. Und oft geschieht dies in einer Umgebung mit hohem Transaktionsaufkommen. Für Umgebungen mit hohem Transaktionsaufkommen, NCache ist ein idealer Kandidat dafür asynchrone ereignisgesteuerte Kommunikation.

NCache bietet umfangreiche Messaging- und Stream-Funktionen, um den Anforderungen dieser Anwendungen gerecht zu werden Pub/Sub-Messaging, Kontinuierliche Abfragen, Ereignisse auf Artikelebene und Ereignisse auf Cache-Ebene. Jeder wird unten beschrieben.

Pub/Sub-Messaging

NCache bietet eine mächtige Pub/Sub-Messaging Plattform für Anwendungen, bei denen die Herausgeberanwendungen Nachrichten senden können, ohne zu wissen, wer die Abonnenten sind. Die Nachrichten können kategorisiert werden Themen Anschließend können Abonnenten bestimmte Themen abonnieren und diese Nachrichten asynchron empfangen.

• - Herausgeber: Die Nachricht wird veröffentlicht und ist für mehrere Abonnenten bestimmt. Der Herausgeber weiß weder, wer die Abonnenten sind, noch wie viele es sind.
• - Teilnehmer: Registriert sich, um Nachrichten zu einem bestimmten Thema zu empfangen. Der Abonnent weiß weder, wer der Herausgeber ist, noch wie viele andere Abonnenten es gibt.
• - Thema: Alle Nachrichten werden themenbezogen veröffentlicht und empfangen. Herausgeber veröffentlichen Nachrichten zu einem Thema, und Abonnenten erhalten Nachrichten zu diesem Thema. Sie können mehrere Themen haben. NCache.
• - Ihre Nachricht: Dies ist die eigentliche Nachricht, die von Herausgebern gesendet und von Abonnenten empfangen wird. Unterstützt werden alle serialisierbaren Objekte, einschließlich Binärdaten, JSON, XML und primitive Datentypen.

Sie können ganz einfach - NCache Pub/Sub-Messaging für verteilte Anwendungen Sobald Sie die Komponenten von Pub/Sub-Messaging und ihre Funktionsweise verstanden haben.

Unten ist ein Quellcode-Beispiel auf wie ein Publisher eine Nachricht veröffentlicht zu einem Thema und dann a Abonnent abonniert das Thema.

• .Netto
• Javac

ITopic topic = _cache.MessagingService.CreateTopic(topicName)

// Creating new message containing a “payload” 
var message = new Message(payload);

// Set the expiration TimeSpan of the message
message.ExpirationTime = TimeSpan.FromSeconds(5000);

// Publishing the above created message.
topic.Publish(message, DeliveryOption.All, true);

// Get the already created topic to subscribe to it
ITopic orderTopic = cache.MessagingService.GetTopic(topic);

// Create and register subscribers for the topic created
// MessageRecieved callback is registered
orderTopic.CreateSubscription(MessageReceived);

String topicName = "orderUpdates";
Topic topic = cache.getMessagingService().createTopic(topicName);

// Creating new message containing a “payload”
Customer payload = new Customer("David Jones", "XYZCompany");
var message = new Message(payload);

// Set the expiration TimeSpan of the message
message.setExpirationTime(TimeSpan.FromSeconds(5000));

// Publishing the above created message.
topic.publish(message, DeliveryOption.All, true);

// Get the already created topic to subscribe to it
Topic orderTopic = cache.getMessagingService().getTopic(topicName);

// Create and register subscribers for the topic created
orderTopic.createSubscription(new MessageReceivedListener() {
    @Override
     public void onMessageReceived(Object sender, MessageEventArgs args) {
// message received
     }
});

Kontinuierliche Abfrage (CQ)

Kontinuierliche Abfrage ist eine leistungsstarke Funktion von NCache Damit können Sie Änderungen an einem „Datensatz“ im verteilten Cache basierend auf SQL-ähnlichen Abfragen überwachen. Sie erstellen eine kontinuierliche Abfrage (CQ) mit SQL-ähnlichen Kriterien für den Cache und registrieren sie bei NCacheDadurch entfällt das regelmäßige Abfragen der Datenbank, die Datenbanklast wird reduziert und Ihre Anwendung zeigt stets den aktuellsten Datensatz in Echtzeit an. NCache dann beginnt mit der Überwachung aller Änderungen an diesem „Datensatz“ im Cache inklusive:

• - Hinzufügen: Wenn dem Cache ein Element hinzugefügt wird, das den CQ-Kriterien entspricht, wird der Client zusammen mit diesem Objekt benachrichtigt.
• - Update: Wenn im Cache ein Element aktualisiert wird, das den CQ-Kriterien entspricht, wird der Client zusammen mit diesem Objekt benachrichtigt.
• - Entfernen: Wenn ein Element aus dem Cache entfernt wird, das den CQ-Kriterien entspricht, wird der Client zusammen mit diesem Objekt benachrichtigt.

So können Sie eine kontinuierliche Abfrage mit bestimmten Kriterien erstellen und sich beim Server registrieren.

• .Netto
• Javac

// Query for required operation
string query = "SELECT $VALUE$ FROM FQN.Customer WHERE Country = ?";

var queryCommand = new QueryCommand(query);
queryCommand.Parameters.Add("Country", "USA");

// Create Continuous Query
var cQuery = new ContinuousQuery(queryCommand);

// Register to be notified when a qualified item is added to the cache
cQuery.RegisterNotification(new QueryDataNotificationCallback(QueryItemCallBack), EventType.ItemAdded, EventDataFilter.None);

// Register continuousQuery on server 
cache.MessagingService.RegisterCQ(cQuery);

String cacheName = "demoCache";

// Initialize an instance of the cache to begin performing operations:
Cache cache = CacheManager.getCache(cacheName);

// Query for required operation
String query = "SELECT $VALUE$ FROM com.alachisoft.ncache.sample.Customer WHERE country = ?";

var queryCommand = new QueryCommand(query);
queryCommand.getParameters().put("Country", "USA");

// Create Continuous Query
var cQuery = new ContinuousQuery(queryCommand);

// Register to be notified when a qualified item is added to the cache
cQuery.addDataModificationListener(new QueryDataModificationListener() {
	@Override
	public void onQueryDataModified(String key, CQEventArg args) {
		switch (args.getEventType())
		{
			case ItemAdded:
				// 'key' has been added to the cache
				break;
		}
	}
  }, EnumSet.allOf(EventType.class), EventDataFilter.None);

// Register continuousQuery on server
cache.getMessagingService().registerCQ(cQuery);

Weitere Details finden Sie unter wie man kontinuierliche Abfragen in Cache-Dokumenten verwendet.

Ereignisbenachrichtigungen

Kunden registrieren sich bei NCache bekommen Ereignisbenachrichtigungen an denen sie interessiert sind, indem sie eine Rückruffunktion bereitstellen. NCache stellt Filter mit Ereignissen bereit, die verwendet werden, um die Menge der mit dem Ereignis zurückgegebenen Informationen anzugeben. Diese Filter sind Keine (Standard), Metadaten und DatenMitMetadaten.

Folgende Ereignistypen werden unterstützt:

• - Ereignisse auf Artikelebene: Der Client hat sich registriert, um benachrichtigt zu werden, wenn ein bestimmtes zwischengespeichertes Element (basierend auf einem Schlüssel) aktualisiert oder aus dem Cache entfernt wird, und stellt einen Rückruf dafür bereit. Wenn dieses Element aktualisiert oder entfernt wird, wird der Client benachrichtigt.
• - Ereignisse auf Cache-Ebene: Cache-Level-Ereignisse sind die allgemeinen Ereignisse, die nützlich sein können, wenn alle Client-Anwendungen über Änderungen im Cache durch eine verbundene Client-Anwendung benachrichtigt werden müssen.

Der unten angegebene Code zeigt die Registrierung von Ereignissen auf Elementebene.

• .Netto
• Javac

public static void RegisterItemLevelEvents()
{
	// Get cache
	string cacheName = "demoCache";
	var cache = CacheManager.GetCache(cacheName); 

	// Register Item level events
	string key = "Product:1001";
	cache.MessagingService.RegisterCacheNotification(key, CacheDataNotificationCallbackImpl, EventType.ItemUpdated | EventType.ItemUpdated | EventType.ItemRemoved, EventDataFilter.DataWithMetadata);
}
public static void CacheDataNotificationCallbackImpl(string key, CacheEventArg cacheEventArgs)
{
	switch (cacheEventArgs.EventType)
	{
	   case EventType.ItemAdded:
		   // 'key' has been added to the cache
		   break;
	   case EventType.ItemUpdated:
		   // 'key' has been updated in the cache
		   break;
	   case EventType.ItemRemoved:
		   // 'key' has been removed from the cache
		   break;
	}
}

private static void RegisterItemLevelEvents() throws Exception {
        String cacheName = "demoCache";

        // Initialize an instance of the cache to begin performing operations:
        Cache cache = CacheManager.getCache(cacheName);
        String key = "Product:1001";
        cache.getMessagingService().addCacheNotificationListener(key, new CacheDataModificationListenerImpl(), EnumSet.allOf(EventType.class), EventDataFilter.DataWithMetadata);
    }
private static class CacheDataModificationListenerImpl implements CacheDataModificationListener {
	@Override
	public void onCacheDataModified(String key, CacheEventArg eventArgs) {
		switch (eventArgs.getEventType())
		{
			case ItemAdded:
				// 'key' has been added to the cache
				break;
			case ItemUpdated:
				// 'key' has been updated in the cache
				break;
			case ItemRemoved:
				// 'key' has been removed from the cache
				break;
		}
	}
	@Override
	public void onCacheCleared(String cacheName) {
		//cacheName cleared
	}
}

Für Ereignisse auf Cache-Ebene können Sie denselben Rückruf verwenden, der auch für Ereignisse auf Elementebene verwendet wird.

• .Netto
• Javac

public static void RegisterCacheLevelEvents()
{
	// Get cache
	string cacheName = "demoCache";
	var cache = CacheManager.GetCache(cacheName);

	// Register cache level events
	cache.MessagingService.RegisterCacheNotification(EventsSample.CacheDataNotificationCallbackImpl, EventType.ItemUpdated | EventType.ItemUpdated | EventType.ItemRemoved, EventDataFilter.DataWithMetadata); 
}

private static void RegisterCacheLevelEvents() throws Exception {
	String cacheName = "demoCache";

	// Initialize an instance of the cache to begin performing operations:
	Cache cache = CacheManager.getCache(cacheName);

	CacheEventDescriptor eventDescriptor = cache.getMessagingService().addCacheNotificationListener(new CacheDataModificationListenerImpl(), EnumSet.allOf(EventType.class), EventDataFilter.DataWithMetadata);
}

Warum NCache?

Für .NET- und Java-Anwendungen NCache ist die ideale Wahl für Pub/Sub-Messaging mit hohem Transaktionsvolumen, kontinuierliche Abfragen und Ereignisse. Hier ist der Grund.

• Extrem schnell & skalierbar: NCache ist extrem schnell, da es sich vollständig im Arbeitsspeicher befindet, und ermöglicht Reaktionszeiten von weniger als einer Millisekunde. Und es handelt sich um einen verteilten Cache, mit dem Sie dem Cache-Cluster Server hinzufügen können Erzielen Sie lineare Skalierbarkeit und bewältigen Sie extreme TransaktionslastenIm Gegensatz zu festplattenbasierten Messaging-Warteschlangen, NCache bietet In-Memory Pub/Sub Messaging und gewährleistet so Latenzzeiten im Submillisekundenbereich für Szenarien mit hohem Durchsatz.
• Hohe Verfügbarkeit durch Datenreplikation: NCache repliziert Daten im Cache auf intelligente Weise, ohne die Leistung zu beeinträchtigen. Sie verlieren also keine Daten (Ereignisse und Nachrichten), selbst wenn ein Cache-Server ausfällt.
• .NET- und Java-Apps Senden Sie sich gegenseitig eine Nachricht: NCache ermöglicht es .NET- und Java-Anwendungen, eine gemeinsame Messaging- und Streams-Plattform zu nutzen und Nachrichten aneinander zu senden, indem sie entweder die Nachrichten in JSON serialisieren oder JSON-Dokumente als Nachrichten senden.

Damit können Sie NCache als leistungsstarke und dennoch einfache Messaging-Plattform. Lesen Sie mehr über NCache über die unten stehenden Links und laden Sie eine voll funktionsfähige 30-Tage-Testversion herunter.

Was macht man als nächstes?

• NCache Details
• Herunterladen NCache
• Fordern Sie eine personalisierte LIVE-Demo an