предисловие
Пакет concurrent JUC предоставляет множество классов инструментов, таких как упомянутые ранее CountDownLatch и CyclicBarrier.Сегодня я расскажу об этом Semaphore — семафоре.Пожалуйста, ознакомьтесь с предыдущими статьями по его использованию.Класс инструмента совместной работы потоков для параллельного программирования, Сегодняшняя задача - разобрать его принцип с уровня исходного кода.
Анализ исходного кода
Если мы сначала не посмотрим на исходный код, основываясь на опыте с исходным кодом CountDownLatch CyclicBarrier, который мы видели в прошлом, как будет спроектирован Semaphore?
Во-первых, он хочет реализовать несколько потоков, одновременно обращающихся к ресурсу, подобно разделяемой блокировке, и контролировать количество потоков, входящих в ресурс.
Если в соответствии с существующими ресурсами JDK, можем ли мы использовать переменную состояния AQS для управления им? Подобно CountDownLatch, есть несколько потоков, для которых мы устанавливаем переменную состояния в несколько.Когда состояние достигает порога, другие потоки не могут получить блокировку и должны ждать. Когда Semaphore вызывает метод освобождения, он снимает блокировку, уменьшает состояние на единицу и пробуждает поток в AQS.
Вышесказанное является нашим предположением, а затем посмотрим, совпадает ли JDK с тем, что мы думали.
Сначала взгляните на структуру UML семафора:
Внутри есть 3 класса, наследующие AQS. Честная блокировка, нечестная блокировка, точно такая же, как ReentrantLock.
Взгляните на его конструктор:
public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
Два конструктора, два параметра, один — количество разрешенных потоков, другой — справедливая ли блокировка, по умолчанию — нечестная.
И у Semaphore есть 2 важных метода, которые мы часто используем:
semaphore.acquire();
// doSomeing.....
semaphore.release();
Методы получения и освобождения. Сегодня мы сосредоточимся на исходном коде этих двух методов и получим представление о Semaphore в целом.
Анализ исходного кода метода получения
код показывает, как показано ниже:
public void acquire() throws InterruptedException {
// 尝试获取一个锁
sync.acquireSharedInterruptibly(1);
}
// 这是抽象类 AQS 的方法
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
// 如果小于0，就获取锁失败了。加入到AQS 等待队列中。
// 如果大于0，就直接执行下面的逻辑了。不用进行阻塞等待。
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
// 这是抽象父类 Sync 的方法，默认是非公平的
protected int tryAcquireShared(int acquires) {
return nonfairTryAcquireShared(acquires);
}
// 非公平锁的释放锁的方法
final int nonfairTryAcquireShared(int acquires) {
// 死循环
for (;;) {
// 获取锁的状态
int available = getState();
int remaining = available - acquires;
// state 变量是否还足够当前获取的
// 如果小于 0，获取锁就失败了。
// 如果大于 0，就循环尝试使用 CAS 将 state 变量更新成减去输入参数之后的。
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
Релиз здесь заключается в уменьшении переменной состояния на единицу (или более).
Возвращает оставшийся размер состояния.
Когда возвращаемое значение меньше 0, это означает, что получение блокировки не удалось, и необходимо войти в очередь ожидания AQS. Введите код ниже:
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
// 添加一个节点 AQS 队列尾部
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
// 死循环
for (;;) {
// 找到新节点的上一个节点
final Node p = node.predecessor();
// 如果这个节点是 head，就尝试获取锁
if (p == head) {
// 继续尝试获取锁，这个方法是子类实现的
int r = tryAcquireShared(arg);
// 如果大于0，说明拿到锁了。
if (r >= 0) {
// 将 node 设置为 head 节点
// 如果大于0，就说明还有机会获取锁，那就唤醒后面的线程，称之为传播
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
// 如果他的上一个节点不是 head，就不能获取锁
// 对节点进行检查和更新状态，如果线程应该阻塞，返回 true。
if (shouldParkAfterFailedAcquire(p, node) &&
// 阻塞 park，并返回是否中断，中断则抛出异常
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
// 取消节点
cancelAcquire(node);
}
}
Общая логика такова:
-
Создайте узел типа общего доступа, который упаковывает текущий поток и добавляет его в конец очереди AQS.
-
Если предыдущим узлом этого узла является head , он пытается получить блокировку, и метод получения блокировки является методом, переопределяемым подклассом. Если сбор данных прошел успешно, установите узел прямо сейчас как головной.
-
Если замок не схвачен, он заблокируется и будет ждать.
Анализ исходного кода метода выпуска
Этот метод используется для снятия блокировки, код выглядит следующим образом:
public void release() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
// 死循环释放成功
if (tryReleaseShared(arg)) {
// 唤醒 AQS 等待对列中的节点，从 head 开始
doReleaseShared();
return true;
}
return false;
}
// Sync extends AbstractQueuedSynchronizer
protected final boolean tryReleaseShared(int releases) {
for (;;) {
int current = getState();
// 对 state 变量 + 1
int next = current + releases;
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
if (compareAndSetState(current, next))
return true;
}
}
Логика снятия блокировки здесь написана в абстрактном классе Sync. Логика проста, просто добавьте переменную состояния.
После успешного добавления выполнитеdoReleaseSharedметод, этот метод AQS.
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
// 设置 head 的等待状态为 0 ，并唤醒 head 上的线程
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
// 成功设置成 0 之后，将 head 状态设置成传播状态
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
Основная функция этого метода — разбудить поток с головного узла AQS.Обратите внимание, что пробуждение здесь — это следующий узел головного узла, который необходимоdoAcquireSharedInterruptiblyметод соответствует, потому чтоdoAcquireSharedInterruptiblyПробужденный методом предыдущий узел текущего узла, то есть головной узел.
В этот момент отпустите переменную состояния и разбудите головной узел AQS.
Суммировать
Кратко опишите принцип работы семафора.
В общем, семафор — это общая блокировка, которая используется путем установки переменной состояния. Когда вызывается метод получения, переменная состояния вычитается на единицу, а когда вызывается метод освобождения, переменная состояния увеличивается на единицу. Когда переменная состояния равна 0, другие потоки не могут войти в блок кода и будут блокироваться и ожидать в AQS.