Анализ исходного кода семафора серии Dead Java Synchronization

вопрос

(1) Что такое семафор?

(2) Какими характеристиками обладает семафор?

(3) В каких сценариях обычно используется семафор?

(4) Можно ли динамически увеличивать или уменьшать количество лицензий на Semaphore?

(5) Как Semaphore реализует ограничение тока?

Введение

Semaphore, семафор, содержит ряд разрешений (разрешений), каждый вызов для приобретения() будет потреблять разрешение, и каждый вызов для выпуска() будет возвращать разрешение.

характеристика

Семафор обычно используется для ограничения количества одновременных обращений к общим ресурсам, что часто называют текущим ограничением.

Давайте узнаем, как семафор реализован в Java.

структура класса

Semaphore включает синхронизатор Sync, который реализует AQS, и два его подкласса FairSync и NonFairSync, которые показывают, что Semaphore также различает честные и несправедливые режимы.

Анализ исходного кода

Основываясь на предыдущем анализе ReentrantLock и ReentrantReadWriteLock, эта статья относительно проста. Некоторые из методов, упомянутых ранее, будут пропущены напрямую. Если вам интересно, вы можете перейти к нижней части статьи, чтобы просмотреть предыдущие статьи.

Синхронизация внутреннего класса

// java.util.concurrent.Semaphore.Sync
abstract static class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 1192457210091910933L;
    // 构造方法，传入许可次数，放入state中
    Sync(int permits) {
        setState(permits);
    }
    // 获取许可次数
    final int getPermits() {
        return getState();
    }
    // 非公平模式尝试获取许可
    final int nonfairTryAcquireShared(int acquires) {
        for (;;) {
            // 看看还有几个许可
            int available = getState();
            // 减去这次需要获取的许可还剩下几个许可
            int remaining = available - acquires;
            // 如果剩余许可小于0了则直接返回
            // 如果剩余许可不小于0，则尝试原子更新state的值，成功了返回剩余许可
            if (remaining < 0 ||
                compareAndSetState(available, remaining))
                return remaining;
        }
    }
    // 释放许可
    protected final boolean tryReleaseShared(int releases) {
        for (;;) {
            // 看看还有几个许可
            int current = getState();
            // 加上这次释放的许可
            int next = current + releases;
            // 检测溢出
            if (next < current) // overflow
                throw new Error("Maximum permit count exceeded");
            // 如果原子更新state的值成功，就说明释放许可成功，则返回true
            if (compareAndSetState(current, next))
                return true;
        }
    }
    // 减少许可
    final void reducePermits(int reductions) {
        for (;;) {
            // 看看还有几个许可
            int current = getState();
            // 减去将要减少的许可
            int next = current - reductions;
            // 检测举出
            if (next > current) // underflow
                throw new Error("Permit count underflow");
            // 原子更新state的值，成功了返回true
            if (compareAndSetState(current, next))
                return;
        }
    }
    // 销毁许可
    final int drainPermits() {
        for (;;) {
            // 看看还有几个许可
            int current = getState();
            // 如果为0，直接返回
            // 如果不为0，把state原子更新为0
            if (current == 0 || compareAndSetState(current, 0))
                return current;
        }
    }
}

С помощью нескольких методов реализации Sync мы получаем следующую информацию:

(1) Разрешение передается при создании метода;

(2) Лицензия хранится в состоянии переменной состояния;

(3) При попытке получить лицензию значение состояния уменьшается на 1;

(4) Когда значение состояния равно 0, лицензию больше нельзя получить;

(5) При выпуске лицензии значение состояния увеличивается на 1;

(6) количество лицензий может изменяться динамически;

Внутренний класс NonfairSync

// java.util.concurrent.Semaphore.NonfairSync
static final class NonfairSync extends Sync {
    private static final long serialVersionUID = -2694183684443567898L;
    // 构造方法，调用父类的构造方法
    NonfairSync(int permits) {
        super(permits);
    }
    // 尝试获取许可，调用父类的nonfairTryAcquireShared()方法
    protected int tryAcquireShared(int acquires) {
        return nonfairTryAcquireShared(acquires);
    }
}

В нечестном режиме вызовите nonfairTryAcquireShared() родительского класса напрямую, чтобы попытаться получить разрешение.

Внутренний класс FairSync

// java.util.concurrent.Semaphore.FairSync
static final class FairSync extends Sync {
    private static final long serialVersionUID = 2014338818796000944L;
    // 构造方法，调用父类的构造方法
    FairSync(int permits) {
        super(permits);
    }
    // 尝试获取许可
    protected int tryAcquireShared(int acquires) {
        for (;;) {
            // 公平模式需要检测是否前面有排队的
            // 如果有排队的直接返回失败
            if (hasQueuedPredecessors())
                return -1;
            // 没有排队的再尝试更新state的值
            int available = getState();
            int remaining = available - acquires;
            if (remaining < 0 ||
                compareAndSetState(available, remaining))
                return remaining;
        }
    }
}

В честном режиме сначала проверьте есть ли впереди очередь.Если очередь есть, то не получает лицензию и попадает в очередь на очередь.В противном случае попробуйте атомарно обновить значение состояния.

Метод строительства

// 构造方法，创建时要传入许可次数，默认使用非公平模式
public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}
// 构造方法，需要传入许可次数，及是否公平模式
public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

Разрешения должны быть переданы при создании семафора.

Семафор также находится в нечестном режиме по умолчанию, но вы можете объявить его в справедливом режиме, вызвав второй конструктор.

Следующие методы кажутся относительно простыми после изучения предыдущего содержания.Брат Тонг перечисляет только некоторые функции, поддерживаемые семафором.

Следующие методы описаны для нечестного режима.

метод приобретения()

public void acquire() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}

Для получения лицензии по умолчанию используется прерываемый метод.Если попытка получить лицензию не удалась, она будет поставлена в очередь в очереди AQS.

методAcquireUninterruptably()

public void acquireUninterruptibly() {
    sync.acquireShared(1);
}

Получите лицензию без перерыва. Если попытка получить лицензию не удалась, она будет помещена в очередь AQS.

Метод tryAcquire()

public boolean tryAcquire() {
    return sync.nonfairTryAcquireShared(1) >= 0;
}

Попытаться получить лицензию, использовать недобросовестный режим Sync, чтобы попытаться получить лицензию, и вернуться независимо от того, получена ли лицензия, попробовать только один раз и не попасть в очередь.

Метод tryAcquire (длительный тайм-аут, модуль TimeUnit)

public boolean tryAcquire(long timeout, TimeUnit unit)
    throws InterruptedException {
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

Попробуйте получить лицензию, сначала попробуйте получить лицензию один раз, если это не удастся, он будет ждать тайм-аута.Если лицензия не будет получена в течение этого времени, он вернет false, в противном случае он вернет true;

метод выпуска()

public void release() {
    sync.releaseShared(1);
}

Когда лицензия освобождается, значение состояния увеличивается на 1, когда лицензия освобождается, и следующий поток, ожидающий лицензии, активируется.

метод получения (разрешения)

public void acquire(int permits) throws InterruptedException {
    if (permits < 0) throw new IllegalArgumentException();
    sync.acquireSharedInterruptibly(permits);
}

Приобретайте сразу несколько лицензий, прерываемый способ.

методAcquireUninterruptably(int разрешает)

public void acquireUninterruptibly(int permits) {
    if (permits < 0) throw new IllegalArgumentException();
    sync.acquireShared(permits);
}

Приобретайте несколько лицензий одновременно, без прерывания работы.

Метод tryAcquire(int разрешает)

public boolean tryAcquire(int permits) {
    if (permits < 0) throw new IllegalArgumentException();
    return sync.nonfairTryAcquireShared(permits) >= 0;
}

Попытайтесь получить несколько лицензий одновременно, попробуйте только один раз.

Метод tryAcquire (разрешения int, долгое время ожидания, блок TimeUnit)

public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
    throws InterruptedException {
    if (permits < 0) throw new IllegalArgumentException();
    return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
}

Попытаться получить несколько лицензий и дождаться тайм-аута. Если в течение этого времени лицензия не будет получена, будет возвращено значение false, в противном случае будет возвращено значение true.

метод освобождения (разрешения)

public void release(int permits) {
    if (permits < 0) throw new IllegalArgumentException();
    sync.releaseShared(permits);
}

Когда несколько лицензий выпускаются одновременно, значение состояния соответственно увеличивает количество разрешений.

Метод доступных разрешений()

public int availablePermits() {
    return sync.getPermits();
}

Получите количество доступных лицензий.

Метод слива разрешений()

public int drainPermits() {
    return sync.drainPermits();
}

Уничтожение количества доступных в настоящее время лицензий не повлияет на уже полученные лицензии и уничтожит все оставшиеся лицензии.

Метод reducePermits(int сокращение)

protected void reducePermits(int reduction) {
    if (reduction < 0) throw new IllegalArgumentException();
    sync.reducePermits(reduction);
}

Уменьшите количество лицензий.

Суммировать

(1) Семафор, также известный как семафор, обычно используется для одновременного управления доступом к общим ресурсам, то есть текущим ограничивающим сценариям;

(2) Внутренняя реализация Semaphore основана на общей блокировке AQS;

(3) При инициализации семафора необходимо указать количество лицензий, и количество лицензий сохраняется в состоянии;

(4) При получении лицензии государственная стоимость уменьшается на 1;

(5) Когда лицензия отпускается, значение состояния увеличивается на 1;

(6) n лицензий можно динамически сокращать;

(7) Можно ли динамически добавлять n лицензий?

пасхальные яйца

(1) Как динамически увеличивать количество лицензий?

Ответ: Вы можете вызвать релиз (разрешения int). Мы знаем, что значение состояния соответственно увеличится при выпуске лицензии. Оглядываясь назад на исходный код выпуска лицензии, мы обнаруживаем, что он немного отличается от блокировки выпуска ReentrantLock. Когда Semaphore выпускает лицензию, он не проверьте, получил ли текущий поток лицензию или нет.Таким образом, вы можете вызвать метод выпуска разрешения для динамического добавления некоторых разрешений.

(2) Как реализовать ограничение тока?

Ответ: Текущее ограничение, то есть, когда трафик внезапно увеличивается, верхний уровень должен быть в состоянии ограничить влияние внезапного большого трафика на нижестоящие службы.В распределенных системах текущее ограничение обычно выполняется на уровне шлюза, конечно, это также можно использовать для отдельных функций.Чтобы ограничить поток просто, например, в сценарии seckill, если только 10 элементов должны быть seckilled, то сам сервис может ограничить входящие запросы только 100 запросами одновременно, и все другие запросы будут признаны недействительными, так что нагрузка на службу не будет слишком большой.

Используя Semaphore, вы можете напрямую ограничить ток для этой функции.Ниже приведена реализация кода:

public class SemaphoreTest {
    public static final Semaphore SEMAPHORE = new Semaphore(100);
    public static final AtomicInteger failCount = new AtomicInteger(0);
    public static final AtomicInteger successCount = new AtomicInteger(0);

    public static void main(String[] args) {
        for (int i = 0; i < 1000; i++) {
            new Thread(()->seckill()).start();
        }
    }

    public static boolean seckill() {
        if (!SEMAPHORE.tryAcquire()) {
            System.out.println("no permits, count="+failCount.incrementAndGet());
            return false;
        }

        try {
            // 处理业务逻辑
            Thread.sleep(2000);
            System.out.println("seckill success, count="+successCount.incrementAndGet());
        } catch (InterruptedException e) {
            // todo 处理异常
            e.printStackTrace();
        } finally {
            SEMAPHORE.release();
        }
        return true;
    }
}