欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

Mysql数据库监听binlog的开启步骤

程序员文章站 2022-04-06 09:52:25
前言 我们经常需要根据用户对自己数据的一些操作来做一些事情. 比如如果用户删除了自己的账号,我们就给他发短信骂他,去发短信求他回来. 类似于这种功能,当然...

前言

我们经常需要根据用户对自己数据的一些操作来做一些事情.

比如如果用户删除了自己的账号,我们就给他发短信骂他,去发短信求他回来.

类似于这种功能,当然可以在业务逻辑层实现,在收到用户的删除请求之后执行这一操作,但是数据库的binlog为我们提供了另外一种操作方法.

要监听binlog,需要两步,第一步当然是你的mysql需要开启这一个功能,第二个是要写程序来对日志进行读取.

mysql开启binlog.

首先mysql的binlog日常是不打开的,因此我们需要:

找到mysql的配置文件my.cnf,这个因操作系统不一样,位置也不一定一样,可以自己找一下,

在其中加入以下内容:

[mysqld]
server_id = 1
log-bin = mysql-bin
binlog-format = row

之后重启mysql.

/ ubuntu
service mysql restart
// mac
mysql.server restart

监测是否开启成功

进入mysql命令行,执行:

show variables like '%log_bin%' ;

如果结果如下图,则说明成功了:

Mysql数据库监听binlog的开启步骤

查看正在写入的binlog状态:

Mysql数据库监听binlog的开启步骤

代码读取binlog

引入依赖

我们使用开源的一些实现,这里因为一些奇怪的原因,我选用了mysql-binlog-connector-java这个包,(官方github仓库)[…]具体依赖如下:

<!-- https://mvnrepository.com/artifact/com.github.shyiko/mysql-binlog-connector-java -->
 <dependency>
 <groupid>com.github.shyiko</groupid>
 <artifactid>mysql-binlog-connector-java</artifactid>
 <version>0.17.0</version>
 </dependency>

当然,对binlog的处理有很多开源实现,阿里的cancl就是一个,也可以使用它.

写个demo

根据官方仓库中readme里面,来简单的写个demo.

 public static void main(string[] args) {
 binarylogclient client = new binarylogclient("hostname", 3306, "username", "passwd");
 eventdeserializer eventdeserializer = new eventdeserializer();
 eventdeserializer.setcompatibilitymode(
 eventdeserializer.compatibilitymode.date_and_time_as_long,
 eventdeserializer.compatibilitymode.char_and_binary_as_byte_array
 );
 client.seteventdeserializer(eventdeserializer);
 client.registereventlistener(new binarylogclient.eventlistener() {

 @override
 public void onevent(event event) {
 // todo
 dosomething();
 logger.info(event.tostring());
 }
 });
 client.connect();
 }

这个完全是根据官方教程里面写的,在onevent里面可以写自己的业务逻辑,由于我只是测试,所以我在里面将每一个event都打印了出来.

之后我手动登录到mysql,分别进行了增加,修改,删除操作,监听到的log如下:

00:23:13.331 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=0, eventtype=rotate, serverid=1, headerlength=19, datalength=28, nextposition=0, flags=32}, data=rotateeventdata{binlogfilename='mysql-bin.000001', binlogposition=886}}
00:23:13.334 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=1556468403000, eventtype=format_description, serverid=1, headerlength=19, datalength=100, nextposition=0, flags=0}, data=formatdescriptioneventdata{binlogversion=4, serverversion='5.7.23-0ubuntu0.16.04.1-log', headerlength=19, datalength=95}}
00:23:23.715 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=1556468603000, eventtype=anonymous_gtid, serverid=1, headerlength=19, datalength=46, nextposition=951, flags=0}, data=null}
00:23:23.716 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=1556468603000, eventtype=query, serverid=1, headerlength=19, datalength=51, nextposition=1021, flags=8}, data=queryeventdata{threadid=4, executiontime=0, errorcode=0, database='pf', sql='begin'}}
00:23:23.721 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=1556468603000, eventtype=table_map, serverid=1, headerlength=19, datalength=32, nextposition=1072, flags=0}, data=tablemapeventdata{tableid=108, database='pf', table='student', columntypes=15, 3, columnmetadata=135, 0, columnnullability={}}}
00:23:23.724 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=1556468603000, eventtype=ext_write_rows, serverid=1, headerlength=19, datalength=23, nextposition=1114, flags=0}, data=writerowseventdata{tableid=108, includedcolumns={0, 1}, rows=[
    [[b@546a03af, 2]
]}}
00:23:23.725 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=1556468603000, eventtype=xid, serverid=1, headerlength=19, datalength=12, nextposition=1145, flags=0}, data=xideventdata{xid=28}}
00:23:55.872 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=1556468635000, eventtype=anonymous_gtid, serverid=1, headerlength=19, datalength=46, nextposition=1210, flags=0}, data=null}
00:23:55.872 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=1556468635000, eventtype=query, serverid=1, headerlength=19, datalength=51, nextposition=1280, flags=8}, data=queryeventdata{threadid=4, executiontime=0, errorcode=0, database='pf', sql='begin'}}
00:23:55.873 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=1556468635000, eventtype=table_map, serverid=1, headerlength=19, datalength=32, nextposition=1331, flags=0}, data=tablemapeventdata{tableid=108, database='pf', table='student', columntypes=15, 3, columnmetadata=135, 0, columnnullability={}}}
00:23:55.875 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=1556468635000, eventtype=ext_update_rows, serverid=1, headerlength=19, datalength=31, nextposition=1381, flags=0}, data=updaterowseventdata{tableid=108, includedcolumnsbeforeupdate={0, 1}, includedcolumns={0, 1}, rows=[
    {before=[[b@6833ce2c, 1], after=[[b@725bef66, 3]}
]}}
00:23:55.875 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=1556468635000, eventtype=xid, serverid=1, headerlength=19, datalength=12, nextposition=1412, flags=0}, data=xideventdata{xid=41}}
00:24:22.333 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=1556468662000, eventtype=anonymous_gtid, serverid=1, headerlength=19, datalength=46, nextposition=1477, flags=0}, data=null}
00:24:22.334 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=1556468662000, eventtype=query, serverid=1, headerlength=19, datalength=51, nextposition=1547, flags=8}, data=queryeventdata{threadid=4, executiontime=0, errorcode=0, database='pf', sql='begin'}}
00:24:22.334 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=1556468662000, eventtype=table_map, serverid=1, headerlength=19, datalength=32, nextposition=1598, flags=0}, data=tablemapeventdata{tableid=108, database='pf', table='student', columntypes=15, 3, columnmetadata=135, 0, columnnullability={}}}
00:24:22.335 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=1556468662000, eventtype=ext_delete_rows, serverid=1, headerlength=19, datalength=23, nextposition=1640, flags=0}, data=deleterowseventdata{tableid=108, includedcolumns={0, 1}, rows=[
    [[b@1888ff2c, 3]
]}}
00:24:22.335 [main] info util.mysqlbinlog - event{header=eventheaderv4{timestamp=1556468662000, eventtype=xid, serverid=1, headerlength=19, datalength=12, nextposition=1671, flags=0}, data=xideventdata{xid=42}}

根据自己的业务,封装一个更好使,更定制的工具类

开始的时候打算贴代码的,,,但是代码越写越多,索性传在github上了,这里只贴部分的实现.代码传送门

实现思路

  1. 支持对单个表的监听,因为我们不想真的对所有数据库中的所有数据表进行监听.
  2. 可以多线程消费.
  3. 把监听到的内容转换成我们喜闻乐见的形式(文中的数据结构不一定很好,我没想到更加合适的了).

所以实现思路大致如下:

  1. 封装个客户端,对外只提供获取方法,屏蔽掉初始化的细节代码.
  2. 提供注册监听器(伪)的方法,可以注册对某个表的监听(重新定义一个监听接口,所有注册的监听器实现这个就好).
  3. 真正的监听器只有客户端,他将此数据库实例上的所有操作,全部监听到并转换成我们想要的格式logitem放进阻塞队列里面.
  4. 启动多个线程,消费阻塞队列,对某一个logitem调用对应的数据表的监听器,做一些业务逻辑.

初始化代码:

 public mysqlbinloglistener(conf conf) {
 binarylogclient client = new binarylogclient(conf.host, conf.port, conf.username, conf.passwd);
 eventdeserializer eventdeserializer = new eventdeserializer();
 eventdeserializer.setcompatibilitymode(
 eventdeserializer.compatibilitymode.date_and_time_as_long,
 eventdeserializer.compatibilitymode.char_and_binary_as_byte_array
 );
 client.seteventdeserializer(eventdeserializer);
 this.parseclient = client;
 this.queue = new arrayblockingqueue<>(1024);
 this.conf = conf;
 listeners = new concurrenthashmap<>();
 dbtablecols = new concurrenthashmap<>();
 this.consumer = executors.newfixedthreadpool(consumerthreads);
 }

注册代码:

 public void reglistener(string db, string table, binloglistener listener) throws exception {
 string dbtable = getdbtable(db, table);
 class.forname("com.mysql.jdbc.driver");
 // 保存当前注册的表的colum信息
 connection connection = drivermanager.getconnection("jdbc:mysql://" + conf.host + ":" + conf.port, conf.username, conf.passwd);
 map<string, colum> cols = getcolmap(connection, db, table);
 dbtablecols.put(dbtable, cols);

 // 保存当前注册的listener
 list<binloglistener> list = listeners.getordefault(dbtable, new arraylist<>());
 list.add(listener);
 listeners.put(dbtable, list);
 }

在这个步骤中,我们在注册监听者的同时,获得了该表的schema信息,并保存到map里面去,方便后续对数据进行处理.

监听代码:

 @override
 public void onevent(event event) {
 eventtype eventtype = event.getheader().geteventtype();

 if (eventtype == eventtype.table_map) {
 tablemapeventdata tabledata = event.getdata();
 string db = tabledata.getdatabase();
 string table = tabledata.gettable();
 dbtable = getdbtable(db, table);
 }

 // 只处理添加删除更新三种操作
 if (iswrite(eventtype) || isupdate(eventtype) || isdelete(eventtype)) {
 if (iswrite(eventtype)) {
 writerowseventdata data = event.getdata();
 for (serializable[] row : data.getrows()) {
  if (dbtablecols.containskey(dbtable)) {
  logitem e = logitem.itemfrominsert(row, dbtablecols.get(dbtable));
  e.setdbtable(dbtable);
  queue.add(e);
  }
 }
 }
 }
 }

我偷懒了,,,这里面只实现了对添加操作的处理,其他操作没有写.

消费代码:

 public void parse() throws ioexception {
 parseclient.registereventlistener(this);

 for (int i = 0; i < consumerthreads; i++) {
 consumer.submit(() -> {
 while (true) {
  if (queue.size() > 0) {
  try {
  logitem item = queue.take();
  string dbtable = item.getdbtable();
  listeners.get(dbtable).foreach(l -> {
  l.onevent(item);
  });

  } catch (interruptedexception e) {
  e.printstacktrace();
  }
  }
  thread.sleep(1000);
 }
 });
 }
 parseclient.connect();
 }

消费时,从队列中获取item,之后获取对应的一个或者多个监听者,分别消费这个item.

测试代码:

 public static void main(string[] args) throws exception {
 conf conf = new conf();
 conf.host = "hostname";
 conf.port = 3306;
 conf.username = conf.passwd = "hhsgsb";

 mysqlbinloglistener mysqlbinloglistener = new mysqlbinloglistener(conf);
 mysqlbinloglistener.parseargsandrun(args);
 mysqlbinloglistener.reglistener("pf", "student", item -> {
 system.out.println(new string((byte[])item.getafter().get("name")));
 logger.info("insert into {}, value = {}", item.getdbtable(), item.getafter());
 });
 mysqlbinloglistener.reglistener("pf", "teacher", item -> system.out.println("teacher ===="));

 mysqlbinloglistener.parse();
 }

在这段很少的代码里,注册了两个监听者,分别监听student和teacher表,并分别进行打印处理,经测试,在teacher表插入数据时,可以独立的运行定义的业务逻辑.

注意:这里的工具类并不能直接投入使用,因为里面有许多的异常处理没有做,且功能仅监听了插入语句,可以用来做实现的参考.

参考文章

总结

以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作具有一定的参考学习价值,谢谢大家对的支持。