序
本文主要研究一下 rocketmq-mysql 的 EventProcessor
EventProcessor
rocketmq-externals/rocketmq-mysql/src/main/java/org/apache/rocketmq/mysql/binlog/EventProcessor.java
public class EventProcessor {private static final Logger LOGGER = LoggerFactory.getLogger(EventProcessor.class);
private Replicator replicator;
private Config config;
private DataSource dataSource;
private BinlogPositionManager binlogPositionManager;
private BlockingQueue<Event> queue = new LinkedBlockingQueue<>(100);
private BinaryLogClient binaryLogClient;
private EventListener eventListener;
private Schema schema;
private Map<Long, Table> tableMap = new HashMap<>();
private Transaction transaction;
public EventProcessor(Replicator replicator) {
this.replicator = replicator;
this.config = replicator.getConfig();}
public void start() throws Exception {initDataSource();
binlogPositionManager = new BinlogPositionManager(config, dataSource);
binlogPositionManager.initBeginPosition();
schema = new Schema(dataSource);
schema.load();
eventListener = new EventListener(queue);
binaryLogClient = new BinaryLogClient(config.mysqlAddr,
config.mysqlPort,
config.mysqlUsername,
config.mysqlPassword);
binaryLogClient.setBlocking(true);
binaryLogClient.setServerId(1001);
EventDeserializer eventDeserializer = new EventDeserializer();
eventDeserializer.setCompatibilityMode(EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG,
EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY);
binaryLogClient.setEventDeserializer(eventDeserializer);
binaryLogClient.registerEventListener(eventListener);
binaryLogClient.setBinlogFilename(binlogPositionManager.getBinlogFilename());
binaryLogClient.setBinlogPosition(binlogPositionManager.getPosition());
binaryLogClient.connect(3000);
LOGGER.info("Started.");
doProcess();}
//......
}- EventProcessor 提供了 start 方法，该方法首先执行 initDataSource；之后创建 BinlogPositionManager 并执行 binlogPositionManager.initBeginPosition()；然后创建 EventListener 及 BinaryLogClient 并执行 binaryLogClient.connect(3000)；最后执行 doProcess 方法
initDataSource
rocketmq-externals/rocketmq-mysql/src/main/java/org/apache/rocketmq/mysql/binlog/EventProcessor.java
public class EventProcessor {
//......
private void initDataSource() throws Exception {Map<String, String> map = new HashMap<>();
map.put("driverClassName", "com.mysql.jdbc.Driver");
map.put("url", "jdbc:mysql://" + config.mysqlAddr + ":" + config.mysqlPort + "?useSSL=true&verifyServerCertificate=false");
map.put("username", config.mysqlUsername);
map.put("password", config.mysqlPassword);
map.put("initialSize", "2");
map.put("maxActive", "2");
map.put("maxWait", "60000");
map.put("timeBetweenEvictionRunsMillis", "60000");
map.put("minEvictableIdleTimeMillis", "300000");
map.put("validationQuery", "SELECT 1 FROM DUAL");
map.put("testWhileIdle", "true");
dataSource = DruidDataSourceFactory.createDataSource(map);
}
//......
}- initDataSource 主要是通过 DruidDataSourceFactory 来创建 dataSource
doProcess
rocketmq-externals/rocketmq-mysql/src/main/java/org/apache/rocketmq/mysql/binlog/EventProcessor.java
public class EventProcessor {
//......
private void doProcess() {while (true) {
try {Event event = queue.poll(1000, TimeUnit.MILLISECONDS);
if (event == null) {checkConnection();
continue;
}
switch (event.getHeader().getEventType()) {
case TABLE_MAP:
processTableMapEvent(event);
break;
case WRITE_ROWS:
case EXT_WRITE_ROWS:
processWriteEvent(event);
break;
case UPDATE_ROWS:
case EXT_UPDATE_ROWS:
processUpdateEvent(event);
break;
case DELETE_ROWS:
case EXT_DELETE_ROWS:
processDeleteEvent(event);
break;
case QUERY:
processQueryEvent(event);
break;
case XID:
processXidEvent(event);
break;
}
} catch (Exception e) {LOGGER.error("Binlog process error.", e);
}
}
}
//......
}- doProcess 方法会执行 queue.poll(1000, TimeUnit.MILLISECONDS) 拉取 event，如果 event 为 null，则会执行 checkConnection；之后根据 event.getHeader().getEventType() 来做不同处理；主要有 processTableMapEvent、processWriteEvent、processUpdateEvent、processDeleteEvent、processQueryEvent、processXidEvent 这几种
processEvent
rocketmq-externals/rocketmq-mysql/src/main/java/org/apache/rocketmq/mysql/binlog/EventProcessor.java
public class EventProcessor {
//......
private void processTableMapEvent(Event event) {TableMapEventData data = event.getData();
String dbName = data.getDatabase();
String tableName = data.getTable();
Long tableId = data.getTableId();
Table table = schema.getTable(dbName, tableName);
tableMap.put(tableId, table);
}
private void processWriteEvent(Event event) {WriteRowsEventData data = event.getData();
Long tableId = data.getTableId();
List<Serializable[]> list = data.getRows();
for (Serializable[] row : list) {addRow("WRITE", tableId, row);
}
}
private void processUpdateEvent(Event event) {UpdateRowsEventData data = event.getData();
Long tableId = data.getTableId();
List<Map.Entry<Serializable[], Serializable[]>> list = data.getRows();
for (Map.Entry<Serializable[], Serializable[]> entry : list) {addRow("UPDATE", tableId, entry.getValue());
}
}
private void processDeleteEvent(Event event) {DeleteRowsEventData data = event.getData();
Long tableId = data.getTableId();
List<Serializable[]> list = data.getRows();
for (Serializable[] row : list) {addRow("DELETE", tableId, row);
}
}
private void processQueryEvent(Event event) {QueryEventData data = event.getData();
String sql = data.getSql();
if (createTablePattern.matcher(sql).find()) {schema.reset();
}
}
private void processXidEvent(Event event) {EventHeaderV4 header = event.getHeader();
XidEventData data = event.getData();
String binlogFilename = binaryLogClient.getBinlogFilename();
Long position = header.getNextPosition();
Long xid = data.getXid();
BinlogPosition binlogPosition = new BinlogPosition(binlogFilename, position);
transaction.setNextBinlogPosition(binlogPosition);
transaction.setXid(xid);
replicator.commit(transaction, true);
transaction = new Transaction(config);
}
private void addRow(String type, Long tableId, Serializable[] row) {if (transaction == null) {transaction = new Transaction(config);
}
Table t = tableMap.get(tableId);
if (t != null) {while (true) {if (transaction.addRow(type, t, row)) {break;} else {transaction.setNextBinlogPosition(replicator.getNextBinlogPosition());
replicator.commit(transaction, false);
transaction = new Transaction(config);
}
}
}
}
//......
}- processWriteEvent、processUpdateEvent、processDeleteEvent 都会执行 addRow 方法，它会执行 transaction.addRow(type, t, row)，如果返回 false 则会执行 transaction.setNextBinlogPosition 以及 replicator.commit；processXidEvent 会执行 binaryLogClient.getBinlogFilename()，更新 transaction 的 xid 及 binlogPosition，然后执行 replicator.commit(transaction, true)，并重置 transaction
小结
EventProcessor 提供了 start 方法，该方法首先执行 initDataSource；之后创建 BinlogPositionManager 并执行 binlogPositionManager.initBeginPosition()；然后创建 EventListener 及 BinaryLogClient 并执行 binaryLogClient.connect(3000)；最后执行 doProcess 方法
doc
- EventProcessor