聊聊Elasticsearch的LazyInitializable

序

本文主要研究一下 Elasticsearch 的 LazyInitializable

LazyInitializable

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

public final class LazyInitializable<T, E extends Exception> {

    private final CheckedSupplier<T, E> supplier;
    private final Consumer<T> onGet;
    private final Consumer<T> onReset;
    private volatile T value;

    /**
     * Creates the simple LazyInitializable instance.
     *
     * @param supplier
     *            The {@code CheckedSupplier} to generate values which will be
     *            served on {@code #getOrCompute()} invocations.
     */
    public LazyInitializable(CheckedSupplier<T, E> supplier) {this(supplier, v -> {}, v -> {});
    }

    /**
     * Creates the complete LazyInitializable instance.
     *
     * @param supplier
     *            The {@code CheckedSupplier} to generate values which will be
     *            served on {@code #getOrCompute()} invocations.
     * @param onGet
     *            A {@code Consumer} which is called on each value, newly forged or
     *            stale, that is returned by {@code #getOrCompute()}
     * @param onReset
     *            A {@code Consumer} which is invoked on the value that will be
     *            erased when calling {@code #reset()}
     */
    public LazyInitializable(CheckedSupplier<T, E> supplier, Consumer<T> onGet, Consumer<T> onReset) {
        this.supplier = supplier;
        this.onGet = onGet;
        this.onReset = onReset;
    }

    /**
     * Returns a value that was created by <code>supplier</code>. The value might
     * have been previously created, if not it will be created now, thread safe of
     * course.
     */
    public T getOrCompute() throws E {
        final T readOnce = value; // Read volatile just once...
        final T result = readOnce == null ? maybeCompute(supplier) : readOnce;
        onGet.accept(result);
        return result;
    }

    /**
     * Clears the value, if it has been previously created by calling
     * {@code #getOrCompute()}. The <code>onReset</code> will be called on this
     * value. The next call to {@code #getOrCompute()} will recreate the value.
     */
    public synchronized void reset() {if (value != null) {onReset.accept(value);
            value = null;
        }
    }

    /**
     * Creates a new value thread safely.
     */
    private synchronized T maybeCompute(CheckedSupplier<T, E> supplier) throws E {if (value == null) {value = Objects.requireNonNull(supplier.get());
        }
        return value;
    }

}

• LazyInitializable 封装了 CheckedSupplier，类似 CachedSupplier，不过它提供了 reset 方法可以重置以反复使用，另外还支持了 onGet、onReset 的回调

实例

elasticsearch-7.0.1/plugins/discovery-ec2/src/main/java/org/elasticsearch/discovery/ec2/AwsEc2ServiceImpl.java

class AwsEc2ServiceImpl implements AwsEc2Service {private static final Logger logger = LogManager.getLogger(AwsEc2ServiceImpl.class);

    private final AtomicReference<LazyInitializable<AmazonEc2Reference, ElasticsearchException>> lazyClientReference =
            new AtomicReference<>();

    //......
    
    public void refreshAndClearCache(Ec2ClientSettings clientSettings) {
        final LazyInitializable<AmazonEc2Reference, ElasticsearchException> newClient = new LazyInitializable<>(() -> new AmazonEc2Reference(buildClient(clientSettings)), clientReference -> clientReference.incRef(),
                clientReference -> clientReference.decRef());
        final LazyInitializable<AmazonEc2Reference, ElasticsearchException> oldClient = this.lazyClientReference.getAndSet(newClient);
        if (oldClient != null) {oldClient.reset();
        }
    }

    //......
}        

• AwsEc2ServiceImpl 的 refreshAndClearCache 方法会根据 clientSettings 创建新的 LazyInitializable，之后更新 lazyClientReference，如果 oldClient 不为 null 则调用其 reset 方法重置 value

小结

LazyInitializable 封装了 CheckedSupplier，类似 CachedSupplier，不过它提供了 reset 方法可以重置以反复使用，另外还支持了 onGet、onReset 的回调

doc

• LazyInitializable