Docker部署Canal并将消息推送到RabbitMQ

水深无声 2023-01-10 10:27 121阅读 0赞

>  *  GitHub：[https://github.com/alibaba/canal][https_github.com_alibaba_canal]
>  *  Docker：[https://blog.csdn.net/qq\_40794973/category\_9639934.html][https_blog.csdn.net_qq_40794973_category_9639934.html]
>  *  MySQL读写分离：[https://blog.csdn.net/qq\_40794973/article/details/106320629][https_blog.csdn.net_qq_40794973_article_details_106320629]

>  *  [https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart][https_github.com_alibaba_canal_wiki_Canal-Kafka-RocketMQ-QuickStart] 
>  *  [https://github.com/alibaba/canal/wiki/Docker-QuickStart][https_github.com_alibaba_canal_wiki_Docker-QuickStart]

--------------------

# 1 Docker部署MySQL #

创建mysql日志目录和配置文件

###
    rm -rf /mydata/mysql
    ###
    mkdir -p /mydata/mysql/log /mydata/mysql/data /mydata/mysql/conf
    ###
    vi /mydata/mysql/conf/my.cnf

[client]
    default-character-set=utf8
    [mysql]
    default-character-set=utf8
    [mysqld]
    init_connect='SET collation_connection = utf8_unicode_ci' 
    init_connect='SET NAMES utf8' 
    character-set-server=utf8
    collation-server=utf8_unicode_ci
    skip-character-set-client-handshake
    skip-name-resolve
    
    ### 开启 binlog
    log-bin=mysql-bin
    ### 选择 ROW 模式
    binlog-format=ROW
    ### 配置MySQL replaction需要定义，不要和canal的 slaveId重复
    server_id=1

docker启动mysql

docker run \
    -p 33060:3306 \
    --name mysql_canal \
    -v /mydata/mysql/log:/var/log/mysql \
    -v /mydata/mysql/data:/var/lib/mysql \
    -v /mydata/mysql/conf:/etc/mysql \
    --env MYSQL_ROOT_HOST=%.%.%.% \
    --env MYSQL_ROOT_PASSWORD=123456 \
    -d mysql:5.7

连接mysql

> jdbc:mysql://121.36.33.154:33060/?serverTimezone=UTC

创建canal账号

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

创建测试数据库

-- 创建数据库
    CREATE DATABASE user_db;
    USE user_db;
    -- 创建测试表
    CREATE TABLE `tb_user`
    (
        `id`       INT(10) UNSIGNED AUTO_INCREMENT COMMENT '主键ID',
        `username` VARCHAR(20) NOT NULL COMMENT '用户名',
        `password` VARCHAR(64) NOT NULL COMMENT '密码',
        `name`     VARCHAR(20) COMMENT '姓名',
        `age`      TINYINT UNSIGNED COMMENT '年龄',
        `email`    VARCHAR(50) COMMENT '邮箱',
        PRIMARY KEY `id` (`id`),
        UNIQUE KEY `uk_user_username` (`username`)
    );
    -- 插入测试数据
    INSERT INTO `tb_user` (`id`, `username`, `password`, `name`, `age`, `email`)
    VALUES ('1', 'zhangsan', '123456', '张三', '18', 'zhangsan@bug.cn'),
           ('2', 'lisi', '123456', '李四', '20', 'lisi@bug.cn'),
           ('3', 'wangwu', '123456', '王五', '28', 'wangwu@bug.cn'),
           ('4', 'zhaoliu', '123456', '赵六', '21', 'zhaoliu@bug.cn'),
           ('5', 'sunqi', '123456', '孙七', '24', 'sunqi@bug.cn');

--------------------

# 2 Docker部署RabbitMQ #

docker run \
    --name rabbitmq_canal \
    -p 5672:5672 \
    -p 15672:15672 \
    -d rabbitmq:management

默认的登录账户用户名和密码都**guest**

> [http://121.36.33.154:15672/][http_121.36.33.154_15672]

创建Exchanges

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz_size_16_color_FFFFFF_t_70][]

创建Queues

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz_size_16_color_FFFFFF_t_70 1][]

绑定

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz_size_16_color_FFFFFF_t_70 2][]

查看

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz_size_16_color_FFFFFF_t_70 3][]

--------------------

# 3 Docker部署Canal #

注意**1.1.5**版本后支持rabbitmq

>  [https://hub.docker.com/r/canal/canal-server/tags?page=1&ordering=last\_updated][https_hub.docker.com_r_canal_canal-server_tags_page_1_ordering_last_updated]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz_size_16_color_FFFFFF_t_70 4][]

先启动下canal，从里面copy出配置文件

docker run \
    --name canal_temp \
    --rm \
    -d canal/canal-server:latest

拷贝出来的配置文件目录

mkdir -p /mydata/canal/conf

从docker里面拷贝

docker cp canal_temp:/home/admin/canal-server/conf/canal.properties /mydata/canal/conf/
    docker cp canal_temp:/home/admin/canal-server/conf/example/instance.properties /mydata/canal/conf

关闭

docker stop canal_temp

修改instace.properties的这几处位置

vim /mydata/canal/conf/instance.properties

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz_size_16_color_FFFFFF_t_70 5][]

<table> 
 <thead> 
  <tr> 
   <th>key</th> 
   <th>value</th> 
   <th>例子</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>canal.instance.master.journal.name</td> 
   <td>mysql主库链接时起始的binlog文件</td> 
   <td>mysql-bin.000001</td> 
  </tr> 
  <tr> 
   <td>canal.instance.master.position</td> 
   <td>mysql主库链接时起始的binlog偏移量</td> 
   <td>155</td> 
  </tr> 
 </tbody> 
</table>

> dbMessage

修改canal.properties配置

vim /mydata/canal/conf/canal.properties

### tcp, kafka, rocketMQ, rabbitMQ
    canal.serverMode = rabbitMQ
    
    ##################################################
    #########         RabbitMQ           #############
    ##################################################
    rabbitmq.host = 121.36.33.154
    rabbitmq.virtual.host = /
    rabbitmq.exchange = canal.exchange
    rabbitmq.username = guest
    rabbitmq.password = guest

docker启动canal

docker run \
    --name canal \
    -p 11111:11111 \
    -v /mydata/canal/conf/instance.properties:/home/admin/canal-server/conf/example/instance.properties \
    -v /mydata/canal/conf/canal.properties:/home/admin/canal-server/conf/canal.properties \
    -d canal/canal-server:latest

--------------------

# 4 JAVA代码读取 #

/**
     * 
     * <dependency>
     *     <groupId>com.rabbitmq</groupId>
     *     <artifactId>amqp-client</artifactId>
     *     <version>5.6.0</version>
     * </dependency>
     */
    public class CanalConsumer {
        public static Connection connection() throws IOException, TimeoutException {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("10.4.42.32");
            factory.setPort(5672);
            factory.setVirtualHost("/");
            factory.setUsername("guest");
            factory.setPassword("guest");
            return factory.newConnection();
        }
        public static void main(String[] args) throws IOException, TimeoutException {
            Connection connection = connection();
            // 创建Channel
            Channel channel = connection.createChannel();
            String queue = "dbMessage";
            // 接收消息
            Consumer consumer = new DefaultConsumer(channel) {
                /**
                 * 回调方法，当收到消息后，会自动执行该方法
                 *
                 * @param consumerTag  标识
                 * @param envelope     获取一些信息，交换机，路由key...
                 * @param properties   配置信息
                 * @param body         数据
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                    JSONObject jsonObject = JSONObject.parseObject(new String(body));
                    String pretty = JSON.toJSONString(jsonObject, SerializerFeature.PrettyFormat, SerializerFeature.WriteMapNullValue, SerializerFeature.WriteDateUseDateFormat);
                    System.out.println(pretty);
                }
            };
            /**
             * @param queue    队列名称
             * @param autoAck  是否自动确认
             * @param callback 回调对象
             */
            channel.basicConsume(queue, true, consumer);
        }
    }

{
        "data": [
            {
                "id": "5",
                "username": "sunqi55",
                "password": "123456",
                "name": "孙七",
                "age": "24",
                "email": "sunqi@bug.cn"
            }
        ],
        "database": "user_db",
        "es": 1611419308000,
        "id": 5,
        "isDdl": false,
        "mysqlType": {
            "id": "int(10) unsigned",
            "username": "varchar(20)",
            "password": "varchar(64)",
            "name": "varchar(20)",
            "age": "tinyint(3) unsigned",
            "email": "varchar(50)"
        },
        "old": null,
        "pkNames": [
            "id"
        ],
        "sql": "",
        "sqlType": {
            "id": 4,
            "username": 12,
            "password": 12,
            "name": 12,
            "age": -6,
            "email": 12
        },
        "table": "tb_user",
        "ts": 1611419308082,
        "type": "DELETE"
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz_size_16_color_FFFFFF_t_70 6][]

>  *  [https://www.cnblogs.com/liuxuebagaomizhe/p/13809317.html][https_www.cnblogs.com_liuxuebagaomizhe_p_13809317.html]
>  *  [https://www.cnblogs.com/stayed/p/13542069.html][https_www.cnblogs.com_stayed_p_13542069.html]
>  *  [https://blog.csdn.net/zyx6a/article/details/111415184][https_blog.csdn.net_zyx6a_article_details_111415184]

TODO 测试不行，后面完善

public class CanalConsumer {
        /**
         * 队列
         */
        private static final String CANAL_QUEUE_NAME = "channel_sync_user_queue";
        /**
         * 交换机
         */
        public static final String CANAL_EXCHANGE_NAME = "channel_sync_user_exchange";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            Connection connection = ConnectionUtils.getConnection();
            //创建频道
            Channel channel = connection.createChannel();
            //声明(创建)队列
            /**
             * 参数1：队列名称
             * 参数2：是否定义持久化队列
             * 参数3：是否独占本次连接,只能有一个Consumer监听这个队列
             * 参数4：是否在不使用的时候自动删除队列,当没有Consumer时,自动删除
             * 参数5：队列其它参数
             */
            channel.queueDeclare(CANAL_QUEUE_NAME, true, false, false, null);
            /**声明交换机
             * 参数1：交换机名称
             * 参数2：交换机类型，fanout, topic, direct, headers
             */
            channel.exchangeDeclare(CANAL_EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
            //队列绑定交换机
            channel.queueBind(CANAL_QUEUE_NAME, CANAL_EXCHANGE_NAME, "user_backup");
            // 接收消息
            Consumer consumer = new DefaultConsumer(channel) {
                /**
                 * 回调方法，当收到消息后，会自动执行该方法
                 *
                 * @param consumerTag  标识
                 * @param envelope     获取一些信息，交换机，路由key...
                 * @param properties   配置信息
                 * @param body         数据
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                    JSONObject jsonObject = JSONObject.parseObject(new String(body));
                    String pretty = JSON.toJSONString(jsonObject, SerializerFeature.PrettyFormat, SerializerFeature.WriteMapNullValue, SerializerFeature.WriteDateUseDateFormat);
                    System.out.println(pretty);
                }
            };
            /**
             * @param queue    队列名称
             * @param autoAck  是否自动确认
             * @param callback 回调对象
             */
            channel.basicConsume(CANAL_QUEUE_NAME, true, consumer);
        }
    }

[https_github.com_alibaba_canal]: https://github.com/alibaba/canal
[https_blog.csdn.net_qq_40794973_category_9639934.html]: https://blog.csdn.net/qq_40794973/category_9639934.html
[https_blog.csdn.net_qq_40794973_article_details_106320629]: https://blog.csdn.net/qq_40794973/article/details/106320629
[https_github.com_alibaba_canal_wiki_Canal-Kafka-RocketMQ-QuickStart]: https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart
[https_github.com_alibaba_canal_wiki_Docker-QuickStart]: https://github.com/alibaba/canal/wiki/Docker-QuickStart
[http_121.36.33.154_15672]: http://121.36.33.154:15672/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz_size_16_color_FFFFFF_t_70]: /images/20221119/5fd28a004d864eebaee5140888664b57.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz_size_16_color_FFFFFF_t_70 1]: /images/20221119/b38bac21d13949be9ccb97e4155331bf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20210124001140940.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/2021012400121571.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz,size_16,color_FFFFFF,t_70
[https_hub.docker.com_r_canal_canal-server_tags_page_1_ordering_last_updated]: https://hub.docker.com/r/canal/canal-server/tags?page=1&ordering=last_updated
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20210123233840899.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20210123220307991.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20210124001705517.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0OTcz,size_16,color_FFFFFF,t_70
[https_www.cnblogs.com_liuxuebagaomizhe_p_13809317.html]: https://www.cnblogs.com/liuxuebagaomizhe/p/13809317.html
[https_www.cnblogs.com_stayed_p_13542069.html]: https://www.cnblogs.com/stayed/p/13542069.html
[https_blog.csdn.net_zyx6a_article_details_111415184]: https://blog.csdn.net/zyx6a/article/details/111415184