Linux环境下Canal安装配置与使用指南

本文档详细介绍了在Linux环境下安装和使用Canal的过程，包括Canal、Kafka的安装配置以及客户端如何消费Kafka消息。Canal主要用于MySQL数据库增量日志解析，提供增量数据订阅和消费服务，适用场景包括数据库镜像、实时备份、索引构建与维护、业务cache刷新和带业务逻辑的增量数据处理。本文档适用于CentOS 7.2系统，MySQL 5.7版本，并推荐使用canal 1.1.4版本。 Canal安装及配置： 1. MySQL配置： - 自建MySQL服务器需开启Binary Log功能，将`binlog-format`设置为ROW模式，并在`my.cnf`配置文件中添加如下内容： ``` [mysqld] log-bin=mysql-bin binlog-format=ROW server_id=1 ``` - 阿里云RDS for MySQL默认已打开binlog，无需额外设置。 2. 授权Canal账号： - 创建名为`canal`的用户并赋予相应的权限： ``` CREATE USER 'canal' IDENTIFIED BY 'canal'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; -- 如果需要全部权限，可以使用 -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%'; FLUSH PRIVILEGES; ``` 3. 安装Canal： - 从GitHub releases页面（https://github.com/alibaba/canal/releases）下载所需版本，例如canal 1.1.4： ``` wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.deployer-1.1.4.tar.gz ``` - 解压并配置Canal，根据实际情况修改`conf/example/instance.properties`中的数据库连接信息。 4. 启动Canal： - 进入解压后的目录，执行启动命令： ``` sh canal.deployer.sh start ``` Kafka安装及配置： 1. 下载Kafka，根据需要选择对应版本，例如Kafka 2.8.0： ``` wget https://downloads.apache.org/kafka/2.8.0/kafka_2.13-2.8.0.tgz ``` - 解压并配置Kafka。 2. 启动Kafka： - 运行Zookeeper： ``` bin/zookeeper-server-start.sh config/zookeeper.properties ``` - 启动Kafka服务： ``` bin/kafka-server-start.sh config/server.properties ``` 3. 创建Kafka主题，例如`canal`： ``` bin/kafka-topics.sh --create --topic canal --partitions 1 --replication-factor 1 --if-not-exists --bootstrap-server localhost:9092 ``` 客户端消费Kafka消息： 1. 编写Java或其他语言的客户端代码，连接Kafka服务器并消费`canal`主题的消息。 2. 示例Java代码（使用Kafka Consumer API）： ```java Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "test"); props.put("enable.auto.commit", "true"); props.put("auto.commit.interval.ms", "1000"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Collections.singletonList("canal")); while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); } ``` 通过以上步骤，你可以在Linux环境中成功安装和配置Canal，实现MySQL的增量数据订阅和消费，同时结合Kafka进行消息中间件的集成。确保每个环节都正确执行，以保证Canal的稳定运行和数据的高效传输。