14

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

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

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

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

这个简单的流程有什么问题?我在 Hazelcast 文档中读到该put()方法是同步的,因此它保证在它返回后,条目被放置到分布式映射并被复制。但在我的实验中似乎并非如此。

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

我注释掉了“处理”部分本身,因为现在我只是想获得地图的一致状态。上面的代码每次打印不同的结果,例如:“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
4

1 回答 1

7

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

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

于 2016-04-30T13:17:46.750 回答