Skip to content

backoff support #24

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
drtrang wants to merge 2 commits into
base: master
Choose a base branch
from
Open

backoff support #24

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
aa523e1
[feature] add backoff
drtrang Nov 9, 2019
3daf328
[failover] support backoff
drtrang Nov 9, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,8 @@
<version>0.1.19-SNAPSHOT</version>

<properties>
<more-lambdas.version>0.1.12</more-lambdas.version>
<slf4j-api.version>1.7.14</slf4j-api.version>
<more-lambdas.version>0.1.34</more-lambdas.version>
<slf4j-api.version>1.7.15</slf4j-api.version>
<guava.version>21.0</guava.version>
<jsr305.version>3.0.2</jsr305.version>

Expand Down
25 changes: 23 additions & 2 deletions src/main/java/com/github/phantomthief/failover/impl/GenericWeightFailoverBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,9 @@
import javax.annotation.CheckReturnValue;
import javax.annotation.Nonnegative;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

import com.github.phantomthief.failover.backoff.BackOff;
import com.github.phantomthief.util.ThrowableFunction;
import com.github.phantomthief.util.ThrowablePredicate;

Expand Down Expand Up @@ -96,12 +98,21 @@ public GenericWeightFailoverBuilder<E> checkDuration(long time, TimeUnit unit) {

@CheckReturnValue
@Nonnull
public GenericWeightFailoverBuilder<E>
checker(@Nonnull ThrowableFunction<? super E, Double, Throwable> failChecker) {
public GenericWeightFailoverBuilder<E> checker(
@Nonnull ThrowableFunction<? super E, Double, Throwable> failChecker) {
builder.checker(failChecker);
return this;
}

@CheckReturnValue
@Nonnull
public GenericWeightFailoverBuilder<E> checker(
@Nonnull ThrowableFunction<? super E, Double, Throwable> failChecker,
@Nullable BackOff backOff) {
builder.checker(failChecker, backOff);
return this;
}

@CheckReturnValue
@Nonnull
public GenericWeightFailoverBuilder<E> checker(
Expand All @@ -111,6 +122,16 @@ public GenericWeightFailoverBuilder<E> checker(
return this;
}

@CheckReturnValue
@Nonnull
public GenericWeightFailoverBuilder<E> checker(
@Nonnull ThrowablePredicate<? super E, Throwable> failChecker,
@Nonnegative double recoveredInitRate,
@Nullable BackOff backOff) {
builder.checker(failChecker, recoveredInitRate, backOff);
return this;
}

@CheckReturnValue
@Nonnull
public GenericWeightFailoverBuilder<E> filter(Predicate<E> filter) {
Expand Down
196 changes: 158 additions & 38 deletions src/main/java/com/github/phantomthief/failover/impl/WeightFailover.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,13 @@
package com.github.phantomthief.failover.impl;

import static com.github.phantomthief.tuple.Tuple.tuple;
import static com.github.phantomthief.util.MoreFunctions.catching;
import static com.github.phantomthief.util.MoreFunctions.runCatching;
import static com.github.phantomthief.util.MoreFunctions.runWithThreadName;
import static com.github.phantomthief.util.MoreFunctions.supplyWithThreadName;
import static com.github.phantomthief.util.MoreSuppliers.lazy;
import static com.google.common.primitives.Ints.constrainToRange;
import static com.google.common.util.concurrent.Futures.addCallback;
import static java.lang.Integer.MAX_VALUE;
import static java.lang.Math.max;
import static java.lang.Math.min;
Expand All @@ -21,20 +26,30 @@
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.Consumer;
import java.util.function.IntUnaryOperator;
import java.util.function.Predicate;
import java.util.function.ToDoubleFunction;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;

import org.slf4j.Logger;

import com.github.phantomthief.failover.Failover;
import com.github.phantomthief.failover.backoff.BackOff;
import com.github.phantomthief.failover.backoff.BackOffExecution;
import com.github.phantomthief.failover.util.SharedCheckExecutorHolder;
import com.github.phantomthief.tuple.TwoTuple;
import com.github.phantomthief.util.MoreSuppliers.CloseableSupplier;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.ListenableFuture;

/**
* 默认权重记录
Expand All @@ -47,14 +62,19 @@ public class WeightFailover<T> implements Failover<T>, Closeable {

private static final Logger logger = getLogger(WeightFailover.class);

private final String name;

private final IntUnaryOperator failReduceWeight;
private final IntUnaryOperator successIncreaseWeight;

private final ConcurrentMap<T, Integer> initWeightMap;
private final ConcurrentMap<T, Integer> currentWeightMap;
private final CloseableSupplier<ScheduledFuture<?>> recoveryFuture;
private final ConcurrentLinkedQueue<RecoveryTask> recoveryTasks;
private final ConcurrentMap<T, RecoveryTask> runningRecoveryTasks;
private final Consumer<T> onMinWeight;
private final int minWeight;
private final ToDoubleFunction<T> checker;

/**
* {@code null} if this feature is off.
Expand All @@ -69,6 +89,7 @@ public class WeightFailover<T> implements Failover<T>, Closeable {
private volatile boolean closed;

WeightFailover(WeightFailoverBuilder<T> builder) {
this.name = builder.name;
this.minWeight = builder.minWeight;
this.failReduceWeight = builder.failReduceWeight;
this.successIncreaseWeight = builder.successIncreaseWeight;
Expand All @@ -77,48 +98,110 @@ public class WeightFailover<T> implements Failover<T>, Closeable {
this.onMinWeight = builder.onMinWeight;
this.weightOnMissingNode = builder.weightOnMissingNode;
this.filter = builder.filter;
this.recoveryFuture = lazy(
() -> SharedCheckExecutorHolder.getInstance().scheduleWithFixedDelay(() -> {
if (closed) {
tryCloseRecoveryScheduler();
return;
this.checker = builder.checker;
this.recoveryTasks = new ConcurrentLinkedQueue<>();
this.runningRecoveryTasks = new ConcurrentHashMap<>();
this.recoveryFuture = builder.backOff == null ? initDefaultChecker(builder) : initBackOffChecker(builder);

@PhantomThief PhantomThief Nov 11, 2019

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

defaultChecker是不是也可以用backoff实现？

@drtrang drtrang Nov 11, 2019

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

可以的，FixedBackOff 可以实现与当前一致的效果，但是考虑到 backoff 的实现还没有经过验证，所以没有直接替换 defaultChecker。
我建议先在几个客户端中手动替换掉（比如 HttpCall），稳定之后在下个版本再替换

}

private CloseableSupplier<ScheduledFuture<?>> initDefaultChecker(WeightFailoverBuilder<T> builder) {
return lazy(() -> SharedCheckExecutorHolder.getInstance().scheduleWithFixedDelay(() -> {
if (closed) {
tryCloseRecoveryScheduler();
return;
}
Thread currentThread = Thread.currentThread();
String origName = currentThread.getName();
if (builder.name != null) {
currentThread.setName(origName + "-[" + builder.name + "]");
}
try {
Map<T, Double> recoveredObjects = new HashMap<>();
this.currentWeightMap.forEach((obj, weight) -> {
if (weight == 0) {
double recoverRate = builder.checker.applyAsDouble(obj);
if (recoverRate > 0) {
recoveredObjects.put(obj, recoverRate);
}
}
});
if (!recoveredObjects.isEmpty()) {
logger.info("found recovered objects:{}", recoveredObjects);
}
recoveredObjects.forEach((recovered, rate) -> {
Integer initWeight = initWeightMap.get(recovered);
if (initWeight == null) {
throw new IllegalStateException("obj:" + recovered);
}
Thread currentThread = Thread.currentThread();
String origName = currentThread.getName();
if (builder.name != null) {
currentThread.setName(origName + "-[" + builder.name + "]");
int recoveredWeight = constrainToRange((int) (initWeight * rate), 1,
initWeight);
currentWeightMap.put(recovered, recoveredWeight);
if (builder.onRecovered != null) {
builder.onRecovered.accept(recovered);
}
try {
Map<T, Double> recoveredObjects = new HashMap<>();
this.currentWeightMap.forEach((obj, weight) -> {
if (weight == 0) {
double recoverRate = builder.checker.applyAsDouble(obj);
if (recoverRate > 0) {
recoveredObjects.put(obj, recoverRate);
});
} catch (Throwable e) {
logger.error("", e);
} finally {
currentThread.setName(origName);
}
}, builder.checkDuration, builder.checkDuration, MILLISECONDS));
}

private CloseableSupplier<ScheduledFuture<?>> initBackOffChecker(WeightFailoverBuilder<T> builder) {
return lazy(() -> SharedCheckExecutorHolder.getInstance().scheduleWithFixedDelay(() -> {
if (closed) {

@PhantomThief PhantomThief Nov 11, 2019

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

整个runnable需要做try/catch保护，避免没处理的意外导致可用性下降；

tryCloseRecoveryScheduler();
return;
}
// 为权重为 0 的资源创建健康检查任务，放入待检查队列
this.currentWeightMap.entrySet().stream()
.filter(entry -> entry.getValue() == 0)
.forEach((entry -> recoveryTasks.offer(new RecoveryTask(entry.getKey(), builder.backOff))));
// 如果队列不为空则将其转发到 scheduler
if (!recoveryTasks.isEmpty()) {
while (true) {
RecoveryTask task = recoveryTasks.poll();
if (task == null) {
break;
}
// 如果健康检查耗时超过 checkDuration，可能导致对同一资源发起多次 hc 请求，这里规避一下
if (runningRecoveryTasks.containsKey(task.getObj())) {
continue;
}
T obj = task.getObj();
// 新建的 task 将其放入 running 队列中
runningRecoveryTasks.put(obj, task);
// 与外层共用一个线程池，貌似也没什么不妥...

@PhantomThief PhantomThief Nov 11, 2019

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

还有问题，这里调度任务是有可能有业务自定制的堵塞逻辑，scheduled线程池通常不多，所以这里需要设计调度线程池和检查线程池隔离，进一步，执行线程池的大小也需要确认，因为风险很大。

@kaori-seasons kaori-seasons Jul 3, 2021

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

您好 如果是定时调度线程池的话 咱觉着是不是可以参考一些咱正在进行的PR,里面有个FutureUtils会比较合适点
apache/pulsar#10544

@kaori-seasons kaori-seasons Jul 3, 2021 •
edited
Loading

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@PhantomThief 其实pulsar的BrokerService相关设计也很优雅 可以参考看下相关的调度线程池实现

ListenableFuture<Integer> future =
SharedCheckExecutorHolder.getInstance().schedule(task, task.nextBackOff(), MILLISECONDS);

@PhantomThief PhantomThief Nov 11, 2019

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

schedule永远不会上抛异常吧？这里addCallback才能remove runningRecoveryTasks，是否要保护下异常分支？

addCallback(future, new FutureCallback<Integer>() {
@Override
public void onSuccess(@Nullable Integer recoveredWeight) {
runWithThreadName(origin -> origin + "-[" + name + "]", () -> runCatching(() -> {
// 不管成不成功执行完毕先将 task 删除
runningRecoveryTasks.remove(obj);
// 如果检查失败再次将其放入待检查队列，delay 时间顺延到下一周期
if (recoveredWeight == null || recoveredWeight == 0) {
recoveryTasks.offer(task);
} else {
currentWeightMap.put(obj, recoveredWeight);
if (builder.onRecovered != null) {
builder.onRecovered.accept(obj);
}
}
}
});
if (!recoveredObjects.isEmpty()) {
logger.info("found recovered objects:{}", recoveredObjects);
}));
}
recoveredObjects.forEach((recovered, rate) -> {
Integer initWeight = initWeightMap.get(recovered);
if (initWeight == null) {
throw new IllegalStateException("obj:" + recovered);
}
int recoveredWeight = constrainToRange((int) (initWeight * rate), 1,
initWeight);
currentWeightMap.put(recovered, recoveredWeight);
if (builder.onRecovered != null) {
builder.onRecovered.accept(recovered);
}
});
} catch (Throwable e) {
logger.error("", e);
} finally {
currentThread.setName(origName);
}
}, builder.checkDuration, builder.checkDuration, MILLISECONDS));

@Override
public void onFailure(@Nonnull Throwable t) {
// 不管成不成功执行完毕先将 task 删除
runningRecoveryTasks.remove(obj);
}
});
}
}
}, builder.checkDuration, builder.checkDuration, MILLISECONDS));
}

/**
Expand Down Expand Up @@ -319,4 +402,41 @@ public Set<T> getFailed() {
public String toString() {
return "WeightFailover [" + initWeightMap + "]" + "@" + Integer.toHexString(hashCode());
}

class RecoveryTask implements Callable<Integer> {

private final T obj;
private final BackOffExecution backOff;

RecoveryTask(T obj, BackOff backOff) {
this.obj = obj;
this.backOff = backOff.start();
}

@Override
public Integer call() throws Exception {
return supplyWithThreadName(origin -> origin + "-[" + name + "]", () -> catching(() -> {
double recoverRate = checker.applyAsDouble(obj);
if (recoverRate > 0) {
Integer initWeight = initWeightMap.get(obj);
if (initWeight == null) {
throw new IllegalStateException("obj:" + obj);
}
return constrainToRange((int) (initWeight * recoverRate), 1, initWeight);
}
return 0;
}));
}

@Nonnull
T getObj() {
return obj;
}

long nextBackOff() {
return backOff.nextBackOff();
}

}

}
Loading