The real-time eventing feature in Coherence is great for relaying state changes to other systems or to users. However, sometimes not all changes need to or can be sent to consumers. For instance;
If rapid changes cannot be consumed or interpreted as fast as they are being sent. A user looking at changing Stock prices may only be able to interpret and react to 1 change per second.
A client may be using low bandwidth connection, so rapidly sending events will only result in them being queued and delayed
A large number of clients may need to be notified of state changes and sending 100 events p/s to 1000 clients cannot be supported with the available hardware, but 10 events p/s to 1000 clients can. Note this example assumes that many of the state changes are to the same value.
One simple approach to throttling Coherence cache events is to use a cache store to capture changes to one cache (data cache) and insert those changes periodically in another cache (events cache). Consumers interested in state changes to entires in the first cache register an interest (event listener) against the second event cache. By using the cache store write-behind feature rapid updates to the same cache entry are coalesced so that updates are merged and written at the interval configured to the event cache. The time interval at which changes are written to the events cache can easily be configured using the write-behind delay time in the cache configuration, as shown below.
<caching-schemes>
<distributed-scheme>
<scheme-name>CustomDistributedCacheScheme</scheme-name>
<service-name>CustomDistributedCacheService</service-name>
<thread-count>1</thread-count>
<backing-map-scheme>
<read-write-backing-map-scheme>
<scheme-name>CustomRWBackingMapScheme</scheme-name>
<internal-cache-scheme>
<local-scheme />
</internal-cache-scheme>
<cachestore-scheme>
<class-scheme>
<scheme-name>CustomCacheStoreScheme</scheme-name>
<class-name>com.oracle.coherence.test.CustomCacheStore</class-name>
<init-params>
<init-param>
<param-type>java.lang.String</param-type>
<param-value>{cache-name}</param-value>
</init-param>
<init-param>
<param-type>java.lang.String</param-type>
<!-- The name of the cache to write events to -->
<param-value>cqc-test</param-value>
</init-param>
</init-params>
</class-scheme>
</cachestore-scheme>
<write-delay>1s</write-delay>
<write-batch-factor>0</write-batch-factor>
</read-write-backing-map-scheme>
</backing-map-scheme>
<autostart>true</autostart>
</distributed-scheme>
</caching-schemes>
The cache store implementation to perform this throttling is trivial and only involves overriding the basic cache store functions.
public class CustomCacheStore implements CacheStore {
private String publishingCacheName;
private String sourceCacheName;
public CustomCacheStore(String sourceCacheStore, String publishingCacheName) {
this.publishingCacheName = publishingCacheName;
this.sourceCacheName = sourceCacheName;
}
@Override
public Object load(Object key) {
return null;
}
@Override
public Map loadAll(Collection keyCollection) {
return null;
}
@Override
public void erase(Object key) {
if (sourceCacheName != publishingCacheName) {
CacheFactory.getCache(publishingCacheName).remove(key);
CacheFactory.log("Erasing entry: " + key, CacheFactory.LOG_DEBUG);
}
}
@Override
public void eraseAll(Collection keyCollection) {
if (sourceCacheName != publishingCacheName) {
for (Object key : keyCollection) {
CacheFactory.getCache(publishingCacheName).remove(key);
CacheFactory.log("Erasing collection entry: " + key,
CacheFactory.LOG_DEBUG);
}
}
}
@Override
public void store(Object key, Object value) {
if (sourceCacheName != publishingCacheName) {
CacheFactory.getCache(publishingCacheName).put(key, value);
CacheFactory.log("Storing entry (key=value): " + key + "=" + value,
CacheFactory.LOG_DEBUG);
}
}
@Override
public void storeAll(Map entryMap) {
if (sourceCacheName != publishingCacheName) {
CacheFactory.getCache(publishingCacheName).putAll(entryMap);
CacheFactory.log("Storing entries: " + entryMap,
CacheFactory.LOG_DEBUG);
}
}
}
As you can see each cache store operation on the data cache results in a similar operation on event cache. This is a very simple pattern which has a lot of additional possibilities, but it also has a few drawbacks you should be aware of:
This event throttling implementation will use additional memory as a duplicate copy of entries held in the data cache need to be held in the events cache too - 2 if the event cache has backups
A data cache may already use a cache store, so a "multiplexing cache store pattern" must also be used to send changes to the existing and throttling cache store.
If you would like to try out this throttling example you can download it here. I hope its useful and let me know if you spot any further optimizations.
