线程隔离
- 线程池隔离：给每个服务调用业务分配一个线程池，利用线程池本身实现隔离效果。
- 信号量隔离：不创建线程池，而是计数器模式，记录业务使用的线程数量，达到信号量上限时，禁止新的请求。
这两种隔离方式都是用于保护系统在高负载情况下不受影响的方法。线程池隔离和信号量隔离都是常见的微服务架构中的隔离策略。
线程池隔离是通过为每个服务调用分配一个专门的线程池来实现隔离。这意味着每个服务都有自己的一组线程可供使用，这些线程不会被其他服务所共享。这种方式可以确保某个服务的高负载不会影响到其他服务，因为每个服务都有自己的线程资源。这种方式的优势在于能够很好地控制每个服务的资源使用情况，但缺点是需要消耗更多的系统资源来维护多个线程池。
信号量隔离则是一种更轻量级的隔离方式，它通过限制并发请求的数量来保护系统免受高负载的影响。当达到信号量上限时，系统会拒绝新的请求，直到之前的请求完成并释放出资源。这种方式的优势在于相对较少的资源消耗，但可能会导致请求被拒绝，需要合理设置信号量上限以平衡系统的负载和可用性。
在选择隔离方式时，需要根据具体的业务需求和系统架构来进行权衡。通常情况下，线程池隔离适用于需要更精细控制资源分配的场景，而信号量隔离适用于需要简单有效地限制并发请求数量的场景。
基于这两种写一下java实现方法
当涉及Java实现时，你可以使用Java的并发工具来实现线程池隔离和信号量隔离。以下是使用Java的Executor框架和Semaphore类的简单示例：
线程池隔离的Java实现：
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class ThreadPoolIsolationExample {
public static void main(String[] args) {
// 创建一个固定大小的线程池，每个服务调用分配一个线程池
ExecutorService service1ThreadPool = Executors.newFixedThreadPool(10);
ExecutorService service2ThreadPool = Executors.newFixedThreadPool(10);// 向线程池提交任务
service1ThreadPool.submit(() -> {
// 执行服务1的业务逻辑
});service2ThreadPool.submit(() -> {
// 执行服务2的业务逻辑
});// 关闭线程池
service1ThreadPool.shutdown();
service2ThreadPool.shutdown();
}
}
信号量隔离的Java实现：
import java.util.concurrent.Semaphore;
public class SemaphoreIsolationExample {
public static void main(String[] args) {
// 创建信号量，限制并发请求的数量
Semaphore semaphore = new Semaphore(10); // 假设限制为10个并发请求// 在处理请求之前尝试获取信号量
try {
semaphore.acquire(); // 获取信号量，如果达到上限，会阻塞直到有信号量可用
// 执行业务逻辑
} catch (InterruptedException e) {
// 处理中断异常
} finally {
semaphore.release(); // 释放信号量
}
}
}
以上是简单的示例代码，实际场景中需要根据具体需求进行更加完善的实现，并考虑异常处理、线程池的生命周期管理、资源的合理分配等问题。
滑动窗口算法
在熔断功能中，需要统计异常请求或慢请求比例，也就是计数。在限流的时候，要统计每秒钟的QPS，同样是计数。可见计数算法在熔断限流中的应用非常多。sentinel中采用的计数器算法就是滑动窗口计数算法。
移动窗口算法通常用于监控一段时间内事件的发生频率或者比例，比如成功请求和失败请求的比例、异常事件的频率等。在移动窗口算法中，我们使用一个固定大小的窗口来追踪一段时间内的事件情况，通过移动窗口来计算事件的比例或者频率。
对于请求监控来说，我们可以使用移动窗口算法来监控一段时间内的请求失败率。每次有新的请求到达时，我们会根据请求的结果（成功或失败）和时间戳来更新窗口中的数据，然后根据窗口中的数据来计算失败请求的比例。这个比例可以帮助我们判断当前系统的稳定性，以及是否需要采取一些措施比如熔断来保护系统免受过多失败请求的影响。
在实际的应用中，移动窗口算法可以帮助我们快速地捕捉到系统的异常情况，并根据异常情况来做出相应的调整和处理。这种算法在微服务架构中的容错机制中经常被使用，比如用于熔断、限流等场景。
import java.util.ArrayDeque;
import java.util.Queue;public class Request {
private long timestamp;
private boolean isFailed;public Request(long timestamp, boolean isFailed) {
this.timestamp = timestamp;
this.isFailed = isFailed;
}public long getTimestamp() {
return timestamp;
}public boolean isFailed() {
return isFailed;
}
}public class FailureRateMonitor {
private final long interval;
private final int n;
private final long timeInterval;
private final Queue<Request> requests;public FailureRateMonitor(long interval, int n) {
this.interval = interval;
this.n = n;
this.timeInterval = interval / n;
this.requests = new ArrayDeque<>();
}public void addRequest(boolean isFailed, long currentTime) {
evictOldRequests(currentTime);
requests.add(new Request(currentTime, isFailed));
}public double getFailureRate(long currentTime) {
evictOldRequests(currentTime);
if (requests.isEmpty()) {
return 0.0;
} else {
long failureCount = requests.stream().filter(r -> r.isFailed()).count();
return (double) failureCount / requests.size();
}
}private void evictOldRequests(long currentTime) {
long startTime = currentTime - interval;
while (!requests.isEmpty() && requests.peek().getTimestamp() < startTime) {
requests.poll();
}
}public static void main(String[] args) {
FailureRateMonitor monitor = new FailureRateMonitor(1000, 2);
long currentTime = System.currentTimeMillis();
for (int i = 0; i < 10; i++) {
boolean isFailed = i % 3 == 0;
monitor.addRequest(isFailed, currentTime + i * 300);
System.out.println("Current failure rate: " + monitor.getFailureRate(currentTime + i * 300));
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
漏桶算法
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @author YY-帆S
* @Date 2024/3/26 22:46
*/
@Slf4j
public class LeakyBucketRateLimiter {
private AtomicInteger bucketLevel; // 当前桶中的请求数量
private int capacity; // 桶的容量
private long leakRate; // 漏水速率，单位：请求/秒
private long lastLeakTime; // 上一次漏水的时间戳
public LeakyBucketRateLimiter(int capacity, long leakRate) {
this.capacity = capacity;
this.leakRate = leakRate;
this.bucketLevel = new AtomicInteger(0);
this.lastLeakTime = System.currentTimeMillis();
}
public synchronized boolean tryAcquire() {
// 获取当前时间
long currentTime = System.currentTimeMillis();
//流出时间
long elapsedTime = currentTime - lastLeakTime;
//计算流出的水量 = （当前时间 - 上次时间） * 出水速率
long leaked = (long) (elapsedTime * (leakRate / 1000.0));
//只有有流出水才更新时间戳，不然会漏不出水
if (leaked > 0) {
//计算桶内水量 = 桶内当前水量 - 流出的水量
int newLevel = Math.max(0, bucketLevel.get() - (int) leaked);
bucketLevel.set(newLevel);
//更新上次漏水时间戳
lastLeakTime = currentTime;
}
// 尝试将请求加入桶中
if (bucketLevel.get() < capacity) {
bucketLevel.incrementAndGet();
return true;
} else {
return false;
}
}
public static void main(String[] args) throws InterruptedException {
LeakyBucketRateLimiter limiter = new LeakyBucketRateLimiter(1, 1); // 容量为1，漏水速率为1请求/秒
// 模拟发送请求
for (int i = 0; i < 20; i++) {
new Thread(() -> {
if (limiter.tryAcquire()) {
log.info(Thread.currentThread().getName() + " 获得了许可，执行操作。");
} else {
log.info(Thread.currentThread().getName() + " 请求被拒绝。");
}
}).start();
//模拟执行时间
Thread.sleep(500);
}
}
}
