在我看来，我对 JCache EntryListener 的回调机制有问题。我正在尝试设置几个 jcache 成员（hazelcast 实现），每个成员在我的本地机器上作为单个 java 应用程序运行（现在在 Ecipse 中为每个节点手动启动 main）。
节点在开始时只保存一个空缓存<Long, SpecificDataType>。使用默认的 hazelcast-default.xml，我自己以编程方式准备了一个最小配置，主要是注册一个 CacheEntryListener，它为创建/更新/删除/过期条目生成一个简单的 sysout。当我启动几个（三个或更多）成员时，我希望每个成员只打印一次修改后的键/值对，其中包含条目（作为操作条目或备份条目）。问题是，在某些情况下，sysout 会出现多次，在我看来，监听器被触发得太频繁（不止一次）。
例如，在一个简单客户端的帮助下创建一些条目时（它只是获取缓存并将密钥为 123689 的条目放入缓存中），系统输出“CREATE EVENT RECEIVED”在第三个成员中出现 4 次，甚至更常见于第四个，依此类推...
CREATE EVENT RECEIVED:
Key: 123689, Value:SpecificDataType[...]
CREATE EVENT RECEIVED:
Key: 123689, Value:SpecificDataType[...]
CREATE EVENT RECEIVED:
Key: 123689, Value:SpecificDataType[...]
似乎听众以某种方式指数化......我做错了什么？我错过了有关配置的任何内容吗？
代码：
public static void main(String[] args) {
Member member = Member.getInstance();
try {
Cache<Long, SpecificDataType> cache = member.getCache("SpecificDataTypeCache", SpecificDataType.class);
System.out.println(cache);
} catch (Exception e) {
e.printStackTrace();
}
// shutdown loop
boolean shutdown = false;
while (!shutdown) {
if (new Scanner(System.in).nextLine().equals("shutdown")) {
shutdown = true;
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// cache.clear();
member.shutdown();
}
public <T> Cache<Long, T> getCache(String name, Class<T> clazz) {
// configure the cache
MutableConfiguration<Long, T> config = new MutableConfiguration<Long, T>();
config.setStoreByValue(true).setTypes(Long.class, clazz)
.setExpiryPolicyFactory(AccessedExpiryPolicy.factoryOf(TEN_SEC))
.setStatisticsEnabled(false);
// create / get cache
Cache<Long, T> cache = cacheManager.getCache(name, Long.class, clazz);
if (cache == null) {
System.out.println("create cache");
cache = cacheManager.createCache(name, config);
// create the EntryListener
MyCacheEntryListener<Long, T> clientListener = new MyCacheEntryListener<Long, T>();
// using out listener, lets create a configuration
CacheEntryListenerConfiguration<Long, T> conf = new MutableCacheEntryListenerConfiguration<Long, T>(
FactoryBuilder.factoryOf(clientListener), null, false, true);
// register to cache
cache.registerCacheEntryListener(conf);
} else {
System.out.println("get cache");
}
return cache;
}
侦听器尽可能简单：
public class MyCacheEntryListener<K, V> implements CacheEntryCreatedListener<K, V>,
CacheEntryUpdatedListener<K, V>, CacheEntryExpiredListener<K, V>,
CacheEntryRemovedListener<K, V>, Serializable {
@Override
public void onCreated(Iterable<CacheEntryEvent<? extends K, ? extends V>> cacheEntryEvents)
throws CacheEntryListenerException {
System.out.println("CREATE EVENT RECEIVED: ");
System.out.println(printEvent(cacheEntryEvents));
}
@Override
public void onExpired(Iterable<CacheEntryEvent<? extends K, ? extends V>> cacheEntryEvents)
throws CacheEntryListenerException {
System.out.println("EXPIRE EVENT RECEIVED: ");
System.out.println(printEvent(cacheEntryEvents));
}
@Override
public void onRemoved(Iterable<CacheEntryEvent<? extends K, ? extends V>> cacheEntryEvents)
throws CacheEntryListenerException {
System.out.println("REMOVE EVENT RECEIVED: ");
System.out.println(printEvent(cacheEntryEvents));
}
@Override
public void onUpdated(Iterable<CacheEntryEvent<? extends K, ? extends V>> cacheEntryEvents)
throws CacheEntryListenerException {
System.out.println("UPDATE EVENT RECEIVED: ");
System.out.println(printEvent(cacheEntryEvents));
}
private String printEvent(Iterable<CacheEntryEvent<? extends K, ? extends V>> cacheEntryEvents) {
StringBuilder sb = new StringBuilder();
final Iterator<CacheEntryEvent<? extends K, ? extends V>> iterator = cacheEntryEvents
.iterator();
while (iterator.hasNext()) {
final CacheEntryEvent<? extends K, ? extends V> next = iterator.next();
sb.append("Key: ");
sb.append(next.getKey());
sb.append(", Value:");
sb.append(next.getValue());
sb.append("\n");
}
return sb.toString();
}
}
任何帮助，将不胜感激！