1.什么是異步IO

，當(dāng)請(qǐng)求外部系統(tǒng)或者耗時(shí)操作，需要異步IO

屏幕快照 2019-03-21 下午3.01.01.png

2.AsyncDataStream

屏幕快照 2019-03-21 下午3.02.27.png

private static class SampleAsyncFunction extends RichAsyncFunction<Integer, String> {
private static final long serialVersionUID = 2098635244857937717L;

private transient ExecutorService executorService;

/**
 * The result of multiplying sleepFactor with a random float is used to pause
 * the working thread in the thread pool, simulating a time consuming async operation.
 */
private final long sleepFactor;

/**
 * The ratio to generate an exception to simulate an async error. For example, the error
 * may be a TimeoutException while visiting HBase.
 */
private final float failRatio;

private final long shutdownWaitTS;

SampleAsyncFunction(long sleepFactor, float failRatio, long shutdownWaitTS) {
    this.sleepFactor = sleepFactor;
    this.failRatio = failRatio;
    this.shutdownWaitTS = shutdownWaitTS;
}

@Override
public void open(Configuration parameters) throws Exception {
    super.open(parameters);

    executorService = Executors.newFixedThreadPool(30);
}

@Override
public void close() throws Exception {
    super.close();
    ExecutorUtils.gracefulShutdown(shutdownWaitTS, TimeUnit.MILLISECONDS, executorService);
}

@Override
public void asyncInvoke(final Integer input, final ResultFuture<String> resultFuture) {
    // 沒(méi)有連接數(shù)據(jù)庫(kù)查詢，所以模擬異步操作
    executorService.submit(() -> {
        // wait for while to simulate async operation here
        long sleep = (long) (ThreadLocalRandom.current().nextFloat() * sleepFactor);
        try {
            Thread.sleep(sleep);

            if (ThreadLocalRandom.current().nextFloat() < failRatio) {
                resultFuture.completeExceptionally(new Exception("wahahahaha..."));
            } else {
                resultFuture.complete(
                        Collections.singletonList("key-" + (input % 10)));
            }
        } catch (InterruptedException e) {
            resultFuture.complete(new ArrayList<>(0));
        }
    });
}
}

3.實(shí)現(xiàn)原理

屏幕快照 2019-03-21 下午3.04.41.png

有序

屏幕快照 2019-03-21 下午3.05.07.png

processing time無(wú)序

屏幕快照 2019-03-21 下午3.05.15.png

屏幕快照 2019-03-21 下午3.05.25.png

4.快照恢復(fù)

屏幕快照 2019-03-21 下午3.07.38.png