import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
@PostConstruct
public void postConstruct() throws MalformedObjectNameException, NotCompliantMBeanException, InstanceAlreadyExistsException, MBeanRegistrationException {

    log.debug("Create BinaryLogClient {}@{}:{}", mysqlUser, mysqlHost, mysqlPort);
    client = new BinaryLogClient(mysqlHost, mysqlPort, mysqlUser, mysqlPassword);
    log.debug("Getting columns names.");
    getColumnName();
    log.debug("Activate JMXBean");

    ObjectName objectName = new ObjectName("mysql.binlog:type=BinaryLogClient");
    mBeanServer.registerMBean(client, objectName);

    BinaryLogClientStatistics stats = new BinaryLogClientStatistics(client);
    ObjectName statsObjectName = new ObjectName("mysql.binlog:type=BinaryLogClientStatistics");
    mBeanServer.registerMBean(stats, statsObjectName);
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
public BinlogClientImpl(Vertx vertx, BinlogClientOptions options) {
  this.vertx = vertx;
  this.context = vertx.getOrCreateContext();
  this.host = options.getHost();
  this.port = options.getPort();
  this.connectTimeout = options.getConnectTimeout();
  this.sqlClient = MySQLClient.createNonShared(vertx,
    new JsonObject()
      .put("host", options.getHost())
      .put("port", options.getPort())
      .put("database", "information_schema")
      .put("maxPoolSize", 1)
      .put("username", options.getUsername())
      .put("password",
          "".equals(options.getPassword()) ?
            null : options.getPassword())
  );
  dispatcher = new EventDispatcher(vertx,
    new SchemaResolver(sqlClient));
  client = new BinaryLogClient(
    host, port,
    options.getUsername(),
    Optional.ofNullable(
      options.getPassword()
    ).orElse("")
  );
  if (options.getFilename() != null) {
    client.setBinlogFilename(options.getFilename());
  }
  if (options.getPosition() != -1) {
    client.setBinlogPosition(options.getPosition());
  }
  client.setHeartbeatInterval(options.getHeartbeatInterval());
  client.setKeepAlive(options.isKeepAlive());
  client.setKeepAliveInterval(options.getKeepAliveInterval());
  client.registerEventListener(this::handle);
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
@Override
public void onCommunicationFailure(BinaryLogClient client, Exception ex) {
    // TODO Auto-generated method stub
    if (ex.getMessage()
            .equals("1236 - Could not find first log file name in binary log index file")) {
        // The binary log has been rotate out from the master, so we can't
        // reconnect to it. Shutdown the listener, and tell the operator
        // that we need to rebootstrap.
        System.out.println(String.format("Binlog %s/%d is no longer available on the master. Need to re-bootstrap.", 
                client.getBinlogFilename(), client.getBinlogPosition()));
        this.failureReason = Reason.BINLOG_NOT_AVAILABLE;
    }
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
public BinlogEventListener(BinaryLogClient binaryLogClient, KafkaProducer kafkaProducer) {
	this.binaryLogClient = binaryLogClient;
	this.tableMap = new HashMap<Long, Table>();
	this.kafkaProducer = kafkaProducer;
	this.gtidSet = binaryLogClient.getGtidSet();

	this.replicateTables = new HashSet<String>();
	
	for(String fqTable: StringUtil.split(App.config.getProperty("replicateTables"), ',')) {
		String cleanFqTable = fqTable.trim();
		
		if(cleanFqTable.length() > 0)
			replicateTables.add(cleanFqTable);
	}
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
public void takeLeadership(CuratorFramework curatorFramework) throws InterruptedException {
	try {
		BinaryLogClient binaryLogClient = new BinaryLogClient(
	   		App.config.getProperty("mysql.host"),
	   		new Integer(App.config.getProperty("mysql.port")),
	   		App.config.getProperty("mysql.user"),
	   		App.config.getProperty("mysql.pass")
	   	);

    	GtidSync gtidSync = new GtidSync("/replicaza_gtid", curatorFramework.getZookeeperClient());
		gtidSync.start();
		
	   	binaryLogClient.setKeepAlive(false);
	   	binaryLogClient.setGtidSet(gtidSync.getGtidSet());
	   	
	   	KafkaProducer kafkaProducer = new KafkaProducer(
	   		App.config.getProperty("kafka.brokers"),
	   		new Integer(App.config.getProperty("kafka.requiredAcks")),
	   		gtidSync
	   	);
	   	
	   	kafkaProducer.start();

	   	binaryLogClient.registerEventListener(new BinlogEventListener(binaryLogClient, kafkaProducer));
	   	binaryLogClient.connect();
	} catch(Exception e) {
		e.printStackTrace();
	} finally {	
		// Sleep some time, but then give someone else a chance to become leader,
		// because someone else could have better connectivity
   		
		Thread.sleep(5000);
		
		System.exit(-1);
	}
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
public boolean isRunning() {
    if (BINLOG_EVENT_PARSER_PROVIDER_CODE == BinlogEventParserProviderCode.OR) {
        return ((OpenReplicator)binlogEventParserProvider).isRunning();
    }
    else { // if (BINLOG_EVENT_PARSER_PROVIDER_CODE == BinlogEventParserProviderCode.OR) {
        return  ((BinaryLogClient)binlogEventParserProvider).isConnected();
    }
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
public void stop(long timeout, TimeUnit unit) throws Exception {
    if (BINLOG_EVENT_PARSER_PROVIDER_CODE == BinlogEventParserProviderCode.OR) {
        stopOpenReplicator((OpenReplicator)binlogEventParserProvider,  timeout, unit);
    }
    else {
        stopBinaryLogClient((BinaryLogClient)binlogEventParserProvider,timeout, unit);
    }
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
public void start() throws Exception {
    if (binlogEventParserProvider instanceof OpenReplicator) {
        startOpenReplicator((OpenReplicator) binlogEventParserProvider);
    }
    else if (binlogEventParserProvider instanceof BinaryLogClient) {
        startBinaryLogClient((BinaryLogClient) binlogEventParserProvider);
    }
    else {
        throw new Exception("Unsupported parser exception");
    }
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
@Override
public void run() {
	client = new BinaryLogClient(host, port, user, password);
	// TODO: GTID support
	client.setBinlogFilename(position.getFilename());
	client.setBinlogPosition(position.getOffset());
	client.registerEventListener(this);
	try {
		client.connect();
	} catch (IOException e) {
		e.printStackTrace();
		System.exit(1);
	}
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
private BinaryLogClient createBinaryLogClient() {
  BinaryLogClient binLogClient = new BinaryLogClient(
      getConfig().hostname,
      port,
      getConfig().username,
      getConfig().password
  );
  if (getConfig().useSsl) {
    binLogClient.setSSLMode(SSLMode.REQUIRED);
  } else {
    binLogClient.setSSLMode(SSLMode.DISABLED);
  }
  binLogClient.setServerId(serverId);
  return binLogClient;
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
public BinaryLogClient getClient() {
  return client;
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
@Override
public void onConnect(BinaryLogClient client) {
    this.binlogFilename = client.getBinlogFilename();
    this.binlogClient = client;
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
@Override
public void onCommunicationFailure(BinaryLogClient client, Exception ex) {
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
@Override
public void onEventDeserializationFailure(BinaryLogClient client,
        Exception ex) {
    // TODO Auto-generated method stub
    
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
@Override
public void onDisconnect(BinaryLogClient client) {
    // TODO Auto-generated method stub
    
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
@Override
public void onConnect(BinaryLogClient client) {
    // TODO Auto-generated method stub

}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
@Override
public void onEventDeserializationFailure(BinaryLogClient client,
        Exception ex) {
    // TODO Auto-generated method stub

}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
@Override
public void onDisconnect(BinaryLogClient client) {
    // TODO Auto-generated method stub
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
private BinaryLogClient createBinaryLogClient() {
	return new BinaryLogClient("127.0.0.1", 3306, "root", "");
}
 

import com.github.shyiko.mysql.binlog.BinaryLogClient; //导入依赖的package包/类
public void stopBinaryLogClient (BinaryLogClient binaryLogClient, long timeout, TimeUnit unit) throws Exception {
    binaryLogClient.disconnect();
}