Как Apache ShardingSphere реализует распределенные транзакции?

С увеличением объемов данных и ростом нагрузки на трафик бизнес-системы сталкиваются с серьезными проблемами, и возникла острая потребность в масштабируемости систем баз данных. Проблемы масштабирования в режиме онлайн традиционных автономных баз данных, таких как Oracle, MySQL, SQL Server и PostgreSQL, сегодня очевидны как никогда. Для решения этих проблем масштабирования и была придумана распределенная база данных, способная масштабироваться. Все это означает, что распределенные транзакции являются проблемой, которую необходимо решать.

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

Содержание

Обработка транзакций
Конструкция транзакций ShardingSphere
Примеры
План на будущее

Обработка транзакций

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

Атомарность

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

Долговечность

Операция, совершенная транзакцией, действительна даже при отключении электроэнергии.

Согласованность

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

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

На банковском счете А имеется 500$, а на счете Б – 500$, итого 1 000$. После того как A и B выполнят операцию перевода в транзакции, их общая сумма по-прежнему будет равна 1 000$.

Изоляция

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

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

Атомарность. Для автономных транзакций использование журнала отмены и журнала повторных операций гарантирует полную фиксацию или полный откат. Однако в распределенных транзакциях задействовано несколько физических узлов, каждый из которых отличается друг от друга. Журналы одних узлов могут быть записаны успешно, а других – неудачно.
Нестабильность сети. Для автономной машины связь стабильна, и на любую операцию можно ответить, успешна она или нет. Однако в распределенном сценарии сеть нестабильна, и ответ на операцию может быть не получен. Поэтому возникает проблема, как обеспечить доступность распределенных транзакций (очистка и восстановление нештатных транзакций и т.д.).
Контроль параллелизма. С появлением MVCC линеаризуемость операций стала жестким требованием. Глобальное увеличение числа транзакций может быть легко осуществлено в автономной базе данных, но не в распределенном сценарии.

Решения
Атомарная фиксация

Основным решением проблем атомарности и нестабильности сети является протокол 2PC (двухфазный протокол фиксации), который определяет две роли – TM (менеджер транзакций) и RM (менеджер ресурсов).

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

Первая фаза: RM блокирует связанные ресурсы и выполняет определенные операции, а затем возвращает TM данные об успехе или неудаче.

Вторая фаза: в соответствии с результатами, полученными RM в первой фазе, TM выполняет финальные операции фиксации (изменение состояния транзакции, удаление состояния блокировки и т.д.), если все операции успешны, и откат в случае неудачи.

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

Примечание: Протокол двухфазной фиксации решает только проблему фиксации. Либо фиксация проходит успешно, либо неудачно. Не существует промежуточного состояния частичного успеха. Это не обязательно связано с уровнем изоляции транзакции.

Контроль параллелизма

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

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

Определение гранулярности параллелизма. Например, в MySQL есть блокировки строк (гранулярность блокировки – одна строка), блокировки таблиц (гранулярность блокировки – одна таблица) и так далее.
Три сценария параллелизма:a. Параллелизм чтения. Специальная обработка не требуется, поскольку данные не изменяются.b. Параллельная запись. Не следует выполнять одновременную запись, иначе произойдет повреждение данных.c. Параллельное чтение/запись. В этом сценарии в основном выполняется оптимизация производительности. Существует множество механизмов управления параллелизмом, среди которых наиболее популярным является Multiversion Concurrency Control.

Модель MVCC
Два основных режима реализации:

На основе ID транзакции и ReadView
Идентификатор транзакции получается для каждой транзакции, чтобы определить последовательность, в которой начинается транзакция. Снимки получаются через активный список для хранения нескольких версий данных с идентификатором транзакции, чтобы достичь эффекта управления параллелизмом. MySQL и Postgres-XL используют этот подход.
На основе временной метки
Введя временную метку, видимость можно определить, добавив к данным атрибуты, связанные с временной меткой, и сравнив коммиты (временную метку коммита) и временную метку снапшота данных, чтобы достичь линеаризованного контроля параллелизма. Именно так поступил Spanner.

Два вышеуказанных режима зависят от генерации глобальных номеров транзакций. К распространенным механизмам генерации относятся TrueTime (используется Spanner), HLC (CockroachDB использует HLC с ошибками) и TSO (Timestamp Oracle).

Конструкция транзакций ShardingSphere

Функция транзакций ShardingSphere построена на локальной транзакции хранилища БД, обеспечивая три режима транзакций: LOCAL, XA и BASE. Вам нужно будет использовать только собственный режим транзакций (begin/commit/roll-back), чтобы использовать эти три режима и сделать соответствующие компромиссы между согласованностью и производительностью.

ЛОКАЛЬНЫЙ

Режим LOCAL непосредственно строится на локальной транзакции БД хранилища. Он имеет самую высокую производительность, хотя существует проблема с атомарностью. Если вы можете мириться с этой проблемой, то это хороший выбор.

В режиме XA протокол XA основан на наборе протоколов взаимодействия, определенных 2PC. Он определяет интерфейс xa start/prepare/end/commit/rollback. Широко используемыми реализациями являются Narayana, Atomics, а ShardingSphere интегрирует XA-реализацию Narayana и Atomics.

Приложение подключается к прокси, и прокси создает объект сессии, привязанный к соединению.
Приложение выполняет команду begin. Proxy создает логическую транзакцию через Narayana TM и привязывает ее к текущей сессии.
Приложение выполняет определенный SQL. Сессия устанавливает соединение с БД хранилища, и соединение может быть зарегистрировано в транзакции через интерфейс Transaction.enlistResource (). Затем выполнитеXA START {XID} для запуска транзакции и выполнения SQL, перезаписанного маршрутизацией.
Приложение выполняет команду commit, выполняет xa prepare для каждого соединения, зарегистрированного в базе данных транзакции, обновляет статус транзакции до prepared, и выполняет xa commit для каждого соединения. Если система возвращает ok, обновляет статус транзакции до Committed, и транзакция успешно фиксируется. Если процесс prepare завершился неудачно, вы можете выполнить команду rollback для отката данных. В противном случае фоновый процесс очистит данные.
Приложение выполняет команду rollback, а соединение, зарегистрированное в транзакции, соединяющей БД хранилища, выполняет команду xa rollback соответственно.

BASE

Режим BASE (Basically Available, Soft State, Eventually Consistent). Транзакция BASE является результатом баланса между C и A в теореме CAP. Режим AT Seata является реализацией транзакции BASE, а ShardingSphere интегрирует реализацию AT в Seata.

Приложение подключается к прокси, и прокси создает объект сессии, привязанный к соединению.
Приложение выполняет команду begin. Proxy создает логическую транзакцию через Seata TM, привязывает ее к текущей сессии и регистрирует ее на Seata Server.
Приложение выполняет логический SQL. Сессия устанавливает соединение с БД хранилища. Каждое соединение является ConnectionProxy экземпляром Seata. А затем разбирает фактический sql, переписанный маршрутизацией и выполняет перехваты. Например, если это операция модификации, выполните begin для получения локальной блокировки, выполните SQL запрос, выполните commit для освобождения локальной блокировки, и сообщите результаты транзакции ветвления на сервер Seata.
После того как приложение выполнит команду commit, Seata TM в Proxy уведомляет Seata Server и напрямую возвращается к приложению. Сервер Seata асинхронно взаимодействует с прокси для удаления журналов транзакций.
Приложение выполняет команду rollback. После того, как Seata TM в прокси уведомит сервер Seata, прокси напрямую возвращается к приложению. Сервер Seata асинхронно взаимодействует с прокси, выполняет операции компенсации и удаляет журналы транзакций.

Примеры

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

Скачайте установочный пакет с официального сайта, распакуйте его в каталог ${ShardingSphere}, и добавьте следующие пакеты jar в каталог ${ShardingSphere}/lib.

（Download：https://mvnrepository.com/）

jta-5.12.4.Final.jar
arjuna-5.12.4.Final.jar
common-5.12.4.Final.jar
jboss-connector-api_1.7_spec-1.0.0.Final.jar                                             | ------------------------------------------------------------------------------------------------------------------------------------
jboss-logging-3.2.1.Final.jar                                                    | ------------------------------------------------------------------------------------------------------------------------------------
jboss-transaction-api_1.2_spec-1.0.0.Alpha3.jar                                           | ------------------------------------------------------------------------------------------------------------------------------------
jboss-transaction-spi-7.6.0.Final.jar
mysql-connector-java-5.1.47.jar                                                   | ------------------------------------------------------------------------------------------------------------------------------------
narayana-jts-integration-5.12.4.Final.jar
shardingsphere-transaction-xa-narayana-5.1.1-SNAPSHOT.jar

Вход в полноэкранный режим Выйти из полноэкранного режима

Подготовка экземпляров MySQL
Подготовьте два экземпляра MySQL: 127.0.0.1:3306 и 127.0.0.1:3307.
Создайте пользователя root с паролем 12345678 для каждого экземпляра MySQL.
Создайте тестовую библиотеку для каждого экземпляра MySQL.

Конфигурация ShardingSphere-Proxy
Измените конфигурацию транзакций в server.yaml.

rules:
  - !AUTHORITY
    users:
      - root@%:root
      - sharding@:sharding
    provider:
      type: ALL_PRIVILEGES_PERMITTED
  - !TRANSACTION
    defaultType: XA
    providerType: Narayana

Вход в полноэкранный режим Выход из полноэкранного режима

Изменить conf/conf-sharding.yaml

dataSources:
  ds_0:
    url: jdbc:mysql://127.0.0.1:3306/test?serverTimezone=UTC&allowPublicKeyRetrieval=true&useSSL=false
    username: root
    password: 12345678
    connectionTimeoutMilliseconds: 30000
    idleTimeoutMilliseconds: 60000
    maxLifetimeMilliseconds: 1800000
    maxPoolSize: 50
    minPoolSize: 1
  ds_1:
    url: jdbc:mysql://127.0.0.1:3307/test?serverTimezone=UTC&allowPublicKeyRetrieval=true&useSSL=false
    username: root
    password: 12345678
    connectionTimeoutMilliseconds: 30000
    idleTimeoutMilliseconds: 60000
    maxLifetimeMilliseconds: 1800000
    maxPoolSize: 50
    minPoolSize: 1

rules:
  - !SHARDING
    tables:
      account:
        actualDataNodes: ds_${0..1}.account${0..1}
        tableStrategy:
          standard:
            shardingColumn: id
            shardingAlgorithmName: account_inline
        keyGenerateStrategy:
          column: id
          keyGeneratorName: snowflake
    defaultDatabaseStrategy:
      standard:
        shardingColumn: id
        shardingAlgorithmName: database_inline
    defaultTableStrategy:
      none:

    shardingAlgorithms:
      database_inline:
        type: INLINE
        props:
          algorithm-expression: ds_${id % 2}
      account_inline:
        type: INLINE
        props:
          algorithm-expression: account${id % 2}

    keyGenerators:
      snowflake:
        type: SNOWFLAKE
        props:
          worker-id: 123

Войти в полноэкранный режим Выйти из полноэкранного режима

Запуск ShardingSphere-Proxy
Запустите прокси, выполнив следующую команду:

cd ${ShardingSphere}
./bin/start.sh

Войти в полноэкранный режим Выйти из полноэкранного режима

Использование ShardingSphere-Proxy
Используйте MySQL Client для подключения ShardingSphere-Proxy для тестирования, выполните следующую команду.

mysql -h127.0.0.1 -P3307 -uroot -proot
mysql> use sharding_db;
Database changed
mysql> create table account(id int, balance float ,transaction_id int);
Query OK, 0 rows affected (0.12 sec)

mysql> select * from account;
Empty set (0.02 sec)

mysql> begin;
Query OK, 0 rows affected (0.09 sec)

mysql> insert into account(id, balance, transaction_id) values(1,1,1),(2,2,2);
Query OK, 2 rows affected (0.53 sec)

mysql> select * from account;
+------+---------+----------------+
| id  | balance | transaction_id |
+------+---------+----------------+
|  2 |   2.0 |       2 |
|  1 |   1.0 |       1 |
+------+---------+----------------+
2 rows in set (0.03 sec)

mysql> commit;
Query OK, 0 rows affected (0.05 sec)

mysql> select * from account;
+------+---------+----------------+
| id  | balance | transaction_id |
+------+---------+----------------+
|  2 |   2.0 |       2 |
|  1 |   1.0 |       1 |
+------+---------+----------------+
2 rows in set (0.02 sec)

Войти в полноэкранный режим Выйти из полноэкранного режима

План на будущее

В настоящее время распределенная транзакция ShardingSphere интегрирует схему реализации 2PC третьей стороны, чтобы гарантировать атомарность. Изоляция зависит от гарантии изоляции БД хранилища, обеспечивая доступные функции транзакций.

Будущая реализация MVCC, основанная на глобальной метке времени и объединенная с 2PC, обеспечит лучшую поддержку семантики изоляции транзакций.

Ссылки на проект Apache ShardingSphere:
ShardingSphere Github

ShardingSphere Twitter

ShardingSphere Slack

Руководство для контрибьюторов

Автор
Лу Цзиншань

Коммиттер Apache ShardingSphere и инженер по инфраструктуре R&D в компании SphereEx.
Увлекается открытым исходным кодом и технологиями баз данных.
Фокусируется на разработке транзакционного модуля Apache ShardingSphere.

Как Apache ShardingSphere реализует распределенные транзакции?

Обработка транзакций

Конструкция транзакций ShardingSphere

Примеры

План на будущее

Рекомендуем почитать: