聊聊maxwell的MysqlPositionStore

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());
        }
    }
​
    //......
​
}   

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;
}

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);
            }
        }
    }
}