《Netty 高性能网络编程》15.2 限流、超时与降级

package com.plume.linkd.v2.server;

import java.util.concurrent.ConcurrentHashMap;

final class TokenBucket {

    private final long capacity;
    private final double tokensPerNano;
    private double tokens;
    private long lastNanos;

    TokenBucket(long capacity, double permitsPerSecond, long nowNanos) {
        this.capacity = capacity;
        this.tokensPerNano = permitsPerSecond / 1_000_000_000d;
        this.tokens = capacity;
        this.lastNanos = nowNanos;
    }

    synchronized boolean tryAcquire(long nowNanos, long permits) {
        refill(nowNanos);
        if (tokens < permits) {
            return false;
        }
        tokens -= permits;
        return true;
    }

    synchronized long nanosUntil(long nowNanos, long permits) {
        refill(nowNanos);
        double missing = permits - tokens;
        return missing <= 0 ? 0 : (long) Math.ceil(missing / tokensPerNano);
    }

    synchronized long idleNanos(long nowNanos) {
        return Math.max(0, nowNanos - lastNanos);
    }

    private void refill(long nowNanos) {
        long elapsed = Math.max(0, nowNanos - lastNanos);
        lastNanos = nowNanos;
        tokens = Math.min(capacity, tokens + elapsed * tokensPerNano);
    }
}

public final class IpRateLimiter {

    private final ConcurrentHashMap<String, TokenBucket> buckets = new ConcurrentHashMap<>();
    private final int maxEntries;
    private final long capacity;
    private final double permitsPerSecond;

    public IpRateLimiter(int maxEntries, long capacity, double permitsPerSecond) {
        this.maxEntries = maxEntries;
        this.capacity = capacity;
        this.permitsPerSecond = permitsPerSecond;
    }

    public boolean tryAcquire(String ip, long nowNanos) {
        TokenBucket existing = buckets.get(ip);
        if (existing != null) {
            return existing.tryAcquire(nowNanos, 1);
        }
        if (buckets.size() >= maxEntries && !evictIdle(nowNanos)) {
            return false;
        }
        return buckets.computeIfAbsent(ip,
                        key -> new TokenBucket(capacity, permitsPerSecond, nowNanos))
                .tryAcquire(nowNanos, 1);
    }

    public long nanosUntil(String ip, long nowNanos) {
        TokenBucket bucket = buckets.get(ip);
        return bucket == null ? 0 : bucket.nanosUntil(nowNanos, 1);
    }

    private boolean evictIdle(long nowNanos) {
        int before = buckets.size();
        buckets.entrySet().removeIf(entry ->
                entry.getValue().idleNanos(nowNanos) > 300_000_000_000L);
        return buckets.size() < before || buckets.size() < maxEntries;
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
    long now = System.nanoTime();
    if (!global.tryAcquire(now, 1)
            || !perConnection.tryAcquire(now, 1)
            || !perIp.tryAcquire(clientIp, now)) {
        reject(ctx, request, now);
        return;
    }
    ctx.fireChannelRead(request.retain());
}

private void reject(ChannelHandlerContext ctx, FullHttpRequest request, long now) {
    long waitNanos = Math.max(perConnection.nanosUntil(now, 1),
            perIp.nanosUntil(clientIp, now));
    FullHttpResponse response = new DefaultFullHttpResponse(
            HttpVersion.HTTP_1_1, HttpResponseStatus.TOO_MANY_REQUESTS,
            Unpooled.copiedBuffer("rate limited\n", CharsetUtil.UTF_8));
    response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=utf-8");
    response.headers().set(HttpHeaderNames.RETRY_AFTER, Long.toString(
            Math.max(1, (waitNanos + 999_999_999L) / 1_000_000_000L)));
    HttpUtil.setContentLength(response, response.content().readableBytes());
    ChannelFuture write = ctx.writeAndFlush(response);
    if (!HttpUtil.isKeepAlive(request)) {
        write.addListener(ChannelFutureListener.CLOSE);
    }
}

package com.plume.linkd.v2.server;

import java.time.Duration;

public final class CircuitBreaker {

    public enum State { CLOSED, OPEN, HALF_OPEN }

    private final int failureThreshold;
    private final long openNanos;
    private State state = State.CLOSED;
    private int failures;
    private long openedAt;
    private boolean probeInFlight;

    public CircuitBreaker(int failureThreshold, Duration openFor) {
        this.failureThreshold = failureThreshold;
        this.openNanos = openFor.toNanos();
    }

    public synchronized boolean tryAcquire(long nowNanos) {
        if (state == State.OPEN) {
            if (nowNanos - openedAt < openNanos) {
                return false;
            }
            state = State.HALF_OPEN;
        }
        if (state == State.HALF_OPEN) {
            if (probeInFlight) {
                return false;
            }
            probeInFlight = true;
        }
        return true;
    }

    public synchronized void onSuccess() {
        if (state == State.OPEN) {
            return;   // 熔断打开前放行的旧请求，迟到成功不能直接恢复
        }
        state = State.CLOSED;
        failures = 0;
        probeInFlight = false;
    }

    public synchronized void onFailure(long nowNanos) {
        if (state == State.OPEN) {
            return;   // 已经打开，旧请求的失败不重复延长打开时间
        }
        probeInFlight = false;
        if (state == State.HALF_OPEN || ++failures >= failureThreshold) {
            state = State.OPEN;
            openedAt = nowNanos;
            failures = 0;
        }
    }

    public synchronized State state() {
        return state;
    }
}

if (!breaker.tryAcquire(System.nanoTime())) {
    write(ctx, HttpResponseStatus.SERVICE_UNAVAILABLE, "circuit open");
    return;
}
// ... 15.1 的 forward ...
result.whenComplete((response, error) -> {
    if (error == null) {
        breaker.onSuccess();
    } else {
        breaker.onFailure(System.nanoTime());
    }
});

HttpServerCodec -> HttpObjectAggregator -> RateLimitHandler
    -> LinkdHttpHandler -> GatewayHandler