Hazelcast射流内部优化

Question

BatchStage<Trade> trades = p.drawFrom(list("trades"));
BatchStage<Entry<Integer, Broker>> brokers =    
p.drawFrom(list("brokers"));
BatchStage<Tuple2<Trade, Broker>> joined = trades.hashJoin(brokers,
    joinMapEntries(Trade::brokerId),
    Tuple2::tuple2);
joined.drainTo(Sinks.logger());

private Pipeline createPipeLine() {
    Pipeline p = Pipeline.create();
    BatchStage stage = p.drawFrom(Sources.<Date>list("master"));
    stage.drainTo(Sinks.logger());
    return p;
}

提前致谢

Answer 1

>

来自Hashjoin的Javadoc：

在实现上，对哈希连接转换进行了吞吐量优化，以便每个计算成员都有存储在哈希表（因此而得名）中的所有丰富数据的本地副本。在从主流中摄取任何数据之前，充实流被全部消耗。

对于您的示例，所有成员将首先使用broker列表中的所有项，然后使用trades列表。

StreamSource<Trade> queueSource = SourceBuilder.<IQueue<Trade>>stream("queueStream", 
        c -> c.jetInstance().getHazelcastInstance().getQueue("trades"))
    .<Trade>fillBufferFn((queue, buf) -> buf.add(queue.poll()))
    .build();