golang爬虫 Netty session 小程序demo源码 时间戳java git下载代码到本地命令 java手机验证码 后台管理网站模板 python定义变量 python模块下载 java基本语法 java使用正则表达式 java方法 javalist java中instanceof java的输入 java连接数据库代码 linux入门基础 linux安装系统 linux装机 php项目实例 linux内核编程 脚本软件 shutil h370主板 oem修改器 collect 摩斯密码在线翻译 模拟按键 t470拆机 电脑代码雨 cad自动保存位置 win98序列号 globalscape qq空间自动点赞 HTA mysql关联查询 小米自动开关机 微信昵称特殊符号 c4d克隆
当前位置: 首页 > 学习教程  > 编程语言

什么?通过maxwell读取binlog日志,把mysql变化数据传入redis!

2020/9/19 15:50:51 文章标签:

什么?通过maxwell读取binlog日志,把mysql变化数据传入redis!

 

Maxwell简介

Maxwell是一个能实时读取MySQL二进制日志binlog,并生成 JSON 格式的消息,作为生产者发送给 Kafka,Kinesis、RabbitMQ、Redis、Google Cloud Pub/Sub、文件或其它平台的应用程序。它的常见应用场景有ETL、维护缓存、收集表级别的dml指标、增量到搜索引擎、数据分区迁移、切库binlog回滚方案等。Maxwell给出了一些无需重新构建整个平台的事件来源的好处。大家可以通过官网下载合适的版本进行使用。Maxwell主要提供了下列功能:

  • 支持 SELECT * FROM table 的方式进行全量数据初始化
  • 支持在主库发生failover后,自动恢复binlog位置(GTID)。
  • 可以对数据进行分区,解决数据倾斜问题,发送到kafka的数据支持database、table、column等级别的数据分区。
  • 工作方式是伪装为Slave,接收binlog events,然后根据schemas信息拼装,可以接受ddl、xid、row等各种event。

MaxWell安装

Maxwell安装相对比较简单,本次主要是修改maxwell的配置文件。1.上传maxwell并解压到指定目录使用linux连接服务器工具,把maxwell.1.24.1.tar.gz上传到/soft目录下。[root@localhost bin]# tar -xvf maxwell.1.24.1.tar.gz -C /opt/maxwell

在把maxwell安装完成后,再在mysql数据库中配置maxwell用户和库。

Mysql服务配置(my.conf)

在/etc/my.conf添加以下内容,在安装mysql时,已在my.cnf文件中添加了相应内容。

[mysqld]
server_id=23
log-bin=bin-log
binlog_format=row  

解释:MySQL必须开启了binlogs,即log-bin指定了目录binlog_format必须是rowserver_id指定mysql的全局唯一id

在修改mysql conf后,需要重启mysql服务

[root@localhost ~]#systemctl stop mysql
[root@localhost ~]#systemctl start mysql

或者[root@localhost ~]#service mysqld restart

Maxwell用户权限配置

Maxwell需要储存它自己的一些状态数据,启动参数schema_database选型来指定,默认是maxwell。MySQL 用户及权限配置(SQL) 创建maxwell用户,并设置密码为’123456’。mysql> CREATE USER 'maxwell'@'%' IDENTIFIED BY '123456';

创建maxwell数据库,存储maxwell工具的一些状态数据。mysql> CREATE DATABASE IF NOT EXISTS maxwell default charset utf8 COLLATE utf8_general_ci;

对maxwell用户进行授权。

mysql> GRANT ALL on *.* to 'maxwell'@'%' identified by 'XXXXXX';
mysql> GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'%'; 

Maxwell的配置文件

  1. 在/opt/maxwell目录下创建一个config.properties文件,写入指定配置:
    [root@localhost maxwell]#vim config.properties
#[mysql]
user=maxwell    #连接mysql用户名password=123456  #连接mysql的密码
host=192.168.1.22   # mysql的主机名(IP地址)
port=3306   #mysql端口
#[producer]
producer=redisredis_host=127.0.0.1  
#redis服务器ip地址redis_port=6379         
#redis的端口,默认是6379
redis_database =0       
#redis中数据库,默认为0
rredis_key=maxwellredis_stream_json_key=message redis_type=pubsub

主要参数解释:

#[mysql]下的参数主要是连接mysql配置信息,填写上述创建的maxwell用户。

#[producer]下的参数是生产者相关信息

producer :生产者类型,可是kafka、redis等,本次使用

redisredis_XX : redis相关配置信息,其中redis_host本文填写的为127.0.0.1,是因为在redis配置文件中,bind配置的为127.0.0.1,如果在config.properties填写实际ip地址,maxwell会无法访问。

redis_type :选择redis的数据生成模式,目前支持[ pubsub | xadd | lpush | rpush ],默认值为pubsub。

通过以下命令启动maxwell

[root@localhost ~]#/opt/maxwell/bin/maxwell --config config.properties --filter=exclude:*.*,include:lipp.* --daemon

参数说明:

config:指定配置文件,一般在参数较多时,需要把相关配置写到配置文件中。

filter:过滤设置,可以通过exclude和include关键字设置排除和包含哪些数据。

daemon:指定后台运行

测试发布/订阅模式

通过上一步配置,设置的redis_type=pubsub,此模式表示创建一个发布主题,当mysql数据库发生变化时,变化的内容将被maxwell转入到redis中发布到maxwell频道,当所有订阅了此频道的订阅者都将会收到相应变化消息。

通过以上步骤,启动mysql、redis和maxwell后,开始进行简单测试1.在client1 通过redis_cli登录到redis并订阅maxwell

[root@client1 ~]# redis-cli 
127.0.0.1:6379> SUBSCRIBE maxwell
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "maxwell"
3) (integer) 1

2.在client2中登录mysql并向lipp数据库中表增加数据。

[root@client2 ~]# mysql -uroot -p111111
mysql: [Warning] Using a password on the command line interface can be insecure.
Welcome to the MySQL monitor.  Commands end with ; or \g.
Your MySQL connection id is 300
Server version: 5.7.23-log MySQL Community Server (GPL)
Copyright (c) 2000, 2018, Oracle and/or its affiliates. All rights reserved.
Oracle is a registered trademark of Oracle Corporation and/or its
affiliates. Other names may be trademarks of their respectiveowners.Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.
mysql> use lipp;Reading table information for completion of table and column names
You can turn off this feature to get a quicker startup with -A
Database changedmysql> insert into t1 values(1,'test101');

3.在client1可以看到maxwell channel的输出

1) "message"
2) "maxwell"
3) "{\"database\":\"lipp\",\"table\":\"t1\",\"type\":\"insert\",\"ts\":1584279262,\"xid\":7726,\"commit\":true,\"data\":{\"id\":1,\"name\":\"test101\"}}"

测试list模式

1.修改maxwell目录下的conf.preporties文件中redis_type=lpush,后重启maxwell服务。

maxwell                          RUNNING   pid 11366, uptime 0:00:08

2.在client1 通过redis_cli登录redis中,并查看当前库中有多少key

[root@client1 ~]# redis-cli 
127.0.0.1:6379> DBSIZE
(integer) 0

目前0号库中没有key。

3.在client2 登录mysql数据库中,并在指定表插入数据。

mysql> insert into t1 values(1,'wangwu');
Query OK, 1 row affected (0.00 sec)

4.在client1中通过dbsize再次查看数据库大小

127.0.0.1:6379> DBSIZE
(integer) 1

通过keys 查看key名称,并通过type查看key的类型。

127.0.0.1:6379> keys *
1) "maxwell"
127.0.0.1:6379> type maxwell
list

当key的类型问list时,可以使用list相关命令进行对key操作。通过llen查看key的长度

127.0.0.1:6379> llen maxwell
(integer) 1

通过lrange命令查看key的内容

127.0.0.1:6379> LRANGE maxwell 0 10
1)"{\"database\":\"lipp\",\"table\":\"t1\",\"type\":\"insert\",\"ts\":1584286244,\"xid\":15381,\"commit\":true,\"data\":{\"id\":1,\"name\":\"wangwu\"}}"

通过lpop或rpop 弹出key中的值

127.0.0.1:6379> LPOP maxwell
"{\"database\":\"lipp\",\"table\":\"t1\",\"type\":\"insert\",\"ts\":1584286244,\"xid\":15381,\"commit\":true,\"data\":{\"id\":1,\"name\":\"wangwu\"}}"

通过修改redis_type的参数为pubsub和lpush,可以实现监控mysql数据库变化,通过发布订阅模式,实现数据同步功能。通过list方式可以获取最新的数据的变化和数据变化数量等需求。

本文只是演示了maxwell读取binlog到redis,其实maxwell可以实现多种producer方式,如kafka,pubsub、redis、自定义等。具体可以通过官网了解,也可以到github了解。

作者:ww51pp

链接:https://blog.51cto.com/jxplpp/2480229

转载自51cto


本文链接: http://www.dtmao.cc/news_show_200330.shtml

附件下载

相关教程

    暂无相关的数据...

共有条评论 网友评论

验证码: 看不清楚?