消息传递和流

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

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

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

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