import java.util.concurrent.RunnableFuture; //導(dǎo)入方法依賴的package包/類

@Test

public void testSnapshotAsyncCancel() throws Exception {

DefaultOperatorStateBackend operatorStateBackend =

new DefaultOperatorStateBackend(OperatorStateBackendTest.class.getClassLoader(), new ExecutionConfig(), true);

ListStateDescriptor stateDescriptor1 =

new ListStateDescriptor<>("test1", new JavaSerializer());

ListState listState1 = operatorStateBackend.getOperatorState(stateDescriptor1);

listState1.add(MutableType.of(42));

listState1.add(MutableType.of(4711));

BlockerCheckpointStreamFactory streamFactory = new BlockerCheckpointStreamFactory(1024 * 1024);

OneShotLatch waiterLatch = new OneShotLatch();

OneShotLatch blockerLatch = new OneShotLatch();

streamFactory.setWaiterLatch(waiterLatch);

streamFactory.setBlockerLatch(blockerLatch);

RunnableFuture runnableFuture =

operatorStateBackend.snapshot(1, 1, streamFactory, CheckpointOptions.forCheckpoint());

ExecutorService executorService = Executors.newFixedThreadPool(1);

executorService.submit(runnableFuture);

// wait until the async checkpoint is in the stream's write code, then continue

waiterLatch.await();

// cancel the future, which should close the underlying stream

runnableFuture.cancel(true);

Assert.assertTrue(streamFactory.getLastCreatedStream().isClosed());

// we allow the stream under test to proceed

blockerLatch.trigger();

try {

runnableFuture.get(60, TimeUnit.SECONDS);

Assert.fail();

} catch (CancellationException ignore) {

}

}

總結(jié)

以上是生活随笔為你收集整理的java cancel_Java RunnableFuture.cancel方法代码示例的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。