序

本文主要研究一下Elasticsearch的RunOnce

RunOnce

elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/common/util/concurrent/RunOnce.java

public class RunOnce implements Runnable {private final Runnable delegate;private final AtomicBoolean hasRun;public RunOnce(final Runnable delegate) {this.delegate = Objects.requireNonNull(delegate);this.hasRun = new AtomicBoolean(false);}@Overridepublic void run() {if (hasRun.compareAndSet(false, true)) {delegate.run();}}/*** {@code true} if the {@link RunOnce} has been executed once.*/public boolean hasRun() {return hasRun.get();}
}

• RunOnce实现了Runnable接口，它的构造器要求输入Runnable，同时构造了hasRun变量；run方法会先使用compareAndSet将hasRun由false设置为true，如果成功则执行代理的Runnable的run方法；hasRun方法则返回hasRun值

实例

elasticsearch-7.0.1/server/src/test/java/org/elasticsearch/common/util/concurrent/RunOnceTests.java

public class RunOnceTests extends ESTestCase {public void testRunOnce() {final AtomicInteger counter = new AtomicInteger(0);final RunOnce runOnce = new RunOnce(counter::incrementAndGet);assertFalse(runOnce.hasRun());runOnce.run();assertTrue(runOnce.hasRun());assertEquals(1, counter.get());runOnce.run();assertTrue(runOnce.hasRun());assertEquals(1, counter.get());}public void testRunOnceConcurrently() throws InterruptedException {final AtomicInteger counter = new AtomicInteger(0);final RunOnce runOnce = new RunOnce(counter::incrementAndGet);final Thread[] threads = new Thread[between(3, 10)];final CountDownLatch latch = new CountDownLatch(1);for (int i = 0; i < threads.length; i++) {threads[i] = new Thread(() -> {try {latch.await();} catch (InterruptedException e) {throw new RuntimeException(e);}runOnce.run();});threads[i].start();}latch.countDown();for (Thread thread : threads) {thread.join();}assertTrue(runOnce.hasRun());assertEquals(1, counter.get());}public void testRunOnceWithAbstractRunnable() {final AtomicInteger onRun = new AtomicInteger(0);final AtomicInteger onFailure = new AtomicInteger(0);final AtomicInteger onAfter = new AtomicInteger(0);final RunOnce runOnce = new RunOnce(new AbstractRunnable() {@Overrideprotected void doRun() throws Exception {onRun.incrementAndGet();throw new RuntimeException("failure");}@Overridepublic void onFailure(Exception e) {onFailure.incrementAndGet();}@Overridepublic void onAfter() {onAfter.incrementAndGet();}});final int iterations = randomIntBetween(1, 10);for (int i = 0; i < iterations; i++) {runOnce.run();assertEquals(1, onRun.get());assertEquals(1, onFailure.get());assertEquals(1, onAfter.get());assertTrue(runOnce.hasRun());}}
}

• testRunOnce方法验证了顺序多次执行runOnce的场景；testRunOnceConcurrently方法则验证了并发多次执行runOnce的场景；testRunOnceWithAbstractRunnable则验证了使用AbstractRunnable作为runnable的场景

小结

RunOnce实现了Runnable接口，它的构造器要求输入Runnable，同时构造了hasRun变量；run方法会先使用compareAndSet将hasRun由false设置为true，如果成功则执行代理的Runnable的run方法；hasRun方法则返回hasRun值

doc

• RunOnce