Mensagens e fluxos de Pub/Sub

Atualmente, muitos aplicativos de servidor precisam coordenar o trabalho com outros aplicativos e compartilhar mensagens e dados de maneira assíncrona. E, muitas vezes, isso é feito em um ambiente de alta transação. Para ambientes de alta transação, NCache é um candidato ideal para estes comunicações assíncronas orientadas por eventos.

NCache fornece um rico conjunto de recursos de mensagens e fluxos para atender às necessidades desses aplicativos, incluindo Mensagens do Pub/Sub, Consultas contínuas, Eventos de nível de item e Eventos de nível de cache. Cada um é descrito abaixo.

Mensagens do Pub/Sub

NCache fornece um poderoso Mensagens do Pub/Sub plataforma para aplicações onde as aplicações Publicadoras podem enviar mensagens sem saber quem são os Assinantes. As mensagens podem ser categorizadas em Temas e então os assinantes podem se inscrever em tópicos específicos e receber essas mensagens de forma assíncrona.

• - Fabricante : Publica a mensagem para ser consumida por vários assinantes. O publicador não sabe quem são os assinantes nem quantos são.
• - Assinante: O assinante se cadastra para receber mensagens sobre um tópico específico. Ele não sabe quem é o editor nem quantos outros assinantes existem.
• - Tópico: Todas as mensagens são publicadas e recebidas em relação a Tópicos. Os editores publicam mensagens em um Tópico e os assinantes recebem mensagens dele. Você pode ter vários Tópicos em um Tópico. NCache.
• - Mensagem: Esta é a mensagem real enviada pelos Publicadores e consumida pelos Assinantes. Suporta todos os objetos serializáveis, incluindo dados binários, JSON, XML e tipos primitivos.

Você pode facilmente usar NCache Mensagens Pub/Sub para aplicativos distribuídos depois de entender os componentes das mensagens do Pub/Sub e seu funcionamento.

Abaixo está um exemplo de código-fonte em como um editor publica uma mensagem para um tópico e, em seguida, um inscrito se inscreve no tópico.

• . NET
• Java

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
     }
});

Consulta Contínua (CQ)

Consulta Contínua é um recurso poderoso NCache que permite monitorar alterações em um “conjunto de dados” no cache distribuído com base em consultas semelhantes a SQL. Você cria uma Consulta Contínua (CQ) com critérios semelhantes a SQL no cache e a registra com NCacheIsso elimina a necessidade de consultar o banco de dados repetidamente, reduzindo a carga sobre ele e garantindo que seu aplicativo sempre exiba o conjunto de dados mais atualizado em tempo real. NCache então começa a monitorar quaisquer alterações neste “conjunto de dados” no cache incluindo:

• - Adicionar: se um item for adicionado ao cache que corresponda aos critérios CQ, o cliente será notificado junto com esse objeto.
• - Update: se um item for atualizado no cache que corresponda aos critérios CQ, o cliente será notificado junto com esse objeto.
• - Remover: se um item for removido do cache que corresponda aos critérios CQ, o cliente será notificado junto com esse objeto.

Aqui está como você pode criar uma Consulta Contínua com critérios específicos e se registrar no servidor.

• . NET
• Java

// 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);

Para mais detalhes, você pode ver como usar a consulta contínua em documentos de cache.

Notificações de eventos

Clientes cadastram-se com NCache para receber notificações de eventos eles estão interessados, fornecendo uma função de retorno de chamada. NCache fornece filtros com eventos que são usados ​​para especificar a quantidade de informações retornadas com o evento. Esses filtros são Nenhum (padrão), metadados e Dados com Metadados.

Os seguintes tipos de eventos são suportados:

• - Eventos de nível de item: o cliente registrado para ser notificado sempre que um item específico em cache (baseado em uma chave) for atualizado ou removido do cache e fornece um retorno de chamada para ele. Quando este item é atualizado ou removido, o cliente é notificado.
• - Eventos de nível de cache: Os eventos de nível de cache são os eventos gerais que podem ser úteis quando todos os aplicativos cliente precisam ser notificados sobre qualquer modificação no cache por qualquer aplicativo cliente conectado.

O código fornecido a seguir mostra o registro de eventos em nível de item.

• . NET
• Java

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
	}
}

Para eventos em nível de cache, você pode usar o mesmo retorno de chamada usado nos eventos em nível de item.

• . NET
• Java

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);
}

Porque NCache?

Para aplicações .NET e Java, NCache é a escolha ideal para realizar mensagens Pub/Sub de alta transação, consultas contínuas e eventos. Aqui está o porquê.

• Extremamente rápido e escalável: NCache é extremamente rápido por estar totalmente na memória e oferece tempos de resposta inferiores a milissegundos. E é um cache distribuído que permite adicionar servidores ao cluster de cache para alcance escalabilidade linear e lide com cargas de transações extremasAo contrário das filas de mensagens baseadas em disco, NCache Oferece mensagens Pub/Sub em memória, garantindo latência inferior a um milissegundo para cenários de alto rendimento.
• Alta disponibilidade por meio de replicação de dados: NCache replica de forma inteligente os dados no cache sem comprometer o desempenho. Assim, você não perderá nenhum dado (eventos e mensagens), mesmo que um servidor de cache fique inativo.
• Aplicativos .NET e Java trocam mensagens: NCache permite que aplicativos .NET e Java compartilhem uma plataforma comum de mensagens e fluxos e enviem mensagens entre si serializando as mensagens em JSON ou enviando documentos JSON como mensagens.

Com isso, você pode usar NCache como uma plataforma de mensagens poderosa e simples. Leia mais sobre NCache nos links abaixo e faça o download de um teste de 30 dias totalmente funcional.

O que fazer a seguir?

• NCache Detalhes
• Baixar NCache
• Solicite uma demonstração AO VIVO personalizada