apache ignite datastreamer 如何将数据设置为 ignitefuture？答案

【问题标题】：apache ignite datastreamer how to set data into ignitefuture?apache ignite datastreamer 如何将数据设置为 ignitefuture？
【发布时间】：2018-03-29 19:40:26
【问题描述】：

我正在 apache ignite 中创建一个批处理数据流，并且需要控制数据接收后发生的情况。我的批次有一个结构：

public class Batch implements Binarylizable, Serializable {

    private String eventKey;
    private byte[] bytes;

    etc..

然后我尝试流式传输我的数据：

 try (IgniteDataStreamer<Integer, Batch> streamer = serviceGrid.getIgnite().dataStreamer(cacheName);
             StreamBatcher batcher = StreamBatcherFactory.create(event) ){
            streamer.receiver(StreamTransformer.from(new BatchDataProcessor(event)));
            streamer.autoFlushFrequency(1000);
            streamer.allowOverwrite(true);
            statusService.updateStatus(event.getKey(), StatusType.EXECUTING);
            int counter = 0;
            Batch batch = null;
            IgniteFuture<?> future = null;
            while ((batch = batcher.batch()) != null) {
                future = streamer.addData(counter++, batch);
            }
            Object getted = future.get();

仅供测试使用，让我们只获取最后一个未来，并尝试分析这个对象。在上面的代码中，我使用的是 BatchDataProcessor，看起来像这样：

public class BatchDataProcessor implements CacheEntryProcessor<Integer, Batch, Object> {

    private final Event event;
    private final String eventKey;

    public BatchDataProcessor(Event event) {
        this.event = event;
        this.eventKey = event.getKey();
    }

    @Override
    public Object process(MutableEntry<Integer, Batch> mutableEntry, Object... objects) throws EntryProcessorException {
        Node node = NodeIgniter.node(Ignition.localIgnite().cluster().localNode().id());
        ServiceGridContainer container = (ServiceGridContainer) node.getEnvironmentContainer().getContainerObject(ServiceGridContainer.class);
        ProcessMarshaller marshaller = (ProcessMarshaller) container.getService(ProcessMarshaller.class);
        LocalProcess localProcess = marshaller.intoProccessing(event.getLambdaExecutionKey());
        try {
            localProcess.addBatch(mutableEntry);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            return new String("111");
        }
    }
}

所以在 localProcess.addBatch(mutableEntry) 之后我想发回有关此特定批次状态的信息，所以我认为我应该在 IgniteFuture 对象中执行此操作，但我没有找到任何信息如何控制在 addData 函数中接收到的 future 对象。

任何人都可以帮助理解，我在哪里可以控制在 addData 函数或其他方式中接收的未来以实现对流批处理的回调？

【问题讨论】：

标签： future ignite data-stream

【解决方案1】：

当您执行StreamTransformer.from() 时，您将失去BatchDataProcessor 的结果，因为

for (Map.Entry<K, V> entry : entries)
    cache.invoke(entry.getKey(), this, entry.getValue());
//  ^ result of cache.invoke() is discarded here

DataStreamer 用于单向数据流。据我所知，它不应该返回值。

如果你依赖cache.invoke()的结果，我建议直接调用而不是依赖DataStreamer。

顺便说一句，小心fut.get()。你应该先做dataStreamer.flush()，否则DataStreamer的未来会无限期地等待。

【讨论】：