时间戳

首页 > 解决方案 > 可以在 StreamReceiver/Visitor 的单独节点上执行 Ignite Streamer.addData 吗？

ignite - 可以在 StreamReceiver/Visitor 的单独节点上执行 Ignite Streamer.addData 吗？

问题描述

是否可以从客户端节点进行流注入并在服务器节点中拦截相同的流以在插入缓存之前处理流？

这样做的原因是客户端节点从外部源接收流，并且需要将相同的流注入到基于 AffinityKey 跨多个服务器节点的分区缓存中。需要在每个节点上截取流并以最低延迟进行处理。我本可以使用缓存事件来做到这一点，但 StreamVisitor 应该更快。

以下是我要执行的示例。启动 2 个节点：一个包含流媒体，另一个包含 streamReciever：

公共类 StreamerNode {
public static void main(String[] args) { ...... Ignition.setClientMode(false); 点燃 ignite = Ignition.start(igniteConfiguration);

    CacheConfiguration<SeqKey, String> myCfg = new CacheConfiguration<SeqKey, String>("myCache");
    ......
    IgniteCache<SeqKey, String> myCache = ignite.getOrCreateCache(myCfg);
    IgniteDataStreamer<SeqKey, String> myStreamer = ignite.dataStreamer(myCache.getName()); // Create Ignite Streamer for windowing data

    for (int i = 51; i <= 100; i++) {
        String paddedString = org.apache.commons.lang.StringUtils.leftPad(i+"", 7, "0") ;
        String word = "TEST_" + paddedString;
        SeqKey seqKey = new SeqKey("TEST", counter++ );
        myStreamer.addData(seqKey, word) ;
    }
}

}

public class VisitorNode {
public static void main(String[] args) { ...... Ignition.setClientMode(false); 点燃 ignite = Ignition.start(igniteConfiguration);

    CacheConfiguration<SeqKey, String> myCfg = new CacheConfiguration<SeqKey, String>("myCache");
    ......
    IgniteCache<SeqKey, String> myCache = ignite.getOrCreateCache(myCfg);
    IgniteDataStreamer<SeqKey, String> myStreamer = ignite.dataStreamer(myCache.getName()); // Create Ignite Streamer for windowing data

    myStreamer.receiver(new StreamVisitor<SeqKey, String>() {
        int i=1 ;
        @Override
        public void apply(IgniteCache<SeqKey, String> cache, Map.Entry<SeqKey, String> e) {
            String tradeGetData = e.getValue();
            System.out.println(nodeID+" : visitorNode ..count="+ i++ + " received key="+e.getKey() + " : val="+ e.getValue());
            //do some processing here before inserting in the cache .. 
            cache.put(e.getKey(), tradeGetData);
        }
    });
}

}

标签： ignite

解决方案

当然它可以在不同的节点上执行。通常，addData()在客户端节点上执行，并StreamReceiver在服务器节点上工作。你不需要做任何特别的事情来实现它。

至于你的帖子的其余部分，你能否用更多的细节和样本来详细说明它？我无法理解所需的设置。

如果您不需要修改数据，则可以使用连续查询，只需对其进行操作。

推荐阅读