Напишите блокировку самостоятельно в серии «Синхронизация Java»

вопрос

(1) Какие знания вам нужны, чтобы написать блокировку самостоятельно?

(2) Насколько легко написать блокировку самостоятельно?

(3) Можете ли вы сами написать идеальный замок?

Введение

Цель этой статьи — написать замок самостоятельно, функция этого замка очень проста, и он может выполнять обычные операции блокировки и разблокировки.

Вторая цель этой статьи — лучше понять принципы AQS и различных реализаций синхронизаторов, которые будут изучены в следующих главах путем самостоятельного написания блокировки.

анализировать

Что нужно подготовить, чтобы написать блокировку самостоятельно?

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

Однако мы не можем сами изменять информацию заголовка объекта, поэтому можем ли мы вместо этого использовать переменную?

Например, когда значение этой переменной равно 1, это означает, что она заблокирована, а когда значение переменной равно 0, это означает, что она не заблокирована, я думаю, что это возможно.

Во-вторых, нам нужно убедиться, что конкуренция нескольких потоков за переменную, которую мы определили выше, управляема.Так называемая управляемость означает, что только один поток может изменить свое значение на 1 в одно и то же время, и когда его значение равно 1, другие Поток больше не может изменять свое значение.Это типичная операция CAS, поэтому нам нужно использовать класс Unsafe для выполнения операции CAS.

Затем мы знаем, что в многопоточной среде только один из нескольких потоков должен преуспеть в борьбе за одну и ту же блокировку, тогда другие потоки будут поставлены в очередь, поэтому нам также нужна очередь.

Наконец, что делают эти потоки, когда они поставлены в очередь? Они больше не могут продолжать выполнять свои собственные программы, поэтому они могут только блокироваться, и после блокировки они проснутся, когда наступит очередь потока, поэтому нам также нужен класс Unsfae для блокировки (парковки) и пробуждения (разпарковки) потоки.

Основываясь на вышеупомянутых четырех пунктах, нам нужны примерно следующие артефакты: переменная, очередь и класс Unsafe, который выполняет CAS/park/unpark.

Приблизительная блок-схема показана на следующем рисунке:

Соответствующее объяснение класса Unsafe см. в предыдущей статье Тонг Ге:

【Небезопасный анализ мертвого магического класса Java】

решить

Переменная

Эта переменная поддерживает только то, что только один поток может изменить ее на 1 одновременно, поэтому она должна быть видна другим потокам после ее изменения.Поэтому эту переменную необходимо изменить с помощью volatile.

private volatile int state;

CAS

Модификация этой переменной должна быть атомарной операцией, поэтому нам нужен CAS для ее обновления.Мы используем здесь Unsafe для непосредственного обновления состояния типа int.

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

private boolean compareAndSetState(int expect, int update) {
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

очередь

Существует множество реализаций очередей. Допустимы как массивы, так и связанные списки. Здесь мы используем связанные списки. Ведь реализовать очереди со связанными списками относительно просто, и нет необходимости рассматривать такие вопросы, как расширение емкости.

Работа этой очереди очень характерна:

При размещении элементов они размещаются в хвосте и могут размещаться несколькими потоками вместе, поэтому работа хвоста должна обновляться CAS;

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

private static class Node {
    // 存储的元素为线程
    Thread thread;
    // 前一个节点（可以没有，但实现起来很困难）
    Node prev;
    // 后一个节点
    Node next;

    public Node() {
    }

    public Node(Thread thread, Node prev) {
        this.thread = thread;
        this.prev = prev;
    }
}
// 链表头
private volatile Node head;
// 链表尾
private volatile Node tail;
// 原子更新tail字段
private boolean compareAndSetTail(Node expect, Node update) {
    return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}

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

замок

public void lock() {
    // 尝试更新state字段，更新成功说明占有了锁
    if (compareAndSetState(0, 1)) {
        return;
    }
    // 未更新成功则入队
    Node node = enqueue();
    Node prev = node.prev;
    // 再次尝试获取锁，需要检测上一个节点是不是head，按入队顺序加锁
    while (node.prev != head || !compareAndSetState(0, 1)) {
        // 未获取到锁，阻塞
        unsafe.park(false, 0L);
    }
    // 下面不需要原子更新，因为同时只有一个线程访问到这里
    // 获取到锁了且上一个节点是head
    // head后移一位
    head = node;
    // 清空当前节点的内容，协助GC
    node.thread = null;
    // 将上一个节点从链表中剔除，协助GC
    node.prev = null;
    prev.next = null;
}
// 入队
private Node enqueue() {
    while (true) {
        // 获取尾节点
        Node t = tail;
        // 构造新节点
        Node node = new Node(Thread.currentThread(), t);
        // 不断尝试原子更新尾节点
        if (compareAndSetTail(t, node)) {
            // 更新尾节点成功了，让原尾节点的next指针指向当前节点
            t.next = node;
            return node;
        }
    }
}

(1) Попробуйте получить замок и вернитесь напрямую в случае успеха;

(2) Если блокировка не получена, она ставится в очередь на очередь;

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

(4) В случае неудачи заблокировать;

(5) В случае успеха переместите головной узел на одно место назад, очистите содержимое текущего узла и разорвите связь с предыдущим узлом;

(6) Блокировка окончена;

разблокировать

// 解锁
public void unlock() {
    // 把state更新成0，这里不需要原子更新，因为同时只有一个线程访问到这里
    state = 0;
    // 下一个待唤醒的节点
    Node next = head.next;
    // 下一个节点不为空，就唤醒它
    if (next != null) {
        unsafe.unpark(next.thread);
    }
}

(1) Измените состояние на 0, здесь нет необходимости в обновлении CAS, потому что он все еще заблокирован, обновляется только один поток, и после этого предложения блокировка снимается;

(2) Если есть следующий узел, разбудить его;

(3) После пробуждения он пройдет через цикл while метода lock() выше и попытается получить блокировку;

(4) Пробужденный поток не может на 100% получить блокировку, потому что состояние разблокируется, когда состояние обновляется до 0, и тогда могут быть потоки, которые попытаются заблокировать.

тестовое задание

Реализация приведенного выше полного замка завершена. Он очень простой, но действительно ли он надежный? Осмелитесь попробовать? !

Перейдите непосредственно к тестовому коду:

private static int count = 0;

public static void main(String[] args) throws InterruptedException {
    MyLock lock = new MyLock();

    CountDownLatch countDownLatch = new CountDownLatch(1000);

    IntStream.range(0, 1000).forEach(i -> new Thread(() -> {
        lock.lock();

        try {
            IntStream.range(0, 10000).forEach(j -> {
                count++;
            });
        } finally {
            lock.unlock();
        }
//            System.out.println(Thread.currentThread().getName());
        countDownLatch.countDown();
    }, "tt-" + i).start());

    countDownLatch.await();

    System.out.println(count);
}

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

Суммировать

(1) Чтобы написать блокировку самостоятельно, вам нужно подготовить: переменную, очередь и класс Unsafe.

(2) переменная атомарного обновления равна 1, что указывает на то, что блокировка была успешно получена;

(3) Если атомарная переменная обновления не равна 1, это означает, что блокировку не удалось получить, и она попадает в очередь за очередью;

(4) при обновлении хвостового узла очереди существует многопоточная конкуренция, поэтому используется атомарное обновление;

(5) При обновлении головного узла очереди существует только один поток, и конкуренция отсутствует, поэтому нет необходимости использовать атомарное обновление;

(6)Использование предыдущего узла prev в узле очереди очень умно.Без него трудно реализовать блокировку.Это понимают только те,кто это написал.Если не верите мне,попробуйте^^

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

(1) Блокировка, которую мы реализуем, поддерживает повторный вход?

Ответ: Не реентерабельный, потому что мы каждый раз обновляем состояние только до 1. Если вы хотите поддерживать реентерабельность, то это тоже очень просто: при получении блокировки проверяйте, занята ли блокировка текущим потоком, если да, то прибавляйте 1 к значению state и уменьшайте его на 1 каждый раз при освобождении блокировка.Когда он уменьшается до 0, это означает, что блокировка была освобождена.

(2) Блокировка, которую мы внедрили, справедливая или нечестная?

Ответ: Несправедливые блокировки, потому что мы пытались один раз при получении замков Это не совсем очередь, так что это нечестная блокировка.

(3) Полный исходный код

Обратите внимание на мою официальную учетную запись «Tong Ge Read Source Code» и ответьте «mylock» в фоновом режиме, чтобы получить полный исходный код этой главы.

Примечание: В следующей главе мы начнем анализ легендарного AQS.Эта глава является основой, пожалуйста, поймите это.