聊聊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

原文 

https://segmentfault.com/a/1190000022539823

本站部分文章源于互联网,本着传播知识、有益学习和研究的目的进行的转载,为网友免费提供。如有著作权人或出版方提出异议,本站将立即删除。如果您对文章转载有任何疑问请告之我们,以便我们及时纠正。

PS:推荐一个微信公众号: askHarries 或者qq群:474807195,里面会分享一些资深架构师录制的视频录像:有Spring,MyBatis,Netty源码分析,高并发、高性能、分布式、微服务架构的原理,JVM性能优化这些成为架构师必备的知识体系。还能领取免费的学习资源,目前受益良多

转载请注明原文出处:Harries Blog™ » 聊聊maxwell的MysqlPositionStore

赞 (0)
分享到:更多 ()

评论 0

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址