Java令牌桶限流器实现异常求助：线程同步问题排查

核心问题分析 #

你的令牌桶限流器出现线程阻塞和功能异常，主要源于以下几个线程同步和对象设计的错误：

1. 错误的锁对象选择 #

你使用currentCapacity（Integer类型）作为同步锁，但Integer是不可变类：每次执行--currentCapacity或++currentCapacity时，都会生成一个新的Integer对象，导致锁对象频繁变更。

• 当currentCapacity回到缓存范围内的数值（比如示例中的5，Integer默认缓存-128到127的数值），会复用初始的Integer对象。此时run线程一直持有该对象的锁，主线程调用consume()时会被永久阻塞，无法继续获取令牌。

修复：使用固定的专属锁对象，比如在类中声明：

private final Object lock = new Object();

所有同步操作都基于这个lock对象，避免锁对象变更。

2. 同步块范围过大，永久占用锁 #

run方法的同步块包裹了整个while(true)死循环：一旦run线程获取到锁，就会在循环中一直持有锁不释放，其他线程无法获取锁执行操作。

• 这直接导致主线程在第一次获取令牌后，后续的consume()调用无法获取锁（当锁对象复用后），程序陷入假死状态。

修复：缩小同步块范围，仅在修改currentCapacity时加锁，其余逻辑（比如时间检查）不需要同步：

@Override
public void run() {
    OffsetDateTime t = OffsetDateTime.now();
    while(true) {
        if(OffsetDateTime.now().isAfter(t.plus(frequencyInMilliseconds, ChronoUnit.MILLIS))){
            synchronized (lock) {
                if(currentCapacity < capacity){
                    System.out.println("token added");
                    ++currentCapacity;
                }
            }
            t = OffsetDateTime.now();
            // 添加休眠避免忙等待
            try {
                Thread.sleep(frequencyInMilliseconds);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }
}

3. 无休眠的忙等待，占用CPU资源 #

run方法中通过循环不断检查时间，没有任何休眠操作，会持续占用CPU资源，同时也会延长锁的持有时间（如果同步块范围没调整）。

修复：在每次添加令牌后，休眠对应的间隔时间（frequencyInMilliseconds），代替忙等待，既节省CPU，也能保证令牌添加的频率准确。

4. 线程可见性问题 #

成员变量currentCapacity没有被声明为volatile，虽然同步块能保证内存可见性，但锁对象变更后，可能导致线程无法及时读取到变量的最新值。

修复：将currentCapacity声明为volatile，或者确保所有对它的读写都在同一个锁的同步块内，保证线程间的可见性。

修复后的核心代码示例 #

public class RateLimiterTokenBucket implements RateLimiter, Runnable{
    private final Integer capacity;
    private final Integer frequencyInMilliseconds;
    private volatile Integer currentCapacity;
    private final Object lock = new Object();

    public RateLimiterTokenBucket(Integer frequencyInMilliseconds,Integer capacity) {
        this.frequencyInMilliseconds = frequencyInMilliseconds;
        this.capacity = capacity;
        this.currentCapacity = capacity;
    }

    @Override
    public boolean consume() {
        synchronized (lock) {
            if(currentCapacity <= 0) return false;
            --currentCapacity;
            return true;
        }
    }

    @Override
    public void run() {
        while(!Thread.currentThread().isInterrupted()) {
            try {
                Thread.sleep(frequencyInMilliseconds);
                synchronized (lock) {
                    if(currentCapacity < capacity){
                        System.out.println("token added");
                        ++currentCapacity;
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

内容的提问来源于stack exchange，提问作者Sourabh