java - Hazelcast JCache CacheEntryListener 触发过于频繁

Question

在我看来，JCache EntryListener 的回调机制有问题。我正在尝试设置几个 jcache 成员(hazelcast 实现)，每个成员在我的本地计算机上作为单个 java 应用程序运行(目前在 Ecipse 中为每个节点手动启动 main)。

节点一开始只保存一个空缓存 <Long, SpecificDataType> 。使用默认的 hazelcast-default.xml，我自己以编程方式准备了一个最小的配置，主要是注册一个 CacheEntryListener，它为创建/更新/删除/过期的条目生成一个简单的系统输出。当我启动多个(三个或更多)成员时，我希望每个成员只打印一次修改后的键/值对，其中包含条目(作为操作条目或备份条目)。问题是，在某些情况下，系统输出会多次显示，在我看来，监听器被触发得太频繁(不止一次)。

例如，在简单客户端的帮助下创建某个条目(它只是获取缓存并将 key 为 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();
}

}

如有任何帮助，我们将不胜感激!

Answer 1

我已尝试使用 4 个实例和 3 个备份计数的示例，并且监听器仅按预期收到一次通知。

• 您确定没有多次注册同一个监听器吗？
• 您能否在 CREATE EVENT RECEIVED 消息之后打印 MyCacheEntryListener 本身？
• 您使用哪个 Hazelcast 版本？

问候。