Canal과 RocketMQ를 활용한 MySQL 데이터 증분 동기화 구현

개요

Canal은 MySQL의 바이너리 로그(binlog)를 기반으로 증분 데이터를 실시간으로 파싱하고, 이를 외부 시스템에 전달하는 오픈소스 미들웨어입니다. 주요 원리는 MySQL 슬레이브처럼 동작하여 마스터 DB로부터 binlog 스트림을 수신한 후, 해당 이벤트를 해석해 외부 소비자에게 제공하는 것입니다.

주요 사용 사례로는 다음과 같은 시나리오가 있습니다:

데이터베이스 실시간 백업
검색 인덱스(예: Elasticsearch)와의 연동
캐시 무효화 또는 갱신
다양한 메시지 큐를 통한 이벤트 기반 아키텍처 구성
데이터 웨어하우스로의 ETL 처리

지원하는 MySQL 버전은 5.1부터 8.0까지이며, 현재 대부분의 프로덕션 환경에서 안정적으로 사용되고 있습니다.

MySQL 사전 설정

Canal이 정상 작동하기 위해선 MySQL 서버가 ROW 포맷의 binlog를 기록하도록 설정되어야 합니다. 다음 내용을 /etc/my.cnf 파일에 추가합니다:

[mysqld]
log-bin=mysql-bin
binlog-format=ROW
server_id=1

설정 완료 후 MySQL 서비스를 재시작합니다. 이후 전용 계정을 생성하여 권한을 부여합니다:

CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';
CREATE DATABASE canal DEFAULT CHARSET utf8mb4;
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
GRANT ALL PRIVILEGES ON canal.* TO 'canal'@'%';
FLUSH PRIVILEGES;

테스트를 위해 아래와 같이 user 테이블을 생성하고 샘플 데이터를 삽입합니다:

USE canal;

CREATE TABLE user (
  id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
  username VARCHAR(50),
  password VARCHAR(50),
  email VARCHAR(45),
  phone VARCHAR(15)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

INSERT INTO user (id, username, password, email, phone) VALUES 
(1, '김민수', 'pass123', 'minsu@example.com', '01012345678'),
(2, '이영희', 'pass456', 'younghui@example.com', '01087654321');

Canal Server 설치 및 실행

최신 안정 버전인 1.1.5를 다운로드하고 압축을 해제합니다:

mkdir -p /opt/canal && cd /opt/canal
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz
tar -xzf canal.deployer-1.1.5.tar.gz

인스턴스 설정 파일(conf/example/instance.properties)을 수정하여 MySQL 연결 정보를 반영합니다:

canal.instance.mysql.slaveId=101
canal.instance.master.address=192.168.3.13:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal

서버를 시작합니다:

sh bin/startup.sh

방화벽에서 Canal 기본 포트인 11111번을 개방합니다:

firewall-cmd --add-port=11111/tcp --permanent && firewall-cmd --reload

Docker를 통한 Canal 배포

컨테이너 기반 운영을 원할 경우, 아래와 같은 docker-compose.yml을 작성할 수 있습니다:

version: '3'
services:
  canal-server:
    image: canal/canal-server:v1.1.5
    container_name: canal_server
    ports:
      - "11111:11111"
    environment:
      - canal.instance.mysql.slaveId=101
      - canal.instance.master.address=192.168.3.13:3306
      - canal.instance.dbUsername=canal
      - canal.instance.dbPassword=canal
    volumes:
      - ./logs:/home/admin/canal-server/logs
      - ./conf:/home/admin/canal-server/conf

RocketMQ 설치 (Docker 기반)

메시지 브로커로 RocketMQ를 사용하기 위해 Docker Compose를 통해 구성합니다. 먼저 docker-compose.yml을 작성합니다:

version: '3.5'
services:
  namesrv:
    image: foxiswho/rocketmq:server
    container_name: mq_namesrv
    ports:
      - "9876:9876"
    networks:
      - rmq_net

  broker:
    image: foxiswho/rocketmq:broker
    container_name: mq_broker
    ports:
      - "10911:10911"
      - "10909:10909"
    volumes:
      - ./broker.conf:/etc/rocketmq/broker.conf
    environment:
      NAMESRV_ADDR: "namesrv:9876"
      JAVA_OPTS: "-Duser.home=/opt"
    command: mqbroker -c /etc/rocketmq/broker.conf
    depends_on:
      - namesrv
    networks:
      - rmq_net

  console:
    image: styletang/rocketmq-console-ng
    container_name: mq_console
    ports:
      - "8076:8080"
    environment:
      JAVA_OPTS: "-Drocketmq.namesrv.addr=namesrv:9876"
    depends_on:
      - namesrv
    networks:
      - rmq_net

networks:
  rmq_net:
    driver: bridge

broker.conf 설정 예시:

brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=0
brokerIP1=192.168.3.13
listenPort=10911
namesrvAddr=namesrv:9876
autoCreateTopicEnable=true
brokerRole=ASYNC_MASTER
flushDiskType=ASYNC_FLUSH

실행 후 http://192.168.3.13:8076에서 관리 콘솔 확인 가능합니다.

Canal + RocketMQ 연동 설정

Canal 서버가 변경 이벤트를 RocketMQ로 전송하도록 설정합니다. conf/canal.properties에서 모드를 지정합니다:

canal.serverMode = rocketMQ
rocketmq.producer.group = GROUP_CANA_PRODUCER
rocketmq.namesrv.addr = 192.168.3.13:9876

인스턴스별 토픽 이름을 설정합니다 (conf/example/instance.properties):

canal.mq.topic = topic_mysql_change_log

변경 사항 적용을 위해 Canal을 재시작합니다:

sh bin/restart.sh

Consumer 구현 예시 (Spring Boot 기반)

RocketMQ로부터 전달받은 데이터를 처리하는 컨슈머 예제입니다:

@Component
@RocketMQMessageListener(
    topic = "topic_mysql_change_log",
    consumerGroup = "group_canal_consumer"
)
public class MysqlChangeConsumer implements RocketMQListener<String> {

    private static final Logger log = LoggerFactory.getLogger(MysqlChangeConsumer.class);

    @Override
    public void onMessage(String message) {
        try {
            // JSON 형식의 메시지를 파싱
            JSONObject json = JSON.parseObject(message);
            String database = json.getString("database");
            String table = json.getString("table");
            String type = json.getString("type"); // INSERT, UPDATE, DELETE

            JSONArray data = json.getJSONArray("data");
            log.info("DB 변경 감지 - DB: {}, Table: {}, Type: {}, Rows: {}건", 
                     database, table, type, data.size());

            for (int i = 0; i < data.size(); i++) {
                log.debug("변경 데이터 {}: {}", i + 1, data.getJSONObject(i));
            }

        } catch (Exception e) {
            log.error("메시지 처리 실패: {}", e.getMessage(), e);
        }
    }
}

실제로 MySQL에서 데이터를 수정하면, 위 리스너가 메시지를 수신하여 로그에 출력합니다.

Canal Admin 관리 콘솔 설정

Canal 인스턴스를 중앙에서 관리하기 위해 Admin 콘솔을 설치합니다:

mkdir /opt/canal-admin && cd /opt/canal-admin
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.admin-1.1.5.tar.gz
tar -xzf canal.admin-1.1.5.tar.gz

conf/application.yml에서 데이터베이스 연결 정보를 설정합니다:

spring.datasource:
  url: jdbc:mysql://192.168.3.13:3306/canal_manager?useUnicode=true&characterEncoding=UTF-8
  username: canal
  password: canal

canal:
  adminUser: admin
  adminPasswd: admin

conf/canal_manager.sql을 DB에 임포트한 후 콘솔을 시작합니다:

sh bin/start.sh

포트 8089를 방화벽에서 개방하고, 웹 브라우저에서 접속합니다. 기본 로그인 정보는 ID: admin / PW: 123456입니다.

기존 Canal 서버의 canal.properties에 아래 설정을 추가해 Admin과 연결합니다:

canal.admin.manager = 192.168.3.13:8089
canal.admin.enable = true

재시작 후 Admin 콘솔에서 인스턴스 상태를 모니터링할 수 있습니다.

태그: Canal RocketMQ MySQL Spring Boot docker

7월 15일 05:18에 게시됨

괴물 클럽