Canal hoạt động bằng cách giả lập vai trò là một máy chủ MySQL phụ (Slave), kết nối tới máy chủ chính (Master) để lấy dữ liệu từ tệp Binlog. Sau đó, Canal xử lý và lưu trữ dữ liệu này để các ứng dụng khác có thể tiêu thụ. Hệ thống Canal bao gồm hai thành phần chính: máy chủ (Server) và máy khách (Client). Máy chủ chịu trách nhiệm kết nối tới các phiên bản MySQL khác nhau, quản lý một hàng đợi thông báo sự kiện cho mỗi phiên bản. Máy khách có thể đăng ký nhận các sự kiện thay đổi dữ liệu từ những hàng đợi này để xử lý và lưu trữ vào kho dữ liệu.
Cấu hình máy chủ MySQL
Phiên bản MySQL được sử dụng trong hướng dẫn này là 5.6.33.
-
Chỉnh sửa tệp cấu hình
my.confcủa MySQL:[mysqld] log-bin=mysql-bin binlog-format=ROW server-id=1log-bin=mysql-bin: Kích hoạt ghi log nhị phân (binlog).binlog-format=ROW: Chọn định dạng ROW cho binlog.server-id=1: Định nghĩa ID cho máy chủ MySQL, cần thiết cho cơ chế nhân bản (replication). Đảm bảo giá trị này không trùng vớislaveIdcủa Canal.
Lưu ý: Đối với dịch vụ Alibaba Cloud RDS for MySQL, binlog thường được bật sẵn và tài khoản mặc định đã có quyền truy cập binlog. Bạn có thể bỏ qua bước này.
-
Cấp quyền cho tài khoản Canal truy cập MySQL với vai trò là slave:
CREATE USER canal IDENTIFIED BY 'canal'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; FLUSH PRIVILEGES;Khởi động lại máy chủ MySQL:
service mysql restart
Cấu hình máy chủ Canal
-
Tải xuống Canal: Truy cập trang releases của Canal và tải về tệp gói phù hợp. Ví dụ với phiên bản 1.1.5:
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz -
Giải nén tệp đã tải về:
mkdir /tmp/canal tar zxvf canal.deployer-1.1.5.tar.gz -C /tmp/canalSau khi giải nén, bạn sẽ thấy cấu trúc thư mục như sau trong
/tmp/canal:bin/ # Thư mục chứa các script thực thi conf/ # Thư mục chứa tệp cấu hình lib/ # Thư mục chứa các thư viện logs/ # Thư mục chứa file log -
Chỉnh sửa tệp cấu hình
conf/example/instance.properties:## mysql serverId canal.instance.mysql.slaveId = 1234 # position info, update with your database details canal.instance.master.address = 127.0.0.1:3306 canal.instance.master.journal.name = canal.instance.master.position = canal.instance.master.timestamp = #username/password, update with your database details canal.instance.dbUsername = canal canal.instance.dbPassword = canal canal.instance.connectionCharset = UTF-8 #table regex canal.instance.filter.regex = .*\\..*canal.instance.mysql.slaveId: ID duy nhất cho Canal instance.canal.instance.master.address: Địa chỉ và cổng của máy chủ MySQL chính.canal.instance.dbUsernamevàcanal.instance.dbPassword: Tên người dùng và mật khẩu để kết nối tới MySQL.canal.instance.connectionCharset: Mã hóa ký tự của cơ sở dữ liệu (ví dụ: UTF-8, GBK).canal.instance.filter.regex: Biểu thức chính quy để lọc các bảng cần theo dõi.
Nếu hệ thống chỉ có 1 CPU, hãy đặt
canal.instance.parser.parallelthànhfalse. -
Khởi động máy chủ Canal:
sh bin/startup.sh -
Kiểm tra log của máy chủ Canal:
tail -f logs/canal/canal.logBạn sẽ thấy các thông báo tương tự như:
2013-02-05 22:45:27.967 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server. 2013-02-05 22:45:28.113 [main] INFO com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[10.1.29.120:11111] 2013-02-05 22:45:28.210 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ...... -
Kiểm tra log của Canal instance (ví dụ:
example):tail -f logs/example/example.logCác thông báo sẽ chỉ ra instance đã khởi động thành công:
2013-02-05 22:50:45.636 [main] INFO c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [canal.properties] 2013-02-05 22:50:45.641 [main] INFO c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [example/instance.properties] 2013-02-05 22:50:45.803 [main] INFO c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example 2013-02-05 22:50:45.810 [main] INFO c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start successful.... -
Dừng máy chủ Canal:
sh bin/stop.sh
Cấu hình máy khách Canal
Cấu hình Maven
Thêm các dependency sau vào tệp pom.xml của dự án Java:
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.1.5</version>
</dependency>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.protocol</artifactId>
<version>1.1.5</version>
</dependency>
Ví dụ mã nguồn máy khách
package canal;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import java.net.InetSocketAddress;
import java.util.List;
public class CanalClientExample {
public static void main(String[] args) {
// Khởi tạo kết nối
CanalConnector connector = CanalConnectors.newSingleConnector(
new InetSocketAddress("127.0.0.1", 11111),
"example", // Tên instance Canal đã cấu hình
"", // Tên đăng nhập (nếu có)
"" // Mật khẩu (nếu có)
);
int batchSize = 1000;
int emptyMessageCount = 0;
int maxEmptyCount = 1200; // Số lần nhận tin nhắn rỗng tối đa trước khi thoát
try {
connector.connect();
connector.subscribe(".*\\..*"); // Đăng ký theo dõi tất cả bảng
connector.rollback(); // Đặt lại vị trí đọc
while (emptyMessageCount < maxEmptyCount) {
Message message = connector.getWithoutAck(batchSize); // Lấy dữ liệu theo lô
long batchId = message.getId();
int entrySize = message.getEntries().size();
if (batchId == -1 || entrySize == 0) {
emptyMessageCount++;
System.out.println("Nhận tin nhắn rỗng, đếm: " + emptyMessageCount);
try {
Thread.sleep(1000); // Chờ trước khi thử lại
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
} else {
emptyMessageCount = 0; // Đặt lại biến đếm nếu nhận được dữ liệu
processEntries(message.getEntries());
}
connector.ack(batchId); // Xác nhận đã xử lý tin nhắn
// connector.rollback(batchId); // Sử dụng khi xử lý thất bại và cần đọc lại
}
System.out.println("Đã nhận quá nhiều tin nhắn rỗng, thoát chương trình.");
} catch (Exception e) {
System.err.println("Lỗi trong quá trình xử lý Canal: " + e.getMessage());
e.printStackTrace();
} finally {
connector.disconnect(); // Ngắt kết nối
}
}
private static void processEntries(List<Entry> entries) {
for (Entry entry : entries) {
if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
continue; // Bỏ qua các sự kiện bắt đầu/kết thúc giao dịch
}
RowChange rowChange;
try {
rowChange = RowChange.parseFrom(entry.getStoreValue());
} catch (Exception e) {
System.err.println("Lỗi phân tích dữ liệu sự kiện: " + entry.toString());
throw new RuntimeException("Lỗi phân tích dữ liệu sự kiện", e);
}
EventType eventType = rowChange.getEventType();
String schemaName = entry.getHeader().getSchemaName();
String tableName = entry.getHeader().getTableName();
String logFileName = entry.getHeader().getLogfileName();
long logFileOffset = entry.getHeader().getLogfileOffset();
System.out.printf("===> Binlog: [%s:%s], Table: [%s,%s], Event Type: %s%n",
logFileName, logFileOffset, schemaName, tableName, eventType);
for (RowData rowData : rowChange.getRowDatasList()) {
if (eventType == EventType.DELETE) {
// Xử lý dữ liệu sau khi xóa
printColumns(rowData.getBeforeColumnsList());
} else if (eventType == EventType.INSERT) {
// Xử lý dữ liệu sau khi chèn
printColumns(rowData.getAfterColumnsList());
} else { // UPDATE
// Xử lý dữ liệu trước và sau khi cập nhật
System.out.println("-------> Before update");
printColumns(rowData.getBeforeColumnsList());
System.out.println("-------> After update");
printColumns(rowData.getAfterColumnsList());
}
}
}
}
private static void printColumns(List<Column> columns) {
for (Column column : columns) {
System.out.println(String.format(" %s : %s (Updated: %s)",
column.getName(), column.getValue(), column.getUpdated()));
}
}
}
Sau khi chạy ứng dụng máy khách và thực hiện thao tác chèn dữ liệu vào một bảng trong cơ sở dữ liệu MySQL, log của máy khách sẽ hiển thị thông tin tương tự như sau:
===> Binlog: [mysql-bin.000042:694], Table: [huagang,users], Event Type: INSERT
id : 14 (Updated: true)
username : AA (Updated: true)
passowrd : AA (Updated: true)
create_date : 2022-01-24 14:40:16 (Updated: true)