在线时间:8:00-16:00
迪恩网络APP
随时随地掌握行业动态
扫描二维码
关注迪恩网络微信公众号
前言 我们经常需要根据用户对自己数据的一些操作来做一些事情. 比如如果用户删除了自己的账号,我们就给他发短信骂他,去发短信求他回来. 类似于这种功能,当然可以在业务逻辑层实现,在收到用户的删除请求之后执行这一操作,但是数据库的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%' ; 如果结果如下图,则说明成功了:
查看正在写入的binlog状态:
代码读取binlog 引入依赖 我们使用开源的一些实现,这里因为一些奇怪的原因,我选用了mysql-binlog-connector-java这个包,(官方github仓库)[github.com/shyiko/mysq…]具体依赖如下: <!-- 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如下:
根据自己的业务,封装一个更好使,更定制的工具类 开始的时候打算贴代码的,,,但是代码越写越多,索性传在github上了,这里只贴部分的实现.代码传送门 实现思路
所以实现思路大致如下:
初始化代码: 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表插入数据时,可以独立的运行定义的业务逻辑. 注意:这里的工具类并不能直接投入使用,因为里面有许多的异常处理没有做,且功能仅监听了插入语句,可以用来做实现的参考. 参考文章 总结 以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作具有一定的参考学习价值,谢谢大家对极客世界的支持。 |
请发表评论