第0章：简介

(1)下面是http://code.google.com中的binlog事件分析结构图：

(2)获取开源包的maven坐标

com.google.code

open-replicator

1.0.5

(3)参考网站

(4)札记

1)Open Replicator是一个用Java编写的MySQL binlog分析程序。Open Replicator 首先连接到MySQL(就像一个普通的MySQL Slave一样)，然后接收和分析binlog，最终将分析得出的binlog events以回调的方式通知应用。

2)Open Replicator可以被应用到MySQL数据变化的实时推送，多Master到单Slave的数据同步等多种应用场景。

3)在程序结构上，最主要的设计原则是高性能，低内存占用。Open Replicator目前只支持MySQL5.0及以上版本。

第1章：实践

(1)包装Open Replicator类(AutoOpenReplicator.java)

package com.mcc.core.openReplicator;

import com.google.code.or.OpenReplicator;

import com.google.code.or.common.glossary.column.StringColumn;

import com.google.code.or.net.Packet;

import com.google.code.or.net.Transport;

import com.google.code.or.net.impl.packet.EOFPacket;

import com.google.code.or.net.impl.packet.ResultSetRowPacket;

import com.google.code.or.net.impl.packet.command.ComQuery;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import java.util.List;

import java.util.concurrent.TimeUnit;

/**

* MySQL binlog分析程序 ,用到open-replicator包

* 增加加自动配置binlog位置及重连机制

*

* @author menergy

*         DateTime: 13-12-26  下午2:22

*/

public class AutoOpenReplicator extends OpenReplicator {

// members

private static Logger logger = LoggerFactory.getLogger(AutoOpenReplicator.class);

private boolean autoReconnect = true;

// timeout auto reconnect , default 30 second

private int delayReconnect = 30;

// default timeout is 60 second, after timeout will be reconnect!

private int defaultTimeout = 120 * 1000;

// COM Query Transport

private Transport comQueryTransport;

// static block

// constructors

// properties

/**

* 是否自动重连

*

* @return 自动重连

*/

public boolean isAutoReconnect() {

return autoReconnect;

}

/**

* 设置自动重连

*

* @param autoReconnect 自动重连

*/

public void setAutoReconnect(boolean autoReconnect) {

this.autoReconnect = autoReconnect;

}

/**

* 断开多少秒后进行自动重连

*

* @param delayReconnect 断开后多少秒

*/

public void setDelayReconnect(int delayReconnect) {

this.delayReconnect = delayReconnect;

}

/**

* 断开多少秒后进行自动重连

*

* @return 断开后多少秒

*/

public int getDelayReconnect() {

return delayReconnect;

}

// public methods

// protected methods

@Override

public void start() {

do {

try {

long current = System.currentTimeMillis();

if (!this.isRunning()) {

if (this.getBinlogFileName() == null) updatePosition();

logger.info("Try to startup dump binlog from mysql master[{}, {}] ...", this.binlogFileName, this.binlogPosition);

this.reset();

super.start();

logger.info("Startup successed! After {} second if nothing event fire will be reconnect ...", defaultTimeout / 1000);

} else {

if (current - this.lastAlive >= this.defaultTimeout) {

this.stopQuietly(0, TimeUnit.SECONDS);

}

}

TimeUnit.SECONDS.sleep(this.getDelayReconnect());

} catch (Exception e) {

if (logger.isErrorEnabled()) {

logger.error("connect mysql failure!", e);

}

// reconnect failure, reget last binlog & position from master node and update cache!

//LoadCenter.loadAll(); // just update all cache, not flush!

updatePosition();

try {

TimeUnit.SECONDS.sleep(this.getDelayReconnect());

} catch (InterruptedException ignore) {

// NOP

}

}

} while (this.autoReconnect);

}

@Override

public void stopQuietly(long timeout, TimeUnit unit) {

super.stopQuietly(timeout, unit);

if (this.getBinlogParser() != null) {

// 重置, 当MySQL服务器进行restart/stop操作时进入该流程

this.binlogParser.setParserListeners(null); // 这句比较关键，不然会死循环

}

}

// friendly methods

// private methods

/**

* 自动配置binlog位置

*/

private void updatePosition() {

// 配置binlog位置

try {

ResultSetRowPacket binlogPacket = query("show master status");

if (binlogPacket != null) {

List values = binlogPacket.getColumns();

this.setBinlogFileName(values.get(0).toString());

this.setBinlogPosition(Long.valueOf(values.get(1).toString()));

}

} catch (Exception e) {

if (logger.isErrorEnabled()) {

logger.error("update binlog position failure!", e);

}

}

}

/**

* ComQuery 查询

*

* @param sql 查询语句

* @return

*/

private ResultSetRowPacket query(String sql) throws Exception {

ResultSetRowPacket row = null;

final ComQuery command = new ComQuery();

command.setSql(StringColumn.valueOf(sql.getBytes()));

if (this.comQueryTransport == null) this.comQueryTransport = getDefaultTransport();

this.comQueryTransport.connect(this.host, this.port);

this.comQueryTransport.getOutputStream().writePacket(command);

this.comQueryTransport.getOutputStream().flush();

// step 1

this.comQueryTransport.getInputStream().readPacket();

//

Packet packet;

// step 2

while (true) {

packet = comQueryTransport.getInputStream().readPacket();

if (packet.getPacketBody()[0] == EOFPacket.PACKET_MARKER) {

break;

}

}

// step 3

while (true) {

packet = comQueryTransport.getInputStream().readPacket();

if (packet.getPacketBody()[0] == EOFPacket.PACKET_MARKER) {

break;

} else {

row = ResultSetRowPacket.valueOf(packet);

}

}

this.comQueryTransport.disconnect();

return row;

}

private void reset() {

this.transport = null;

this.binlogParser = null;

}

// inner class

// test main

}

(2)事件监听器模板类(NotificationListener.java)

package com.mcc.core.openReplicator;

import com.google.code.or.binlog.BinlogEventListener;

import com.google.code.or.binlog.BinlogEventV4;

import com.google.code.or.binlog.impl.event.*;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

/**

* Binlog事件监听器模板

*

* @author menergy

*         DateTime: 13-12-26  下午2:34

*/

public class NotificationListener implements BinlogEventListener {

private static Logger logger = LoggerFactory.getLogger(NotificationListener.class);

private String eventDatabase;

/**

* 这里只是实现例子，该方法可以自由处理逻辑

* @param event

*/

@Override

public void onEvents(BinlogEventV4 event) {

Class> eventType = event.getClass();

// 事务开始

if (eventType == QueryEvent.class) {

QueryEvent actualEvent = (QueryEvent) event;

this.eventDatabase = actualEvent.getDatabaseName().toString();

//TODO,这里可以获取事件数据库信息,可做其它逻辑处理

logger.info("事件数据库名：{}",eventDatabase);

return;

}

// 只监控指定数据库

if (eventDatabase != null && !"".equals(eventDatabase.trim())) {

if (eventType == TableMapEvent.class) {

TableMapEvent actualEvent = (TableMapEvent) event;

long tableId = actualEvent.getTableId();

String tableName = actualEvent.getTableName().toString();

//TODO,这里可以获取事件表信息,可做其它逻辑处理

logger.info("事件数据表ID：{}， 事件数据库表名称：{}",tableId, tableName);

} else if (eventType == WriteRowsEvent.class) { // 插入事件

WriteRowsEvent actualEvent = (WriteRowsEvent) event;

long tableId = actualEvent.getTableId();

//TODO,这里可以获取写行事件信息,可做其它逻辑处理

logger.info("写行事件ID：{}",tableId);

} else if (eventType == UpdateRowsEvent.class) { // 更新事件

UpdateRowsEvent actualEvent = (UpdateRowsEvent) event;

long tableId = actualEvent.getTableId();

//TODO,这里可以获取更新事件信息,可做其它逻辑处理

logger.info("更新事件ID：{}",tableId);

} else if (eventType == DeleteRowsEvent.class) {// 删除事件

DeleteRowsEvent actualEvent = (DeleteRowsEvent) event;

long tableId = actualEvent.getTableId();

//TODO,这里可以获取删除事件信息,可做其它逻辑处理

logger.info("删除事件ID：{}",tableId);

} else if (eventType == XidEvent.class) {// 结束事务

XidEvent actualEvent = (XidEvent) event;

long xId = actualEvent.getXid();

//TODO,这里可以获取结束事件信息,可做其它逻辑处理

logger.info("结束事件ID：{}",xId);

}

}

}

}

(3)MySQL binlog分析程序测试类(OpenReplicatorTest.java)

package com.mcc.core.test; import com.mcc.core.openReplicator.AutoOpenReplicator; import com.mcc.core.openReplicator.NotificationListener; /**  * MySQL binlog分析程序测试  *  * @author menergy  *         DateTime: 13-12-26  下午2:26  */ public class OpenReplicatorTest {     public static void main(String args[]){         // 配置从MySQL Master进行复制         final AutoOpenReplicator aor = new AutoOpenReplicator();         aor.setServerId(100001);         aor.setHost("192.168.1.1");         aor.setUser("admin");         aor.setPassword("123456");         aor.setAutoReconnect(true);         aor.setDelayReconnect(5);         aor.setBinlogEventListener(new NotificationListener());         aor.start();     } }