聊聊rocketmq-mysql的BinlogPositionManager

本篇文章主要是结合我之前面试的各种经历和实战开发中遇到的问题解决经验整理的，希望这篇《聊聊rocketmq-mysql的BinlogPositionManager》对你有很大帮助！欢迎收藏，分享给更多的需要的朋友学习~

序

本文主要研究一下rocketmq-mysql的BinlogPositionManager

BinlogPositionManager

rocketmq-externals/rocketmq-mysql/src/main/java/org/apache/rocketmq/mysql/position/BinlogPositionManager.java

public class BinlogPositionManager {
    private Logger logger = LoggerFactory.getLogger(BinlogPositionManager.class);

    private DataSource dataSource;
    private Config config;

    private String binlogFilename;
    private Long nextPosition;

    public BinlogPositionManager(Config config, DataSource dataSource) {
        this.config = config;
        this.dataSource = dataSource;
    }

    public void initBeginPosition() throws Exception {

        if (config.startType == null || config.startType.equals("DEFAULT")) {
            initPositionDefault();

        } else if (config.startType.equals("NEW_EVENT")) {
            initPositionFromBinlogTail();

        } else if (config.startType.equals("LAST_PROCESSED")) {
            initPositionFromMqTail();

        } else if (config.startType.equals("SPECIFIED")) {
            binlogFilename = config.binlogFilename;
            nextPosition = config.nextPosition;

        }

        if (binlogFilename == null || nextPosition == null) {
            throw new Exception("binlogFilename | nextPosition is null.");
        }
    }

    //......

    public String getBinlogFilename() {
        return binlogFilename;
    }

    public Long getPosition() {
        return nextPosition;
    }
}

• BinlogPositionManager提供了initBeginPosition、getBinlogFilename、getPosition方法；其中initBeginPosition方法根据config.startType的类型来执行不同逻辑，DEFAULT的是执行initPositionDefault，NEW_EVENT是执行initPositionFromBinlogTail，LAST_PROCESSED是执行initPositionFromMqTail，SPECIFIED是设置指定的binlogFilename及nextPosition

initPositionDefault

rocketmq-externals/rocketmq-mysql/src/main/java/org/apache/rocketmq/mysql/position/BinlogPositionManager.java

public class BinlogPositionManager {

    //......

    private void initPositionDefault() throws Exception {

        try {
            initPositionFromMqTail();
        } catch (Exception e) {
            logger.error("Init position from mq error.", e);
        }

        if (binlogFilename == null || nextPosition == null) {
            initPositionFromBinlogTail();
        }

    }

    //......

}

• initPositionDefault执行的是initPositionFromMqTail方法，在binlogFilename或者nextPosition为null时还会执行initPositionFromBinlogTail

initPositionFromBinlogTail

rocketmq-externals/rocketmq-mysql/src/main/java/org/apache/rocketmq/mysql/position/BinlogPositionManager.java

public class BinlogPositionManager {

    //......

    private void initPositionFromBinlogTail() throws SQLException {
        String sql = "SHOW MASTER STATUS";

        Connection conn = null;
        ResultSet rs = null;

        try {
            Connection connection = dataSource.getConnection();
            rs = connection.createStatement().executeQuery(sql);

            while (rs.next()) {
                binlogFilename = rs.getString("File");
                nextPosition = rs.getLong("Position");
            }

        } finally {

            if (conn != null) {
                conn.close();
            }
            if (rs != null) {
                rs.close();
            }
        }

    }

    //......

}

• initPositionFromBinlogTail方法通过

  SHOW MASTER STATUS

  来获取最新的binlogFilename及nextPosition

initPositionFromMqTail

rocketmq-externals/rocketmq-mysql/src/main/java/org/apache/rocketmq/mysql/position/BinlogPositionManager.java

public class BinlogPositionManager {

    //......

    private void initPositionFromMqTail() throws Exception {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("BINLOG_CONSUMER_GROUP");
        consumer.setNamesrvAddr(config.mqNamesrvAddr);
        consumer.setMessageModel(MessageModel.valueOf("BROADCASTING"));
        consumer.start();

        Set queues = consumer.fetchSubscribeMessageQueues(config.mqTopic);
        MessageQueue queue = queues.iterator().next();

        if (queue != null) {
            Long offset = consumer.maxOffset(queue);
            if (offset > 0)
                offset--;

            PullResult pullResult = consumer.pull(queue, "*", offset, 100);

            if (pullResult.getPullStatus() == PullStatus.FOUND) {
                MessageExt msg = pullResult.getMsgFoundList().get(0);
                String json = new String(msg.getBody(), "UTF-8");

                JSONObject js = JSON.parseObject(json);
                binlogFilename = (String) js.get("binlogFilename");
                nextPosition = js.getLong("nextPosition");
            }
        }

    }

    //......

}

• initPositionFromMqTail方法创建DefaultMQPullConsumer，然后从指定的topic拉取MessageQueue，然后通过consumer.maxOffset(queue)获取offset，再执行pull，找到第一个msg，然后设置binlogFilename及nextPosition

小结

BinlogPositionManager提供了initBeginPosition、getBinlogFilename、getPosition方法；其中initBeginPosition方法根据config.startType的类型来执行不同逻辑，DEFAULT的是执行initPositionDefault，NEW_EVENT是执行initPositionFromBinlogTail，LAST_PROCESSED是执行initPositionFromMqTail，SPECIFIED是设置指定的binlogFilename及nextPosition

doc

• BinlogPositionManager

今天关于《聊聊rocketmq-mysql的BinlogPositionManager》的内容就介绍到这里了，是不是学起来一目了然！想要了解更多关于mysql的内容请关注golang学习网公众号！