资讯专栏INFORMATION COLUMN

聊聊Elasticsearch的CircuitBreaker

番茄西红柿 / 3293人阅读

摘要:序本文主要研究一下的定义了枚举它还定义了等方法它有两个实现类分别是实现了接口,它不做任何操作实现了接口其方法会抛出方法首先判断,如果为,则执行方法如果为则调用,否则调用计算,没有抛出异常的话,则最后执

本文主要研究一下Elasticsearch的CircuitBreaker

CircuitBreaker

elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/common/breaker/CircuitBreaker.java

/** * Interface for an object that can be incremented, breaking after some * configured limit has been reached. */ public interface CircuitBreaker { /** * The parent breaker is a sum of all the following breakers combined. With * this we allow a single breaker to have a significant amount of memory * available while still having a "total" limit for all breakers. Note that * its not a "real" breaker in that it cannot be added to or subtracted * from by itself. */ String PARENT = "parent"; /** * The fielddata breaker tracks data used for fielddata (on fields) as well * as the id cached used for parent/child queries. */ String FIELDDATA = "fielddata"; /** * The request breaker tracks memory used for particular requests. This * includes allocations for things like the cardinality aggregation, and * accounting for the number of buckets used in an aggregation request. * Generally the amounts added to this breaker are released after a request * is finished. */ String REQUEST = "request"; /** * The in-flight request breaker tracks bytes allocated for reading and * writing requests on the network layer. */ String IN_FLIGHT_REQUESTS = "in_flight_requests"; /** * The accounting breaker tracks things held in memory that is independent * of the request lifecycle. This includes memory used by Lucene for * segments. */ String ACCOUNTING = "accounting"; enum Type { // A regular or ChildMemoryCircuitBreaker MEMORY, // A special parent-type for the hierarchy breaker service PARENT, // A breaker where every action is a noop, it never breaks NOOP; public static Type parseValue(String value) { switch(value.toLowerCase(Locale.ROOT)) { case "noop": return Type.NOOP; case "parent": return Type.PARENT; case "memory": return Type.MEMORY; default: throw new IllegalArgumentException("No CircuitBreaker with type: " + value); } } } enum Durability { // The condition that tripped the circuit breaker fixes itself eventually. TRANSIENT, // The condition that tripped the circuit breaker requires manual intervention. PERMANENT } /** * Trip the circuit breaker * @param fieldName name of the field responsible for tripping the breaker * @param bytesNeeded bytes asked for but unable to be allocated */ void circuitBreak(String fieldName, long bytesNeeded); /** * add bytes to the breaker and maybe trip * @param bytes number of bytes to add * @param label string label describing the bytes being added * @return the number of "used" bytes for the circuit breaker */ double addEstimateBytesAndMaybeBreak(long bytes, String label) throws CircuitBreakingException; /** * Adjust the circuit breaker without tripping */ long addWithoutBreaking(long bytes); /** * @return the currently used bytes the breaker is tracking */ long getUsed(); /** * @return maximum number of bytes the circuit breaker can track before tripping */ long getLimit(); /** * @return overhead of circuit breaker */ double getOverhead(); /** * @return the number of times the circuit breaker has been tripped */ long getTrippedCount(); /** * @return the name of the breaker */ String getName(); /** * @return whether a tripped circuit breaker will reset itself (transient) or requires manual intervention (permanent). */ Durability getDurability(); }

CircuitBreaker定义了Type、Durability枚举;它还定义了circuitBreak、addEstimateBytesAndMaybeBreak、addWithoutBreaking、getUsed、getLimit、getOverhead、getTrippedCount等方法;它有两个实现类分别是NoopCircuitBreaker、ChildMemoryCircuitBreaker

NoopCircuitBreaker

elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/common/breaker/NoopCircuitBreaker.java

public class NoopCircuitBreaker implements CircuitBreaker { public static final int LIMIT = -1; private final String name; public NoopCircuitBreaker(String name) { this.name = name; } @Override public void circuitBreak(String fieldName, long bytesNeeded) { // noop } @Override public double addEstimateBytesAndMaybeBreak(long bytes, String label) throws CircuitBreakingException { return 0; } @Override public long addWithoutBreaking(long bytes) { return 0; } @Override public long getUsed() { return 0; } @Override public long getLimit() { return LIMIT; } @Override public double getOverhead() { return 0; } @Override public long getTrippedCount() { return 0; } @Override public String getName() { return this.name; } @Override public Durability getDurability() { return Durability.PERMANENT; } }

NoopCircuitBreaker实现了CircuitBreaker接口,它不做任何操作

ChildMemoryCircuitBreaker

elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/common/breaker/ChildMemoryCircuitBreaker.java

public class ChildMemoryCircuitBreaker implements CircuitBreaker { private final long memoryBytesLimit; private final double overheadConstant; private final Durability durability; private final AtomicLong used; private final AtomicLong trippedCount; private final Logger logger; private final HierarchyCircuitBreakerService parent; private final String name; /** * Create a circuit breaker that will break if the number of estimated * bytes grows above the limit. All estimations will be multiplied by * the given overheadConstant. This breaker starts with 0 bytes used. * @param settings settings to configure this breaker * @param parent parent circuit breaker service to delegate tripped breakers to * @param name the name of the breaker */ public ChildMemoryCircuitBreaker(BreakerSettings settings, Logger logger, HierarchyCircuitBreakerService parent, String name) { this(settings, null, logger, parent, name); } /** * Create a circuit breaker that will break if the number of estimated * bytes grows above the limit. All estimations will be multiplied by * the given overheadConstant. Uses the given oldBreaker to initialize * the starting offset. * @param settings settings to configure this breaker * @param parent parent circuit breaker service to delegate tripped breakers to * @param name the name of the breaker * @param oldBreaker the previous circuit breaker to inherit the used value from (starting offset) */ public ChildMemoryCircuitBreaker(BreakerSettings settings, ChildMemoryCircuitBreaker oldBreaker, Logger logger, HierarchyCircuitBreakerService parent, String name) { this.name = name; this.memoryBytesLimit = settings.getLimit(); this.overheadConstant = settings.getOverhead(); this.durability = settings.getDurability(); if (oldBreaker == null) { this.used = new AtomicLong(0); this.trippedCount = new AtomicLong(0); } else { this.used = oldBreaker.used; this.trippedCount = oldBreaker.trippedCount; } this.logger = logger; if (logger.isTraceEnabled()) { logger.trace("creating ChildCircuitBreaker with settings {}", settings); } this.parent = parent; } /** * Method used to trip the breaker, delegates to the parent to determine * whether to trip the breaker or not */ @Override public void circuitBreak(String fieldName, long bytesNeeded) { this.trippedCount.incrementAndGet(); final String message = "[" + this.name + "] Data too large, data for [" + fieldName + "]" + " would be [" + bytesNeeded + "/" + new ByteSizeValue(bytesNeeded) + "]" + ", which is larger than the limit of [" + memoryBytesLimit + "/" + new ByteSizeValue(memoryBytesLimit) + "]"; logger.debug("{}", message); throw new CircuitBreakingException(message, bytesNeeded, memoryBytesLimit, durability); } /** * Add a number of bytes, tripping the circuit breaker if the aggregated * estimates are above the limit. Automatically trips the breaker if the * memory limit is set to 0. Will never trip the breaker if the limit is * set < 0, but can still be used to aggregate estimations. * @param bytes number of bytes to add to the breaker * @return number of "used" bytes so far */ @Override public double addEstimateBytesAndMaybeBreak(long bytes, String label) throws CircuitBreakingException { // short-circuit on no data allowed, immediately throwing an exception if (memoryBytesLimit == 0) { circuitBreak(label, bytes); } long newUsed; // If there is no limit (-1), we can optimize a bit by using // .addAndGet() instead of looping (because we dont have to check a // limit), which makes the RamAccountingTermsEnum case faster. if (this.memoryBytesLimit == -1) { newUsed = noLimit(bytes, label); } else { newUsed = limit(bytes, label); } // Additionally, we need to check that we havent exceeded the parents limit try { parent.checkParentLimit((long) (bytes * overheadConstant), label); } catch (CircuitBreakingException e) { // If the parent breaker is tripped, this breaker has to be // adjusted back down because the allocation is "blocked" but the // breaker has already been incremented this.addWithoutBreaking(-bytes); throw e; } return newUsed; } private long noLimit(long bytes, String label) { long newUsed; newUsed = this.used.addAndGet(bytes); if (logger.isTraceEnabled()) { logger.trace("[{}] Adding [{}][{}] to used bytes [new used: [{}], limit: [-1b]]", this.name, new ByteSizeValue(bytes), label, new ByteSizeValue(newUsed)); } return newUsed; } private long limit(long bytes, String label) { long newUsed;// Otherwise, check the addition and commit the addition, looping if // there are conflicts. May result in additional logging, but its // trace logging and shouldnt be counted on for additions. long currentUsed; do { currentUsed = this.used.get(); newUsed = currentUsed + bytes; long newUsedWithOverhead = (long) (newUsed * overheadConstant); if (logger.isTraceEnabled()) { logger.trace("[{}] Adding [{}][{}] to used bytes [new used: [{}], limit: {} [{}], estimate: {} [{}]]", this.name, new ByteSizeValue(bytes), label, new ByteSizeValue(newUsed), memoryBytesLimit, new ByteSizeValue(memoryBytesLimit), newUsedWithOverhead, new ByteSizeValue(newUsedWithOverhead)); } if (memoryBytesLimit > 0 && newUsedWithOverhead > memoryBytesLimit) { logger.warn("[{}] New used memory {} [{}] for data of [{}] would be larger than configured breaker: {} [{}], breaking", this.name, newUsedWithOverhead, new ByteSizeValue(newUsedWithOverhead), label, memoryBytesLimit, new ByteSizeValue(memoryBytesLimit)); circuitBreak(label, newUsedWithOverhead); } // Attempt to set the new used value, but make sure it hasnt changed // underneath us, if it has, keep trying until we are able to set it } while (!this.used.compareAndSet(currentUsed, newUsed)); return newUsed; } /** * Add an exact number of bytes, not checking for tripping the * circuit breaker. This bypasses the overheadConstant multiplication. * * Also does not check with the parent breaker to see if the parent limit * has been exceeded. * * @param bytes number of bytes to add to the breaker * @return number of "used" bytes so far */ @Override public long addWithoutBreaking(long bytes) { long u = used.addAndGet(bytes); if (logger.isTraceEnabled()) { logger.trace("[{}] Adjusted breaker by [{}] bytes, now [{}]", this.name, bytes, u); } assert u >= 0 : "Used bytes: [" + u + "] must be >= 0"; return u; } /** * @return the number of aggregated "used" bytes so far */ @Override public long getUsed() { return this.used.get(); } /** * @return the number of bytes that can be added before the breaker trips */ @Override public long getLimit() { return this.memoryBytesLimit; } /** * @return the constant multiplier the breaker uses for aggregations */ @Override public double getOverhead() { return this.overheadConstant; } /** * @return the number of times the breaker has been tripped */ @Override public long getTrippedCount() { return this.trippedCount.get(); } /** * @return the name of the breaker */ @Override public String getName() { return this.name; } /** * @return whether a tripped circuit breaker will reset itself (transient) or requires manual intervention (permanent). */ @Override public Durability getDurability() { return this.durability; } }

ChildMemoryCircuitBreaker实现了CircuitBreaker接口;其circuitBreak方法会抛出CircuitBreakingException

addEstimateBytesAndMaybeBreak方法首先判断memoryBytesLimit,如果为0,则执行circuitBreak方法;如果为-1则调用noLimit,否则调用limit计算newUsed,没有抛出异常的话,则最后执行 parent.checkParentLimit方法

noLimit方法直接执行this.used.addAndGet(bytes);limit方法首先计算newUsed,然后根据overheadConstant得出newUsedWithOverhead,如果newUsedWithOverhead大于memoryBytesLimit则执行circuitBreak方法,否则将newUsed更新到this.used中

小结

CircuitBreaker定义了Type、Durability枚举;它还定义了circuitBreak、addEstimateBytesAndMaybeBreak、addWithoutBreaking、getUsed、getLimit、getOverhead、getTrippedCount等方法;它有两个实现类分别是NoopCircuitBreaker、ChildMemoryCircuitBreaker

NoopCircuitBreaker实现了CircuitBreaker接口,它不做任何操作

ChildMemoryCircuitBreaker实现了CircuitBreaker接口;其circuitBreak方法会抛出CircuitBreakingException;addEstimateBytesAndMaybeBreak方法则先判断newUsed是否超出memoryBytesLimit,超出则执行circuitBreak方法,最后执行parent.checkParentLimit方法

doc

CircuitBreaker

文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。

转载请注明本文地址:https://www.ucloud.cn/yun/7145.html

相关文章

  • 聊聊ElasticsearchCircuitBreakerService

    摘要:序本文主要研究一下的继承了,定义了抽象方法它有两个实现类分别为包含了属性包含了属性不做熔断处理的构造器读取了,并使用它们之后对执行,添加一系列的方法会创建并更新到中小结继承了,定义了抽象方法它有两个实现类分别为不做熔断处理的 序 本文主要研究一下Elasticsearch的CircuitBreakerService CircuitBreakerService elasticsearch...

    SillyMonkey 评论0 收藏0
  • 聊聊ElasticsearchCircuitBreaker

    摘要:序本文主要研究一下的定义了枚举它还定义了等方法它有两个实现类分别是实现了接口,它不做任何操作实现了接口其方法会抛出方法首先判断,如果为,则执行方法如果为则调用,否则调用计算,没有抛出异常的话,则最后执序 本文主要研究一下Elasticsearch的CircuitBreaker CircuitBreaker elasticsearch-7.0.1/server/src/main/java/or...

    番茄西红柿 评论0 收藏0
  • 聊聊ElasticsearchCircuitBreaker

    摘要:序本文主要研究一下的定义了枚举它还定义了等方法它有两个实现类分别是实现了接口,它不做任何操作实现了接口其方法会抛出方法首先判断,如果为,则执行方法如果为则调用,否则调用计算,没有抛出异常的话,则最后执序 本文主要研究一下Elasticsearch的CircuitBreaker CircuitBreaker elasticsearch-7.0.1/server/src/main/java/or...

    yangrd 评论0 收藏0
  • 聊聊ElasticsearchCircuitBreaker

    摘要:序本文主要研究一下的定义了枚举它还定义了等方法它有两个实现类分别是实现了接口,它不做任何操作实现了接口其方法会抛出方法首先判断,如果为,则执行方法如果为则调用,否则调用计算,没有抛出异常的话,则最后执行方法方法直接执 序 本文主要研究一下Elasticsearch的CircuitBreaker CircuitBreaker elasticsearch-7.0.1/server/src/...

    raledong 评论0 收藏0
  • Spring Cloud Gateway断路器(CircuitBreaker)功能

    摘要:欢迎访问我的内容所有原创文章分类汇总及配套源码,涉及等本篇概览一起深入了解的断路器功能先聊聊理论再结合官方和大神的信息确定技术栈再动手开发,先实现再验证再趁热打铁,看看它的源码最后,回顾一下有哪些不足下一篇文章解决这些不足关于断路器下图来自欢迎访问我的GitHub https://github.com/zq2599/blog_demos 内容:所有原创文章分类汇总及配套源码,涉及...

    microcosm1994 评论0 收藏0

发表评论

0条评论

最新活动
阅读需要支付1元查看
<