Это первая часть того, почему технологии74оригинальная статья

Не спрашивай, просто не спрашивай

Вы должны знать о распределенных транзакциях. Но эта многопоточная транзакция...

Все в порядке, я буду говорить вам медленно.

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

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

Так есть ли решение?

До этого я всегда отвечал утвердительно: без сомнения, нет.

Зачем?

Начнем с теоретических рассуждений.

Давай, позвольте мне сначала спросить вас, каковы характеристики транзакции?

Разве это не сложно? Одно из содержаний восьминогого текста должно быть произнесено, КИСЛОТА должна прийти в рот:

• атомарность
• Последовательность
• Изоляция
• Долговечность

Итак, снова вопрос: как вы думаете, какую функцию мы нарушаем, если есть многопоточные транзакции?

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

Разве это не многопоточная транзакция?

В этом сценарии вы не подумали о том, как управлять транзакционными операциями двух пользователей по отдельности, верно?

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

Итак, каковы самые основные принципы между несколькими транзакциями?

изоляция. Две транзакционные операции не должны мешать друг другу.

Многопоточная транзакция хочет добиться того, чтобы поток A был ненормальным. Транзакции потоков A и B откатываются вместе.

Характеристики транзакции застряли внутри. Поэтому многопоточные транзакции теоретически не работают.

Теория и практика не позволяют написать код многопоточной транзакции.

Я уже говорил об изоляции. Так пожалуйста,Как в исходном коде Spring гарантируется изоляция транзакций?

Ответ — ThreadLocal.

Когда транзакция запускается, текущая ссылка сохраняется в ThreadLocal, чтобы обеспечить изоляцию между несколькими потоками:

Как видите, этот объект ресурса является объектом ThreadLocal.

Присвоение выполняется следующим методом:

org.springframework.jdbc.datasource.DataSourceTransactionManager#doBegin

В методе bindResource текущая ссылка привязана к текущему потоку, а ресурс — это ThreadLocal, о котором мы только что сказали:

То есть каждый поток играет по-своему Мы не можем нарушать правила использования ThreadLocal и позволять каждому потоку использовать один и тот же ThreadLocal, верно?

Тиези, если ты сделаешь это, ты далеко не уйдешь?

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

По крайней мере, так я думал раньше.

Но все немного изменилось.

Скажи сцену, рутинная реализация

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

Итак, давайте сначала посмотрим, что это за сцена.

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

Для бизнес-системы все эти 50-ваттные фрагменты данных должны храниться в библиотеке, а не один. Или вообще не вставлять.

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

Так как не плохо сказать одно, то для всех интуитивно должно быть два решения:

1. Транзакции вставляются одна за другой в цикл for.
2. Прямая вставка оператора в пакетах.

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

Эффективность очень низкая, позвольте мне показать вам.

Например, у нас есть таблица Student, структура таблицы очень проста, а именно:

CREATE TABLE `student` (
  `id` bigint(63) NOT NULL AUTO_INCREMENT,
  `name` varchar(32) DEFAULT NULL,
  `home` varchar(64) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

В нашем проекте мы вставляем данные через цикл for, а метод аннотируется @Transactional:

Параметр num — это данные, которые мы передаем через внешний запрос, что означает, что нужно вставить num фрагментов данных:

В этом случае мы можем имитировать вставку указанного количества данных по следующей ссылке:

http://127.0.0.1:8081/insertOneByOne?num=xxx

Я попытался установить число на 50 Вт и позволить ему работать медленно, но я был еще слишком молод, чтобы ждать очень долго результата.

Поэтому я изменил число на 5000, и результат выглядит следующим образом:

insertOneByOne执行耗时:133449ms,num=5000

Вставка 5000 элементов данных один за другим занимает 133,5 с.

При такой скорости для вставки 50-ваттных фрагментов данных требуется 13350 с, что, вероятно, составляет столько-то часов:

Кто это выдержит.

Поэтому эта схема имеет огромное поле для оптимизации.

Например, мы оптимизируем для пакетной вставки следующим образом:

Соответствующий оператор sql выглядит следующим образом:

insert into table ([列名],[列名]) VALUES ([列值],[列值]), ([列值],[列值]);

Мы по-прежнему звоним через фронтальный интерфейс:

Когда для нашего числа установлено значение 5000, моя страница обновляется 10 раз, и вы можете видеть, что время в основном находится в пределах 200 мс:

От 133,5с до 200мс, друзья, что это?

Это качественный скачок. Производительность увеличилась почти в 667 раз.

Почему объемные вставки сделали такой большой скачок?

Вы думаете, что до того, как был вставлен цикл for, хотя SpringBoot 2.0 по умолчанию использует HikariPool, пул соединений по умолчанию даст вам 10 соединений.

Но вам нужно только одно соединение, чтобы начать транзакцию. Это не требует много времени.

Место, отнимающее много времени, — это ваши 5000 операций ввода-вывода.

Поэтому это занимает много времени.

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

Почему бы не начать сделку?

У вас есть молоток, чтобы начать транзакцию с sql?

Итак, что, если мы вставим 50 Вт фрагментов данных за один раз?

Давай, подними волну и попробуй:

http://127.0.0.1:8081/insertBatch?num=500000

Вы можете видеть, что было выбрано исключение. И сообщение об ошибке очень ясное:

Packet for query is too large (42777840 > 1048576). You can change this value on the server by setting the max_allowed_packet' variable.; nested exception is com.mysql.jdbc.PacketTooBigException: Packet for query is too large (42777840 > 1048576).You can change this value on the server by setting the max_allowed_packet' variable.

Скажи, что твоя сумка слишком большая. Размер пакета можно изменить, установив max_allowed_packet.

Мы можем запросить текущий размер конфигурации с помощью следующего оператора:

select @@max_allowed_packet;

Видно, что это 1048576, то есть 1024*1024, размер 1М.

И размер пакета, который нам нужно передать, составляет 42777840 байт, что составляет около 41 МБ.

Поэтому нам нужно изменить размер конфигурации.

Это место также напоминает всем:Если ваш оператор sql очень большой и содержит большие поля, не забудьте настроить этот параметр mysql.

Его можно изменить, изменив файл конфигурации или непосредственно выполнив оператор sql.

Я использую оператор sql, чтобы изменить его на 64M здесь:

set global max_allowed_packet = 1024*1024*64;

Затем выполните его снова, и вы увидите, что вставка прошла успешно:

Данные 50w, внешний вид 74s.

Либо все данные представлены, либо ни один из данных не представлен, и требования выполнены.

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

Так как же еще сократить время?

Появилась дерзкая идея

Все, о чем я могу думать, это предложить многопоточность.

Данные 50 Вт. Открываем пять потоков, и один поток обрабатывает 10w данные, если исключения нет, то он будет храниться в библиотеке, а если есть проблема, то будет откатываться.

Это требование хорошо выполняется. Пишите в минутах.

Но добавьте еще одно требование:Данные этих пяти потоков, если проблема с одним потоком, нужно откатить все.

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

Я уже говорил, что это совершенно невозможно реализовать, потому что, когда я упомянул транзакции, я подумал об аннотации @Transactional для их реализации.

Нам нужно только правильно его использовать, и тогда бизнес-логика может быть связана, и нам не нужно или нельзя вмешиваться в открытие и фиксацию или откат транзакции.

Мы называем этот код написанием декларативной транзакции.

Соответствующая декларативная транзакция является программной транзакцией.

С программными транзакциями у нас есть полный контроль над открытием и фиксацией или откатом транзакций.

Подумайте о программных транзакциях, и это, по сути, половина дела.

Думаете, сначала у нас есть глобальная переменная типа Boolean, которая по умолчанию является коммитируемой.

В дочернем потоке мы можем запустить транзакцию через программную транзакцию, а затем вставить 10w кусков данных, но не фиксировать. Заодно сообщить основному потоку, что я здесь готов и ввести ожидание.

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

Наконец, основной поток собирает статус 5 дочерних потоков.

Если есть проблема с потоком, установите для глобальной переменной значение uncommittable.

Затем разбудите все ожидающие дочерние потоки и выполните откат.

В соответствии с описанным выше процессом код моделирования написан следующим образом, вы можете напрямую скопировать его и запустить:

public class MainTest {
    //是否可以提交
    public static volatile boolean IS_OK = true;

    public static void main(String[] args) {
        //子线程等待主线程通知
        CountDownLatch mainMonitor = new CountDownLatch(1);
        int threadCount = 5;
        CountDownLatch childMonitor = new CountDownLatch(threadCount);
        //子线程运行结果
        List<Boolean> childResponse = new ArrayList<Boolean>();
        ExecutorService executor = Executors.newCachedThreadPool();
        for (int i = 0; i < threadCount; i++) {
            int finalI = i;
            executor.execute(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + "：开始执行");
// if (finalI == 4) {
// throw new Exception("出现异常");
// }
                    TimeUnit.MILLISECONDS.sleep(ThreadLocalRandom.current().nextInt(1000));
                    childResponse.add(Boolean.TRUE);
                    childMonitor.countDown();
                    System.out.println(Thread.currentThread().getName() + "：准备就绪,等待其他线程结果,判断是否事务提交");
                    mainMonitor.await();
                    if (IS_OK) {
                        System.out.println(Thread.currentThread().getName() + "：事务提交");
                    } else {
                        System.out.println(Thread.currentThread().getName() + "：事务回滚");
                    }
                } catch (Exception e) {
                    childResponse.add(Boolean.FALSE);
                    childMonitor.countDown();
                    System.out.println(Thread.currentThread().getName() + "：出现异常,开始事务回滚");
                }
            });
        }
        //主线程等待所有子线程执行response
        try {
            childMonitor.await();
            for (Boolean resp : childResponse) {
                if (!resp) {
                    //如果有一个子线程执行失败了，则改变mainResult，让所有子线程回滚
                    System.out.println(Thread.currentThread().getName()+":有线程执行失败，标志位设置为false");
                    IS_OK = false;
                    break;
                }
            }
            //主线程获取结果成功，让子线程开始根据主线程的结果执行（提交或回滚）
            mainMonitor.countDown();
            //为了让主线程阻塞，让子线程执行。
            Thread.currentThread().join();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

Когда все дочерние потоки работают нормально, вывод выглядит следующим образом:

Результаты соответствуют нашим ожиданиям.

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

Исключение возникает в одном потоке, и все потоки откатываются, что вроде бы соответствует ожиданиям.

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

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

Другая половина — двухэтапная фиксация (2PC).

Рисуем тыкву по совку

С совком впереди, разве вам не очень просто нарисовать тыкву?

Не так много кода,Пример кода можно нажатьздесьполучить, поэтому позвольте мне сделать скриншот здесь:

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

Излишне говорить, что вы можете знать пальцами ног, это определенно быстрее, чем вставка фрагментов данных по 50 Вт в пакетах за один раз.

Насчет скорости, ничего лишнего, просто посмотрите на эффект исполнения.

Как наш контроллер, как это:

Итак, звоните по ссылке:

http://127.0.0.1:8081/batchHandle

Результат выглядит следующим образом:

Помните, как трудоемки были наши пакетные вставки?

73791 мс.

С 73791 мс до 15719 мс. Почти как 58s.

Уже очень хорошо.

Так что, если поток выдает исключение? Например:

Посмотрим на вывод журнала:

Судя по анализу журнала, он соответствует требованиям.

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

Действительно ли он соответствует требованиям?

Соответствуйте требованиям, просто посмотрите.

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

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

Недостаток на 2ПК.

Как я обсуждаю с читателями, как это:

Мы не можем идти дальше, и тогда есть распределенные транзакции, такие как 3PC, TCC и Seata.

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

Фактически, когда мы понимаем каждый подпоток как каждую подсистему в микросервисе, это сценарий распределенной транзакции.

И решение, которое мы придумали, не является идеальным решением.

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

Поэтому я называю это решение: программирование на основе удачи, обмен удачи на время.

Меры предосторожности

Есть несколько вещей, на которые стоит обратить внимание в приведенном выше коде.

Напоминание всем.

Первый: сколько потоков включить для вставки данных распределения, этот параметр можно настроить.

Например, я изменил его на 10 потоков, и каждый поток вставлял 5w фрагментов данных. Тогда время выполнения на 2 секунды быстрее:

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

секунда: Именно потому, что количество запущенных потоков можно регулировать и даже вычислять каждый раз.

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

Вы думаете, если нам нужно запустить 5 дочерних потоков, а количество основных потоков всего 4, то задача попала в очередь.

Тогда эти 4 основных потока будут все время блокироваться, ожидая пробуждения основного потока.

А что в это время делает основной поток?

Ожидание 5 потоков для запуска результатов, но он может собрать только 4 результата.

Так что будет ждать.

Третий: Вот несколько потоков, которые открыли транзакции для вставки данных в таблицу, остерегайтесь взаимоблокировок базы данных.

четвертый: Обратите внимание на код в программе.Стандартный способ написания установки countDown - поместить его в блок finally code.Ради красоты скриншота я пропустил этот шаг:

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

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

Это также еще раз доказывает, что многопоточные транзакции — ложное утверждение.

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

шестой: Многопоточные транзакции можно рассматривать как распределенные транзакции с другой точки зрения. , этот случай можно использовать для понимания распределенных транзакций. Но лучший способ решения распределенных транзакций: не иметь распределенных транзакций!

И большинство посадочных решений для решения распределенных транзакций: согласованность в конечном итоге.

Экономичен и приемлем для большинства предприятий.

Седьмой: Если вы хотите получить это решение для производственного использования, не забудьте сначала связаться со своими коллегами по бизнесу, чтобы узнать, можете ли вы принять эту ситуацию. Дилемма между скоростью и безопасностью.

При этом оставьте интерфейс для ручной доработки:

Последнее слово

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

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

Кроме того, пожалуйста, следуйте за мной.