序

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

Replicator

rocketmq-externals/rocketmq-mysql/src/main/java/org/apache/rocketmq/mysql/Replicator.java

public class Replicator {

private static final Logger LOGGER = LoggerFactory.getLogger(Replicator.class);

private static final Logger POSITION_LOGGER = LoggerFactory.getLogger("PositionLogger");

private Config config;

private EventProcessor eventProcessor;

private RocketMQProducer rocketMQProducer;

private Object lock = new Object();

private BinlogPosition nextBinlogPosition;

private long nextQueueOffset;

private long xid;

public static void main(String[] args) {

Replicator replicator = new Replicator();

replicator.start();

}

public void start() {

try {

config = new Config();

config.load();

rocketMQProducer = new RocketMQProducer(config);

rocketMQProducer.start();

BinlogPositionLogThread binlogPositionLogThread = new BinlogPositionLogThread(this);

binlogPositionLogThread.start();

eventProcessor = new EventProcessor(this);

eventProcessor.start();

} catch (Exception e) {

LOGGER.error("Start error.", e);

System.exit(1);

}

}

public void commit(Transaction transaction, boolean isComplete) {

String json = transaction.toJson();

for (int i = 0; i < 3; i++) {

try {

if (isComplete) {

long offset = rocketMQProducer.push(json);

synchronized (lock) {

xid = transaction.getXid();

nextBinlogPosition = transaction.getNextBinlogPosition();

nextQueueOffset = offset;

}

} else {

rocketMQProducer.push(json);

}

break;

} catch (Exception e) {

LOGGER.error("Push error,retry:" + (i + 1) + ",", e);

}

}

}

public void logPosition() {

String binlogFilename = null;

long xid = 0L;

long nextPosition = 0L;

long nextOffset = 0L;

synchronized (lock) {

if (nextBinlogPosition != null) {

xid = this.xid;

binlogFilename = nextBinlogPosition.getBinlogFilename();

nextPosition = nextBinlogPosition.getPosition();

nextOffset = nextQueueOffset;

}

}

if (binlogFilename != null) {

POSITION_LOGGER.info("XID: {}, BINLOG_FILE: {}, NEXT_POSITION: {}, NEXT_OFFSET: {}",

xid, binlogFilename, nextPosition, nextOffset);

}

}

public Config getConfig() {

return config;

}

public BinlogPosition getNextBinlogPosition() {

return nextBinlogPosition;

}

}

Replicator提供了start、commit、logPosition方法；start方法会创建RocketMQProducer、BinlogPositionLogThread及EventProcessor，然后执行其start方法；commit方法会通过rocketMQProducer将transaction.toJson()发送出去，对于isComplete为true的会更新xid、nextBinlogPosition、nextQueueOffset；logPosition方法会打印binlogFilename、nextPosition、nextOffset

RocketMQProducer

rocketmq-externals/rocketmq-mysql/src/main/java/org/apache/rocketmq/mysql/productor/RocketMQProducer.java

public class RocketMQProducer {

private static final Logger LOGGER = LoggerFactory.getLogger(RocketMQProducer.class);

private DefaultMQProducer producer;

private Config config;

public RocketMQProducer(Config config) {

this.config = config;

}

public void start() throws MQClientException {

producer = new DefaultMQProducer("BINLOG_PRODUCER_GROUP");

producer.setNamesrvAddr(config.mqNamesrvAddr);

producer.start();

}

public long push(String json) throws Exception {

LOGGER.debug(json);

Message message = new Message(config.mqTopic, json.getBytes("UTF-8"));

SendResult sendResult = producer.send(message);

return sendResult.getQueueOffset();

}

}

RocketMQProducer的start方法创建DefaultMQProducer并执行其start方法；其push方法则通过producer.send(message)发送消息，并返回sendResult.getQueueOffset()

BinlogPositionLogThread

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

public class BinlogPositionLogThread extends Thread {

private Logger logger = LoggerFactory.getLogger(BinlogPositionLogThread.class);

private Replicator replicator;

public BinlogPositionLogThread(Replicator replicator) {

this.replicator = replicator;

setDaemon(true);

}

@Override

public void run() {

while (true) {

try {

Thread.sleep(1000);

} catch (InterruptedException e) {

logger.error("Offset thread interrupted.", e);

}

replicator.logPosition();

}

}

}

BinlogPositionLogThread会定时执行replicator.logPosition()来打印position信息

小结

Replicator提供了start、commit、logPosition方法；start方法会创建RocketMQProducer、BinlogPositionLogThread及EventProcessor，然后执行其start方法；commit方法会通过rocketMQProducer将transaction.toJson()发送出去，对于isComplete为true的会更新xid、nextBinlogPosition、nextQueueOffset；logPosition方法会打印binlogFilename、nextPosition、nextOffset

doc

Replicator

关于找一找教程网

本站文章仅代表作者观点，不代表本站立场，所有文章非营利性免费分享。

本站提供了软件编程、网站开发技术、服务器运维、人工智能等等IT技术文章，希望广大程序员努力学习，让我们用科技改变世界。

[聊聊rocketmq-mysql的Replicator]http://www.zyiz.net/tech/detail-137260.html