com.github.shyiko.mysql.binlog.BinaryLogClient类的使用及代码示例

x33g5p2x  于2022-01-17 转载在 其他  
字(12.6k)|赞(0)|评价(0)|浏览(517)

本文整理了Java中com.github.shyiko.mysql.binlog.BinaryLogClient类的一些代码示例,展示了BinaryLogClient类的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台,是从一些精选项目中提取出来的代码,具有较强的参考意义,能在一定程度帮忙到你。BinaryLogClient类的具体详情如下:
包路径:com.github.shyiko.mysql.binlog.BinaryLogClient
类名称:BinaryLogClient

BinaryLogClient介绍

[英]MySQL replication stream client.
[中]MySQL复制流客户端。

代码示例

代码示例来源:origin: apache/nifi

binlogClient.registerEventListener(eventListener);
  binlogClient.registerLifecycleListener(lifecycleListener);
  binlogClient.setBinlogFilename(currentBinlogFile);
  if (currentBinlogPosition != DO_NOT_SET) {
    binlogClient.setBinlogPosition(currentBinlogPosition);
    binlogClient.setServerId(serverId);
      connectTimeout = Long.MAX_VALUE;
    binlogClient.connect(connectTimeout);
    transitUri = "mysql://" + connectedHost.getHostString() + ":" + connectedHost.getPort();
if (!binlogClient.isConnected()) {
  binlogClient.disconnect();
  binlogClient = null;
  throw new IOException("Could not connect binlog client to any of the specified hosts due to: " + lastConnectException.getMessage(), lastConnectException);
    binlogClient.disconnect();
    binlogClient = null;
    throw new IOException("Error creating binlog enrichment JDBC connection to any of the specified hosts", e);

代码示例来源:origin: apache/rocketmq-externals

public void start() throws Exception {
  initDataSource();
  binlogPositionManager = new BinlogPositionManager(config, dataSource);
  binlogPositionManager.initBeginPosition();
  schema = new Schema(dataSource);
  schema.load();
  eventListener = new EventListener(queue);
  binaryLogClient = new BinaryLogClient(config.mysqlAddr,
    config.mysqlPort,
    config.mysqlUsername,
    config.mysqlPassword);
  binaryLogClient.setBlocking(true);
  binaryLogClient.setServerId(1001);
  EventDeserializer eventDeserializer = new EventDeserializer();
  eventDeserializer.setCompatibilityMode(EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG,
    EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY);
  binaryLogClient.setEventDeserializer(eventDeserializer);
  binaryLogClient.registerEventListener(eventListener);
  binaryLogClient.setBinlogFilename(binlogPositionManager.getBinlogFilename());
  binaryLogClient.setBinlogPosition(binlogPositionManager.getPosition());
  binaryLogClient.connect(3000);
  LOGGER.info("Started.");
  doProcess();
}

代码示例来源:origin: shyiko/mysql-binlog-connector-java

public BinaryLogClientStatistics(BinaryLogClient binaryLogClient) {
  binaryLogClient.registerEventListener(this);
  binaryLogClient.registerLifecycleListener(this);
}

代码示例来源:origin: shyiko/mysql-binlog-connector-java

@Override
  public void run() {
    try {
      setConnectTimeout(timeout);
      connect();
    } catch (IOException e) {
      exceptionReference.set(e);
      countDownLatch.countDown(); // making sure we don't end up waiting whole "timeout"
    }
  }
};

代码示例来源:origin: zendesk/maxwell

private void ensureReplicatorThread() throws Exception {
  checkCommErrors();
  if ( !client.isConnected() && !stopOnEOF ) {
    if (this.gtidPositioning) {
      // When using gtid positioning, reconnecting should take us to the top
      // of the gtid event.  We throw away any binlog position we have
      // (other than GTID) and bail out of getTransactionRows()
      LOGGER.warn("replicator stopped at position: {} -- restarting", client.getGtidSet());
      client.setBinlogFilename("");
      client.setBinlogPosition(4L);
      client.connect(5000);
      throw new ClientReconnectedException();
    } else {
      // standard binlog positioning is a lot easier; we can really reconnect anywhere
      // we like, so we don't have to bail out of the middle of an event.
      LOGGER.warn("replicator stopped at position: {} -- restarting", client.getBinlogFilename() + ":" + client.getBinlogPosition());
      client.connect(5000);
    }
  }
}

代码示例来源:origin: eventuate-local/eventuate-local

public void start(Optional<BinlogFileOffset> binlogFileOffset, Consumer<M> eventConsumer) {
 client = new BinaryLogClient(host, port, dbUserName, dbPassword);
 client.setServerId(binlogClientUniqueId);
 client.setKeepAliveInterval(5 * 1000);
 client.setBinlogFilename(bfo.getBinlogFilename());
 client.setBinlogPosition(bfo.getOffset());
 client.setEventDeserializer(getEventDeserializer());
 client.registerEventListener(event -> {
  switch (event.getHeader().getEventType()) {
   case TABLE_MAP: {

代码示例来源:origin: debezium/debezium

protected void startClient(Consumer<BinaryLogClient> preConnect) throws IOException, TimeoutException, SQLException {
  // Connect the bin log client ...
  counters = new EventQueue(DEFAULT_TIMEOUT, this::logConsumedEvent, this::logIgnoredEvent);
  client = new BinaryLogClient(config.getHostname(), config.getPort(), "replicator", "replpass");
  client.setServerId(client.getServerId() - 1); // avoid clashes between BinaryLogClient instances
  client.setKeepAlive(false);
  client.setSSLMode(SSLMode.DISABLED);
  client.registerEventListener(counters);
  client.registerEventListener(this::recordEvent);
  client.registerLifecycleListener(new TraceLifecycleListener());
  EventDeserializer eventDeserializer = new EventDeserializer();
  eventDeserializer.setEventDataDeserializer(EventType.STOP, new StopEventDataDeserializer());
  client.setEventDeserializer(eventDeserializer);
  if (preConnect != null) preConnect.accept(client);
  client.connect(DEFAULT_TIMEOUT); // does not block
  // Set up the table as one transaction and wait to see the events ...
  conn.execute("DROP TABLE IF EXISTS person",
         "CREATE TABLE person (" +
             "  name VARCHAR(255) primary key," +
             "  age INTEGER NULL DEFAULT 10," +
             "  createdAt DATETIME NULL DEFAULT CURRENT_TIMESTAMP," +
             "  updatedAt DATETIME NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP" +
             ")");
  counters.consume(2, EventType.QUERY);
  counters.reset();
}

代码示例来源:origin: zendesk/maxwell

this.client = new BinaryLogClient(mysqlConfig.host, mysqlConfig.port, mysqlConfig.user, mysqlConfig.password);
this.client.setSSLMode(mysqlConfig.sslMode);
  String gtidStr = startBinlog.getGtidSetStr();
  LOGGER.info("Setting initial gtid to: " + gtidStr);
  this.client.setGtidSet(gtidStr);
  this.gtidPositioning = true;
} else {
  LOGGER.info("Setting initial binlog pos to: " + startBinlog.getFile() + ":" + startBinlog.getOffset());
  this.client.setBinlogFilename(startBinlog.getFile());
  this.client.setBinlogPosition(startBinlog.getOffset());
  this.gtidPositioning = false;
  this.client.setKeepAlive(false);
  EventDeserializer.CompatibilityMode.INVALID_DATE_AND_TIME_AS_MIN_VALUE
);
this.client.setEventDeserializer(eventDeserializer);
this.binlogEventListener = new BinlogConnectorEventListener(client, queue, metrics, outputConfig);
this.client.setBlocking(!stopOnEOF);
this.client.registerEventListener(binlogEventListener);
this.client.registerLifecycleListener(binlogLifecycleListener);
this.client.setServerId(replicaServerID.intValue());

代码示例来源:origin: perfectsense/dari

username = ObjectUtils.firstNonNull(username, "");
password = ObjectUtils.firstNonNull(password, "");
this.client = new BinaryLogClient(host, port, catalog, username, password);
this.lifecycleListener = new MySQLBinaryLogLifecycleListener(cache);
client.setServerId(RANDOM.nextLong());
client.registerLifecycleListener(lifecycleListener);
client.registerEventListener(new MySQLBinaryLogEventListener(database, cache, catalog));
eventDataDeserializers.put(EventType.GTID, new GtidEventDataDeserializer());
client.setEventDeserializer(
    new EventDeserializer(
        new EventHeaderV4Deserializer(),

代码示例来源:origin: debezium/debezium

client = new BinaryLogClient(connectionContext.hostname(), connectionContext.port(), connectionContext.username(), connectionContext.password());
client.setThreadFactory(Threads.threadFactory(MySqlConnector.class, context.getConnectorConfig().getLogicalName(), "binlog-client", false));
client.setServerId(serverId);
client.setSSLMode(sslModeFor(connectionContext.sslMode()));
client.setKeepAlive(context.config().getBoolean(MySqlConnectorConfig.KEEP_ALIVE));
client.setKeepAliveInterval(context.config().getLong(MySqlConnectorConfig.KEEP_ALIVE_INTERVAL_MS));
client.registerEventListener(context.bufferSizeForBinlogReader() == 0
    ? this::handleEvent
    : (new EventBuffer(context.bufferSizeForBinlogReader(), this))::add);
client.registerLifecycleListener(new ReaderThreadLifecycleListener());
if (logger.isDebugEnabled()) client.registerEventListener(this::logEvent);
                      new RowDeserializers.DeleteRowsDeserializer(
                          tableMapEventByTableId).setMayContainExtraInformation(true));
client.setEventDeserializer(eventDeserializer);

代码示例来源:origin: mysql-time-machine/replicator

@Override
public void connect(Checkpoint checkpoint) {
  if (this.client == null || !this.client.isConnected()) {
    this.executor.submit(() -> {
      for (String hostname : this.hostname) {
            this.client.registerEventListener(
                event -> {
                  try {
            this.client.setServerId(checkpoint.getServerId());
              this.client.setBinlogFilename(checkpoint.getBinlog().getFilename());
              this.client.setBinlogPosition(checkpoint.getBinlog().getPosition());
              this.client.setGtidSet(checkpoint.getGTID().getValue());
          this.client.connect();

代码示例来源:origin: debezium/debezium

protected void rewindBinaryLogClient(BinlogPosition position) {
  try {
    if (isRunning()) {
      logger.debug("Rewinding binlog to position {}", position);
      client.disconnect();
      client.setBinlogFilename(position.getFilename());
      client.setBinlogPosition(position.getPosition());
      client.connect();
    }
  } catch (IOException e) {
    logger.error("Unexpected error when re-connecting to the MySQL binary log reader", e);
  }
}

代码示例来源:origin: apache/rocketmq-externals

private void checkConnection() throws Exception {
  if (!binaryLogClient.isConnected()) {
    BinlogPosition binlogPosition = replicator.getNextBinlogPosition();
    if (binlogPosition != null) {
      binaryLogClient.setBinlogFilename(binlogPosition.getBinlogFilename());
      binaryLogClient.setBinlogPosition(binlogPosition.getPosition());
    }
    binaryLogClient.connect(3000);
  }
}

代码示例来源:origin: debezium/debezium

client.setGtidSet(filteredGtidSetStr);
  source.setCompletedGtidSet(filteredGtidSetStr);
  gtidSet = new com.github.shyiko.mysql.binlog.GtidSet(filteredGtidSetStr);
} else {
  client.setBinlogFilename(source.binlogFilename());
  client.setBinlogPosition(source.binlogPosition());
  gtidSet = new com.github.shyiko.mysql.binlog.GtidSet("");
client.setBinlogFilename(source.binlogFilename());
client.setBinlogPosition(source.binlogPosition());
try {
  logger.debug("Attempting to establish binlog reader connection with timeout of {} ms", timeoutInMilliseconds);
  client.connect(context.timeoutInMilliseconds());
} catch (TimeoutException e) {

代码示例来源:origin: zendesk/maxwell

@Override
protected void beforeStop() throws Exception {
  this.binlogEventListener.stop();
  this.client.disconnect();
}

代码示例来源:origin: debezium/debezium

@Override
protected void doStop() {
  try {
    if (client.isConnected()) {
      logger.debug("Stopping binlog reader '{}', last recorded offset: {}", this.name(), lastOffset);
      client.disconnect();
    }
    cleanupResources();
  } catch (IOException e) {
    logger.error("Unexpected error when disconnecting from the MySQL binary log reader '{}'", this.name(), e);
  }
}

代码示例来源:origin: zendesk/maxwell

public void startReplicator() throws Exception {
  this.client.connect(5000);
  replicatorStarted = true;
}

代码示例来源:origin: shyiko/mysql-binlog-connector-java

try {
    event = eventDeserializer.nextEvent(packetLength == MAX_PACKET_LENGTH ?
      new ByteArrayInputStream(readPacketSplitInChunks(inputStream, packetLength - 1)) :
      inputStream);
    if (event == null) {
      throw e;
    if (isConnected()) {
      for (LifecycleListener lifecycleListener : lifecycleListeners) {
        lifecycleListener.onEventDeserializationFailure(this, e);
  if (isConnected()) {
    eventLastSeen = System.currentTimeMillis();
    updateGtidSet(event);
    notifyEventListeners(event);
    updateClientBinlogFilenameAndPosition(event);
if (isConnected()) {
  for (LifecycleListener lifecycleListener : lifecycleListeners) {
    lifecycleListener.onCommunicationFailure(this, e);
if (isConnected()) {
  if (completeShutdown) {
    disconnect(); // initiate complete shutdown sequence (which includes keep alive thread)
  } else {
    disconnectChannel();

代码示例来源:origin: debezium/debezium

/**
 * Test case that is normally commented out since it is only useful to print out the DDL statements recorded by
 * the binlog during a MySQL server initialization and startup.
 * 
 * @throws Exception if there are problems
 */
@Ignore
@Test
public void shouldCaptureQueryEventData() throws Exception {
  // Testing.Print.enable();
  startClient(client -> {
    client.setBinlogFilename("mysql-bin.000001");
    client.setBinlogPosition(4);
  });
  counters.consumeAll(5, TimeUnit.SECONDS);
  List<QueryEventData> allQueryEvents = recordedEventData(QueryEventData.class, -1);
  allQueryEvents.forEach(event -> {
    String sql = event.getSql();
    if (sql.equalsIgnoreCase("BEGIN") || sql.equalsIgnoreCase("COMMIT")) return;
    System.out.println(event.getSql());
  });
}

代码示例来源:origin: mysql-time-machine/replicator

private BinaryLogClient getClient(String hostname) {
  // TODO: Implement status variable parser: https://github.com/shyiko/mysql-binlog-connector-java/issues/174
  BinaryLogClient client = new BinaryLogClient(
      hostname,
      this.port,
      this.schema,
      this.username,
      this.password
  );
  EventDeserializer eventDeserializer = new EventDeserializer();
  eventDeserializer.setCompatibilityMode(EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY);
  client.setEventDeserializer(eventDeserializer);
  return client;
}

相关文章