java - 如何使apache ignit使用所有可用于过滤缓存的CPU功能？

Question

我试图强制Apache Ignite 1.5.0.final在单个节点上使用所有可用的CPU功能来并行处理本地缓存数据，但是我可以清楚地看到它没有使用所有可用的内核。

缓存创建如下:

    CacheConfiguration<Integer, MyObject> cfg = new CacheConfiguration<Integer, MyObject>();
    cfg.setSwapEnabled(false);
    cfg.setName(CACHE_NAME);
    cfg.setCacheMode(CacheMode.PARTITIONED);
    cfg.setMemoryMode(CacheMemoryMode.ONHEAP_TIERED);
    cfg.setBackups(0);
    cfg.setCopyOnRead(false);
    this.cache = ignite.getOrCreateCache(CACHE_NAME);

CPU使用率看起来只有一个线程在完成工作。当我将实现切换到ArrayList时-不使用Ignite时，CPU使用率将达到400％。

这段代码用于过滤缓存:

            IgniteCache<Integer, MyObject> cache = ignite.getOrCreateCache(CACHE_NAME);
            Spliterator<Entry<Integer, MyObject>> split = cache.localEntries().spliterator();
            Stream<MyObject> stream = StreamSupport.stream(split, true).map( entry -> entry.getValue());    
            aggregate.setCount(stream.filter(new SomePredicate()).count());

使用Ignite运行时进行了一些性能分析，并发现一次仅存在一个Runnable线程，而使用ArrayList则需要3或4个线程来完成工作。

非常感谢您的帮助，

巴特

Answer 1

localEntries返回内部Ignite数据结构的迭代器，因此，我认为如果没有其他支持，就不可能使用Java流API对其进行拆分。但是您可以使用ScanQuery并行化任务，该代码允许迭代特定分区。获取本地分区号列表，为每个分区创建一个单独的查询，并在单独的线程中执行查询。该代码将如下所示:

// Get list of local partitions.
int[] partitions = ignite.affinity(CACHE_NAME).allPartitions(ignite.cluster().localNode());

// For each partition submit a task to a thread pool.
for (final int partition : partitions) {
    executor.submit(new Runnable() {
        @Override public void run() {
            ScanQuery<Integer, MyObject> query = new ScanQuery<>();

            query.setPartition(partition);
            query.setLocal(true);

            QueryCursor<Cache.Entry<Integer, MyObject>> cursor = cache.query(query);

            for (Cache.Entry<Integer, MyObject> entry : cursor) {
                // ...
            }
        }
    });
}