0

带代码：

try (Grid grid = GridGain.start(AnswerYagoFactTests.class.getResource("yago.gridgain.xml"))) {
    GridCache<Integer, YagoRule> cache = grid.cache("yagoRules");
    grid.compute().broadcast(new Runnable() {
        @Override
        public void run() {
            try {
                log.info("Cache formerly has size={} offheap={} overflow={}",
                        cache.size(), cache.offHeapEntriesCount(), cache.overflowSize());
            } catch (GridException e) {
                log.error("Cannot get overflow size", e);
            }
        }
    }).get();

    log.info("1 is {}", cache.get(1));
    grid.compute().apply(new GridClosure<String, String>() {
        @Override
        public String apply(String e) {
            log.info("Found {} YAGO rules", cache.size());
            cache.forEach(new GridInClosure<GridCacheEntry<Integer,YagoRule>>() {
                @Override
                public void apply(GridCacheEntry<Integer, YagoRule> e) {
                    log.info("Processing rule #{} {}", e.getKey(), e.getValue());
                }
            });
            return null;
        }
    }, msg).get();
}

在 3 节点配置中。GridGain 选择一个节点（似乎是随机的），然后仅处理该节点中的每个“处理规则”。

我想做的是forEach并行，所以理想情况下对于 3 个节点和 30 个条目，每个节点应该处理 10 个条目。缓存是partitioned这样的，每个节点都有自己的条目。

4

1 回答 1

1

在您的代码中，不是grid.compute().apply(...)尝试使用，而是grid.compute().broadcast(...)将闭包广播到网格中的所有节点。

此外，如果您只需要遍历主集（不包括备份），您可以执行以下操作：

grid.compute().broadcast(new GridClosure<String, Integer>() {
    @Override public String apply(String e) {
        for (GridCacheEntry<Integer, YagoRule> e : cache.primaryEntrySet()
            log.info("Processing rule #{} {}", e.getKey(), e.getValue());
    }
}, msg).get();
于 2014-07-06T02:34:40.123 回答