如何使用Maxwell实时同步mysql数据

吾爱主题阅读：167 2024-04-02 08:01:24 评论：0

maxwell简介

maxwell是由java编写的守护进程，可以实时读取mysql binlog并将行更新以json格式写入kafka、rabbitmq、redis等中，这样有了mysql增量数据流，使用场景就很多了，比如：实时同步数据到缓存，同步数据到elasticsearch，数据迁移等等。

maxwell官网：http://maxwells-daemon.io
maxwell源代码：https://github.com/zendesk/maxwell

maxwell的配置与使用

maxwell 依赖java sdk，所以需要先配置jdk环境。

1.下载maxwell安装包

1 2 3 4 5 root@xxx maxwell]# pwd /usr/ local /maxwell [root@xxx maxwell]# wget https://github.com/zendesk/maxwell/releases/download/v1.19.5/maxwell-1.19.5.tar.gz [root@xxx maxwell]# tar zxvf maxwell-1.19.5.tar.gz [root@xxx maxwell]# cd maxwell-1.19.5

2.配置mysql，打开mysql binlog日志

1 2 3 4 5 6 [root@xxx mysql]# vi /usr/ local /mysql/my.cnf [mysqld] log-bin=mysql-bin #添加这一行就 binlog-format=row #选择row模式 server_id=1 #随机指定一个不能和其他集群中机器重名的字符串，如果只有一台机器，那就可以随便指定了

重启mysql服务，登陆mysql，查看binlog日志模式

1 2 3 4 5 6 7 8 9 10 11 12 mysql> show variables like '%log_bin%' + ---------------------------------+-------------------------------+ | variable_name | value | + ---------------------------------+-------------------------------+ | log_bin | on | | log_bin_basename | /data/mysqldb/mysql-bin | | log_bin_index | /data/mysqldb/mysql-bin. index | | log_bin_trust_function_creators | off | | log_bin_use_v1_row_events | off | | sql_log_bin | on | + ---------------------------------+-------------------------------+ 6 rows in set (0.11 sec)

maxwell需要在schema_database选项指定的数据库中存储状态的权限（默认库名称为maxwell），所以需要提前给权限：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 #创建一个有同步数据的用户yhrepl mysql> create user 'yhrepl' @ '*' identified by 'scgaopan' ; query ok, 0 rows affected (0.10 sec) #此用户yhrepl要有对需要同步的数据库表有操作权限 mysql> grant all privileges on test.* to 'yhrepl' @ '%' identified by 'scgaopan' ; query ok, 0 rows affected (0.13 sec) #给yhrepl有同步数据的权限 mysql> grant select ,replication client,replication slave on *.* to 'yhrepl' @ '%' identified by 'scgaopan' ; query ok, 0 rows affected (0.10 sec) # maxwell需要在schema_database选项指定的数据库中存储状态的权限（默认库名称为maxwell） mysql> grant all privileges on maxwell.* to 'yhrepl' @ '%' identified by 'scgaopan' ; query ok, 0 rows affected (0.09 sec)

3.启动maxwell，主要介绍数据写入rabbitmq的实战：

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 [root@xxx maxwell-1.19.5]# vi /usr/ local /maxwell/maxwell-1.19.5/config.properties #日志级别 log_level=debug producer=rabbitmq daemon= true #监控的数据库, mysql用户必须拥有读取binlog权限和新建库表的权限 host=47.105.110.xxx user =yhrepl password =scgaopan output_nulls= true jdbc_options=autoreconnet= true #监控数据库中的哪些表 filter=exclude: *.*,include: test.aa #replica_server_id 和 client_id 唯一标示，用于集群部署 replica_server_id=64 client_id=test-id #metrics_type=http #metrics_slf4j_interval=60 #http_port=8111 #http_diagnostic= true # default false #rabbitmq rabbitmq_host=47.105.110.xxx rabbitmq_port=5672 rabbitmq_user=guest rabbitmq_pass=guest rabbitmq_virtual_host=/ rabbitmq_exchange=maxwell rabbitmq_exchange_type=topic rabbitmq_exchange_durable= false rabbitmq_exchange_autodelete= false rabbitmq_routing_key_template=%db%.% table % rabbitmq_message_persistent= false rabbitmq_declare_exchange= true