奇怪的 Hazelcast IMap#put() 行为

Question

我的基于 Hazelcast 的程序可以在两种模式下工作：提交者和工作者。

提交者通过一些键将一些POJO放入分布式映射中，例如：hazelcastInstance.getMap(MAP_NAME).put(key, value);

Worker 有一个无限循环（内部有 Thread.sleep(1000L); 超时），它必须处理地图中的实体。现在我只是在这个循环中打印地图大小。

问题来了。我启动工人应用程序。然后我同时启动四个提交者（每个提交者向地图添加一个条目并终止其工作）。但是在所有提交者应用程序完成后，工作应用程序打印任意大小：有时它检测到只添加了一个条目，有时是两个，有时是三个（实际上它从来没有看到所有四个条目）。

这个简单的流程有什么问题？我在 Hazelcast 文档中读到 put() 方法是同步的，因此它保证在它 returns 之后，条目被放置到分布式地图并且 被复制 。但是在我的实验中好像不是这样。

UPD（代码）

提交者：

public void submit(String key) {
    Object mySerializableObject = ...
    IMap<String, Object> map = hazelcastInstance.getMap(MAP_NAME);
    map.putIfAbsent(key, mySerializableObject, TASK_TTL_IN_HOURS, TimeUnit.HOURS);
}

工人：

public void process() {
    while (true) {
        IMap<String, Object> map = hazelcastInstance.getMap(MAP_NAME);
        System.out.println(map.size());

        // Optional<Map.Entry<String, Object>> objectToProcess = getObjectToProcess();
        // objectToProcess.ifPresent(objectToProcess-> processObject(id, objectToProcess));
        try {
            Thread.sleep(PAUSE);
        } catch (InterruptedException e) {
            LOGGER.error(e.getMessage(), e);
        }
    }
}

我注释掉了 "processing" 部分本身，因为现在我只是想获得地图的一致状态。上面的代码每次打印不同的结果，例如：“4, 3, 1, 1, 1, 1, 1...”（所以它甚至可以暂时看到 4 个已提交的任务，但随后它们...消失了） .

UPD（日志）

工人：

...
tasksMap.size() = 0
tasksMap.size() = 0
tasksMap.size() = 0
tasksMap.size() = 0
tasksMap.size() = 1
tasksMap.size() = 2
tasksMap.size() = 2
tasksMap.size() = 2
tasksMap.size() = 2
tasksMap.size() = 2
...

提交者 1：

Before: tasksMap.size() = 0
After: tasksMap.size() = 1

提交者 2：

Before: tasksMap.size() = 1
After: tasksMap.size() = 4

提交者 3：

Before: tasksMap.size() = 1
After: tasksMap.size() = 2

提交者 4：

Before: tasksMap.size() = 3
After: tasksMap.size() = 4

Answer 1

嗯，我想，我已经找到问题所在了。据我了解，hazelcastInstance.getMap 返回的分布式 IMap 并不能保证数据在集群中的所有现有节点上都被复制：某些数据部分可能会被复制到某些节点，另一部分 - 到另一个节点。这就是为什么在我的示例中，一些提交的任务没有被复制到工作节点（永久工作），而是复制到其他一些提交者，这些提交者在提交后终止执行。所以这些条目在提交者退出时丢失了。

我通过将 hazelcastInstance.getMap 替换为 hazelcastInstance.getReplicatedMap 解决了这个问题。此方法 returns ReplicatedMap，据我所知，它保证放入其中的条目将被复制到集群的所有节点。所以现在我的系统一切正常。

奇怪的 Hazelcast IMap#put() 行为

Weird Hazelcat IMap#put() behaviour

java

multithreading

hazelcast

hazelcast-imap