1.什么是異步IO
摇邦，當(dāng)請求外部系統(tǒng)或者耗時操作蝌戒，需要異步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) {
// 沒有連接數(shù)據(jù)庫查詢卵史，所以模擬異步操作
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.實現(xiàn)原理
屏幕快照 2019-03-21 下午3.04.41.png
有序
屏幕快照 2019-03-21 下午3.05.07.png
processing time無序
屏幕快照 2019-03-21 下午3.05.15.png
event time無序
屏幕快照 2019-03-21 下午3.05.25.png
4.快照恢復(fù)
屏幕快照 2019-03-21 下午3.07.38.png
