devbox
很楠不爱3
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Java：SpringBoot整合Canal+RabbitMQ组合实现MySQL数据监听_java 处理canal 到mq中的数据

作者：很楠不爱3 | 2024-04-18 09:56:12

赞

踩

java 处理canal 到mq中的数据

在这里插入图片描述

canal [kə’næl]，译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费

目录

文档资料

一、MySQL设置

修改 my.cnf 中配置

[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1       # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

  • 1
  • 2
  • 3
  • 4

授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant

CREATE USER canal IDENTIFIED BY 'canal';  

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;

FLUSH PRIVILEGES;

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

查看授权情况

select * from mysql.user where User = 'canal'\G

*************************** 1. row ***************************
                    Host: %
                    User: canal
             Select_priv: Y
         Repl_slave_priv: Y
        Repl_client_priv: Y

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

二、启动Canal服务端

下载解压

wget https://github.com/alibaba/canal/releases/download/canal-1.1.6/canal.deployer-1.1.6.tar.gz

  • 1

配置修改

conf/example/instance.properties

## mysql serverId
canal.instance.mysql.slaveId = 100

#position info，需要改成自己的数据库信息
canal.instance.master.address = 127.0.0.1:3306 

#username/password，需要改成自己的数据库信息
canal.instance.dbUsername = canal  
canal.instance.dbPassword = 123456

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9

启动服务

# 启动
bash bin/startup.sh

# 关闭
bash bin/stop.sh

# 重启
bash bin/restart.sh

# 查看日志
tail -f logs/canal/canal.log

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

三、通过Canal客户端消费数据

可以直接消费数据；也可以先放到MQ中，再消费数据

在这里插入图片描述

直接通过Canal客户端消费数据

依赖

<!-- canal -->
<dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.client</artifactId>
    <version>1.1.0</version>
</dependency>

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

客户端代码

package com.example.demo;

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 SimpleCanalClientExample {

    public static void run() {
        // 连接信息配置
        String hostname = "127.0.0.1";
        int port = 11111;
        String destination = "example";
        String username = "";
        String password = "";

        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(hostname, port), destination, username, password
        );

        int batchSize = 1000;

        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();

            while (true) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();

                // 没有拿到数据
                if (batchId == -1 || size == 0) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                    }
                } else {
                    System.out.printf("message[batchId=%s, size=%s] \n", batchId, size);
                    printEntry(message.getEntries());
                }

                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }
        } finally {
            connector.disconnect();
        }
    }

    private static void printEntry(List<Entry> entries) {
        for (Entry entry : entries) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            RowChange rowChange = null;
            try {
                rowChange = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }

            EventType eventType = rowChange.getEventType();
            System.out.println(String.format("binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(),
                    entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(),
                    entry.getHeader().getTableName(),
                    eventType));

            // 数据变化
            for (RowData rowData : rowChange.getRowDatasList()) {
                if (eventType == EventType.DELETE) {
                    printColumn(rowData.getBeforeColumnsList());
                } else if (eventType == EventType.INSERT) {
                    printColumn(rowData.getAfterColumnsList());
                } else {
                    printColumn(rowData.getAfterColumnsList());
                }
            }
        }
    }

    private static void printColumn(List<Column> columns) {
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue());
        }
    }

    public static void main(String[] args) {
        run();
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102

修改MySQL中的数据，就可以监听到

四、通过RabbitMQ消费数据

1、启动RabbitMQ

采用docker会简单很多

docker run \
 -e RABBITMQ_DEFAULT_USER=root \
 -e RABBITMQ_DEFAULT_PASS=123456 \
 --name mq \
 --hostname mq1 \
 -p 15672:15672 \
 -p 5672:5672 \
 -d \
 rabbitmq:3-management

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9

或者 docker-compose.yml

version: "3.7"

services:
  rabbitmq:
    image: rabbitmq:3-management
    container_name: rabbitmq
    hostname: rabbitmq
    restart: always
    ports:
      - "15672:15672"
      - "5672:5672"
    volumes:
      - ./app/rabbitmq/data/:/var/lib/rabbitmq/
      - ./app/rabbitmq/log/:/var/log/rabbitmq/
    environment:
      - RABBITMQ_DEFAULT_USER=root
      - RABBITMQ_DEFAULT_PASS=123456


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18

管理界面：http://localhost:15672/

2、修改canal配置

canal.properties

# tcp, kafka, rocketMQ, rabbitMQ, pulsarMQ
# canal.serverMode = tcp
canal.serverMode = rabbitMQ

##################################################
#########           RabbitMQ         #############
##################################################
rabbitmq.host = 127.0.0.1
rabbitmq.virtual.host = /
rabbitmq.exchange = canal-exchange
rabbitmq.username = root
rabbitmq.password = 123456

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12

instance.properties

# mq config
# canal.mq.topic=example
canal.mq.topic=canal-routing-key

  • 1
  • 2
  • 3

重启canal服务端

3、消费RabbitMQ中的数据

SpringBoot 项目结构

$ tree -I target
.
├── pom.xml
└── src
    ├── main
    │   ├── java
    │   │   └── com
    │   │       └── example
    │   │           └── demo
    │   │               ├── Application.java
    │   │               ├── config
    │   │               │   ├── CanalProvider.java
    │   │               │   └── RabbitConfig.java
    │   │               ├── constant
    │   │               │   ├── EventTypeConstant.java
    │   │               │   └── RabbitConstant.java
    │   │               ├── consumer
    │   │               │   ├── CanalConsumer.java
    │   │               │   └── SimpleCanalClientExample.java
    │   │               └── controller
    │   │                   └── IndexController.java
    │   └── resources
    │       ├── application.yml
    │       ├── static
    │       └── templates
    └── test
        └── java
            └── com
                └── example
                    └── demo
                        └── ApplicationTests.java



  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33

依赖 pom.xml

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

  • 1
  • 2
  • 3
  • 4

配置 application.yml


spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: root
    password: 123321
    # 消息确认配置项
    # 确认消息已发送到交换机(Exchange)
    publisher-confirm-type: correlated
    # 确认消息已发送到队列(Queue)
    publisher-returns: true


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13

RabbitMQ配置类 RabbitConfig.java

package com.example.demo.config;

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate();
        template.setConnectionFactory(connectionFactory);
        template.setMessageConverter(new Jackson2JsonMessageConverter());

        return template;
    }

    /**
     * template.setMessageConverter(new Jackson2JsonMessageConverter());
     * 这段和上面这行代码解决RabbitListener循环报错的问题
     */
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        return factory;
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32

Canal消息生产者：

package com.example.demo.config;

import com.example.demo.constant.RabbitConstant;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Canal消息提供者，canal-server生产的消息通过RabbitMQ消息队列发送
 */
@Configuration
public class CanalProvider {
    /**
     * 队列
     */
    @Bean
    public Queue canalQueue() {
        /**
         * durable:是否持久化，默认false，持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在；暂存队列：当前连接有效
         * exclusive:默认为false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
         * autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除
         */
        return new Queue(RabbitConstant.CanalQueue, true);
    }

    /**
     * 交换机，这里使用直连交换机
     */
    @Bean
    DirectExchange canalExchange() {
        return new DirectExchange(RabbitConstant.CanalExchange, true, false);
    }

    /**
     * 绑定交换机和队列，并设置匹配键
     */
    @Bean
    Binding bindingCanal() {
        return BindingBuilder.bind(canalQueue()).to(canalExchange()).with(RabbitConstant.CanalRouting);
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44

Canal消息消费者：

package com.example.demo.consumer;

import com.example.demo.constant.EventTypeConstant;
import com.example.demo.constant.RabbitConstant;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

/**
 * Canal消息消费者
 */
@Component
@RabbitListener(queues = RabbitConstant.CanalQueue)
public class CanalConsumer {
    @RabbitHandler
    public void process(Map<String, Object> msg) {
        System.out.println("收到canal消息：" + msg);
        // 收到canal消息：{data=[{id=21, name=Mini+1, age=30}], database=data, es=1684302821000, id=2, isDdl=false, mysqlType={id=int unsigned, name=varchar(20), age=tinyint unsigned}, old=[{name=Mini}], pkNames=[id], sql=, sqlType={id=4, name=12, age=-6}, table=tb_user, ts=1684302821452, type=UPDATE}

        boolean isDdl = (boolean) msg.get("isDdl");

        // 不处理DDL事件
        if (isDdl) {
            return;
        }

        // 数据库
        String database = (String) msg.get("database");
        // 表
        String table = (String) msg.get("table");
        // 类型：INSERT/UPDATE/DELETE
        String type = (String) msg.get("type");
        // 每一列的数据值
        List<?> data = (List<?>) msg.get("data");

        System.out.printf("%s.%s", database, table);

        // 只处理指定类型
        if (EventTypeConstant.INSERT.equalsIgnoreCase(type)) {
            System.out.println("INSERT");
        } else if (EventTypeConstant.UPDATE.equalsIgnoreCase(type)) {
            System.out.println("UPDATE");
        } else if (EventTypeConstant.DELETE.equalsIgnoreCase(type)) {
            System.out.println("DELETE");
        } else {
            // 其他事件
        }
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52

数据变化事件类型

package com.example.demo.constant;

/**
 * 数据变化事件类型
 */
public class EventTypeConstant {
    public static final String INSERT = "INSERT";
    public static final String UPDATE = "UPDATE";
    public static final String DELETE = "DELETE";
}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

配置

package com.example.demo.constant;

/**
 * 配置
 */
public class RabbitConstant {
    public static final String CanalQueue = "canal-queue";
    public static final String CanalExchange = "canal-exchange";
    public static final String CanalRouting = "canal-routing-key";
}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

参考
Canal实战，一套带走
SpringBoot整合 Canal、RabbitMQ 监听数据变更-微信
SpringBoot整合 Canal、RabbitMQ 监听数据变更-掘金

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/很楠不爱3/article/detail/445317
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号