Обязательное к прочтению руководство по синхронизации данных клиент-адаптер на основе канала.

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

Являясь компонентом mysql для подписки на данные в режиме реального времени, canal осуществляет сбор данных binlog-файлов mysql.

Хотя Али также открыл исходный код проекта otter (github.com/alibaba/otter, основанный на канале), который синхронизирует данные из mysql в mysql, который реализует одностороннюю синхронизацию и двустороннюю синхронизацию mysql. Однако нам часто нужно синхронизировать данные из mysql с хранилищем, таким как es и hbase, поэтому пользователям приходится использовать canal-client для получения данных для потребления, что хлопотно.

Начиная с версии 1.1.1, Canal реализует вспомогательный лендинговый модуль, который реализует сообщение для подписки Canal, то есть Client-Adapter (GitHub.com/alibaba/canal/wiki/clientadapter).

Текущая версия 1.1.4 последней стабильной версии, адаптер клиента достигла возможности синхронизации данных в RDS, ES, HBASE.

1. Основные возможности Клиент-Адаптера

В настоящее время адаптер имеет следующие основные возможности:

Стыковка с апстримными новостями, включая kafka, RocketMQ, Canal-Server
Инкрементная синхронизация данных mysql
Реализовать полную синхронизацию данных mysql
Нисходящая запись поддерживает mysql, es, hbase

2. Архитектура клиент-адаптер

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

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

3. Настройка миграции и синхронизации (на примере Mysql)

Официальный адрес документа: github.com/alibaba/canal/wiki/Sync-RDB

Меры предосторожности в процессе практики приведены ниже.

3.1 Конфигурация параметров

1) Общий профиль Application.yml

инструкция:

Часть данных может использоваться несколькими группами одновременно. Будет параллельное выполнение между несколькими группами и последовательное выполнение нескольких внешних адаптеров внутри группы, таких как logger и hbase в примере.
В настоящее время клиентский адаптер поддерживает два способа подписки на данные: прямое подключение к серверу канала или подписка на сообщения kafka/RocketMQ.
После заполнения zookeeperHosts он может поддерживать распределенные блокировки, если соединение с Canal-Server находится в кластерном режиме, его все равно необходимо заполнить. Конкретные причины см. в разделе высокой доступности ниже.

2) Настройка адаптера под соответствующую задачу

Конфигурация задач для синхронизации на MySQL находится под траектором CONF / RDB. Файл конфигурации задач, используемый в этой статье, называется mysql1.yml

Уведомление! TargetPK заполняет отношение сопоставления исходного ключа и целевого первичного ключа, SRCPK: Targetpk.

3) Изменение формата журнала

Уровень журнала по умолчанию в logback.xml — отладка. При использовании в Интернете не забудьте изменить его на информацию, иначе журнал взорвется.

3.2 Возможность инкрементной синхронизации

1) Инкрементная синхронизация DML

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

Он будет воспроизводить следующий конкретный журнал.

2) DDL-синхронизация

Если вам нужно использовать возможность синхронизации DDL, вы должны настроить mirroDb на true в rdb.

3.3 Возможность полной синхронизации

Адаптер обеспечивает полные возможности синхронизации, подробнее см. в разделе 3.2 на официальном сайте github.com/alibaba/canal/wiki/ClientAdapter.

Здесь мы используем команду

curl http://127.0.0.1:8081/etl/rdb/mysql1/mysql1.yml -X POST

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

4. Динамическая конфигурация

4.1 Переключатель задач

curl http://127.0.0.1:8081/syncSwitch/dts-dbvtest-insertdata/on -X PUT

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

Соответствующий исходный код реализован следующим образом:

Получить информацию о статусе переключения задач на zk
Если ложь, отключить

4.2 Изменения конфигурации

1) Локальный файл конфигурации

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

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

Рассмотрим принцип реализации.

Унаследованный FileAlterationListenerAdaptor
После обнаружения изменений в файлах
Уничтожить текущий canalAdapterService
Обновить контекстRefresher
Спи в течение 2 секунд
Повторно инициализировать canalAdapterService

Окончательный журнал будет напечатан

2) удаленная настройка на базе mysql

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

Принцип реализации этого также относительно прост:

Обучение локальному асинхронному потоку mysql
Если есть обновление, запишите обновленную конфигурацию в локальный файл конфигурации.
Динамическое обновление

5. Анализ достоверности данных

5.1 механизм подтверждения

Задача адаптера использует многопоточную модель.

Основной поток захватывает очередь записи сообщений MQ Очередь, ожидание блокировки обратного отсчета
Опрос очереди асинхронной очереди потоков, нисходящая доставка
После успешной доставки основной поток освобождает защелку и возвращает подтверждение mq.

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

5.2 Механизм повторных попыток

Параметр retries в application.yml используется для доставки повторных попыток нисходящего потока после опроса из очереди.

Это необходимо тщательно взвесить.Интервал повтора составляет 0,5 с, и можно установить x раз, чтобы избежать дрожания сети и потери данных.

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

6. Анализ проблем с производительностью

Конкретные требования к производительности все еще должны быть подтверждены посредством испытаний под давлением.

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

6.1 Полная синхронная многопоточность

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

Адаптер сделал некоторые конструкции для повышения эффективности полной синхронизации данных. Когда объем полной синхронизации превышает 1 Вт, он открывает несколько потоков. Код выглядит следующим образом:

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

6.2 Выбор полной синхронизации *

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

Глядя на исходный код, я уже рассмотрел эту проблему и запустил потоковый запрос JDBC.

7. Мониторинг сигналов тревоги

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

Хотя адаптер не предоставляет соответствующих API-интерфейсов для мониторинга индикаторов, таких как canal-server, мы все же можем создавать некоторые вспомогательные сигналы мониторинга.

1) тревога накопления сообщений mq

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

2) Зарегистрировать аномальную тревогу

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

Затем измените формат печати журнала, изменив шаблон CONF / Logback.xml для сбора конфигурации.

8. Высокая доступность

Обнаружено путем чтения исходного кода

режим tcp поддерживает HA через zk (не собственная высокая доступность), режим mq не поддерживает zk как HA

Режим TCP требует высокой доступности, и наше понимание высокой доступности отличается.

Поскольку ему необходимо напрямую подключаться к вышестоящему Canal-Server, а высокая доступность Canal-Server приведет к изменению IP-адреса, высокая доступность режима tcp адаптера должна поддерживать это.Он может отслеживать изменения IP-адресов и подключаться к различным восходящим серверы, а не собственная высокая доступность.Архитектура.

Сам режим MQ не поддерживает HA.

Однако, если мы подключим восходящий режим MQ, мы можем сделать хитрую высокую доступность.

После ползания из текущего Binlog MQ, он будет доставляться только в указанную тему в очереди (даже если хеш сделали мульти-очередь, причина такой же), и, следовательно, при использовании режима потребления Cluster MQ, Клиент сможет заказать только соответствующие сообщения расхода в очереди.

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

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

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