聊聊maxwell的MysqlPositionStore

序

本文主要研究一下maxwell的MysqlPositionStore

MysqlPositionStore

maxwell-1.25.1/src/main/java/com/zendesk/maxwell/schema/MysqlPositionStore.java

public class MysqlPositionStore {
	static final Logger LOGGER = LoggerFactory.getLogger(MysqlPositionStore.class);
	private static final Long DEFAULT_GTID_SERVER_ID = new Long(0);
	private final Long serverID;
	private String clientID;
	private final boolean gtidMode;
	private final ConnectionPool connectionPool;

	public MysqlPositionStore(ConnectionPool pool, Long serverID, String clientID, boolean gtidMode) {
		this.connectionPool = pool;
		this.clientID = clientID;
		this.gtidMode = gtidMode;
		if (gtidMode) {
			// we don't use server id for position store in gtid mode this.serverID = DEFAULT_GTID_SERVER_ID; } else { this.serverID = serverID; } } public void set(Position newPosition) throws SQLException, DuplicateProcessException { if ( newPosition == null ) return; Long heartbeat = newPosition.getLastHeartbeatRead(); String sql = "INSERT INTO `positions` set " + "server_id = ?, " + "gtid_set = ?, " + "binlog_file = ?, " + "binlog_position = ?, " + "last_heartbeat_read = ?, " + "client_id = ? " + "ON DUPLICATE KEY UPDATE " + "last_heartbeat_read = ?, " + "gtid_set = ?, binlog_file = ?, binlog_position=?"; BinlogPosition binlogPosition = newPosition.getBinlogPosition(); connectionPool.withSQLRetry(1, (c) -> { PreparedStatement s = c.prepareStatement(sql); LOGGER.debug("Writing binlog position to " + c.getCatalog() + ".positions: " + newPosition + ", last heartbeat read: " + heartbeat); s.setLong(1, serverID); s.setString(2, binlogPosition.getGtidSetStr()); s.setString(3, binlogPosition.getFile()); s.setLong(4, binlogPosition.getOffset()); s.setLong(5, heartbeat); s.setString(6, clientID); s.setLong(7, heartbeat); s.setString(8, binlogPosition.getGtidSetStr()); s.setString(9, binlogPosition.getFile()); s.setLong(10, binlogPosition.getOffset()); s.execute(); }); } public Position get() throws SQLException { try ( Connection c = connectionPool.getConnection() ) { PreparedStatement s = c.prepareStatement("SELECT * from `positions` where server_id = ? and client_id = ?"); s.setLong(1, serverID); s.setString(2, clientID); return positionFromResultSet(s.executeQuery()); } } //...... } 複製代碼

• MysqlPositionStore提供了set、get方法，其中set方法會insert一條記錄到positions表中，get方法則從positions表中取出指定server_id和client_id的position記錄；其中set方法使用了connectionPool.withSQLRetry來執行sql

ConnectionPool

maxwell-1.25.1/src/main/java/com/zendesk/maxwell/util/ConnectionPool.java

public interface ConnectionPool {
	@FunctionalInterface
	public interface RetryableSQLFunction<T> {
		void apply(T t) throws SQLException, NoSuchElementException, DuplicateProcessException;
	}

	Connection getConnection() throws SQLException;
	void release();

	void withSQLRetry(int nTries, RetryableSQLFunction<Connection> inner)
		throws SQLException, NoSuchElementException, DuplicateProcessException;
}
複製代碼

• ConnectionPool定義了RetryableSQLFunction、getConnection、release、withSQLRetry

C3P0ConnectionPool

maxwell-1.25.1/src/main/java/com/zendesk/maxwell/util/C3P0ConnectionPool.java

public class C3P0ConnectionPool implements ConnectionPool {
	private final ComboPooledDataSource cpds;
	static final Logger LOGGER = LoggerFactory.getLogger(C3P0ConnectionPool.class);

	@Override
	public Connection getConnection() throws SQLException {
		return cpds.getConnection();
	}

	@Override
	public void release() {
		cpds.close();
	}

	public C3P0ConnectionPool(String url, String user, String password) {
		cpds = new ComboPooledDataSource();
		cpds.setJdbcUrl(url);
		cpds.setUser(user);
		cpds.setPassword(password);


		// the settings below are optional -- c3p0 can work with defaults
		cpds.setMinPoolSize(1);
		cpds.setMaxPoolSize(5);
	}

	@Override
	public void withSQLRetry(int nTries, RetryableSQLFunction<Connection> inner)
		throws SQLException, DuplicateProcessException, NoSuchElementException {
		try ( final Connection c = getConnection() ){
			inner.apply(c);
			return;
		} catch (SQLException e) {
			if ( nTries > 0 ) {
				LOGGER.error("got SQL Exception: {}, {}, retrying...",
					e.getLocalizedMessage(),
					e.getCause().getLocalizedMessage()
				);
				withSQLRetry(nTries - 1, inner);
			} else {
				throw(e);
			}
		}
	}
}
複製代碼

• C3P0ConnectionPool實現了ConnectionPool接口，其withSQLRetry接口經過RetryableSQLFunction來執行邏輯，而後捕獲SQLException，在nTries大於0時遞歸執行withSQLRetry(nTries - 1, inner)

小結

MysqlPositionStore提供了set、get方法，其中set方法會insert一條記錄到positions表中，get方法則從positions表中取出指定server_id和client_id的position記錄；其中set方法使用了connectionPool.withSQLRetry來執行sql

doc

• MysqlPositionStore