聊聊maxwell的BinlogConnectorDiagnostic

序

本文主要研究一下maxwell的BinlogConnectorDiagnostic

MaxwellDiagnostic

maxwell-1.25.1/src/main/java/com/zendesk/maxwell/monitoring/MaxwellDiagnostic.java

public interface MaxwellDiagnostic {

	String getName();

	boolean isMandatory();

	String getResource();

	CompletableFuture<MaxwellDiagnosticResult.Check> check();

}
複製代碼

• MaxwellDiagnostic接口定義了getName、isMandatory、getResource、check方法；check方法返回的是MaxwellDiagnosticResult.Check類型的CompletableFuture

MaxwellDiagnosticResult

maxwell-1.25.1/src/main/java/com/zendesk/maxwell/monitoring/MaxwellDiagnosticResult.java

public class MaxwellDiagnosticResult {

	private final boolean success;
	private final boolean mandatoryFailed;
	private final List<Check> checks;

	public MaxwellDiagnosticResult(List<Check> checks) {
		success = checks.stream().allMatch(Check::isSuccess);
		mandatoryFailed = checks.stream().anyMatch(check -> !check.success && check.mandatory);
		this.checks = checks;
	}

	public boolean isSuccess() {
		return success;
	}

	public boolean isMandatoryFailed() {
		return mandatoryFailed;
	}

	public List<Check> getChecks() {
		return checks;
	}

	public static class Check {
		private final String name;
		private final boolean success;
		private final boolean mandatory;
		private final String resource;
		private final Map<String, String> info;

		public Check(MaxwellDiagnostic diagnostic, boolean success, Map<String, String> info) {
			this.name = diagnostic.getName();
			this.success = success;
			this.mandatory = diagnostic.isMandatory();
			this.resource = diagnostic.getResource();
			this.info = info;
		}

		public String getName() {
			return name;
		}

		public boolean isSuccess() {
			return success;
		}

		public boolean isMandatory() {
			return mandatory;
		}

		public String getResource() {
			return resource;
		}

		public Map<String, String> getInfo() {
			return info;
		}
	}
}
複製代碼

• MaxwellDiagnosticResult的構造器接收checks列表；Check定義了name、success、mandatory、resource、info屬性

BinlogConnectorDiagnostic

maxwell-1.25.1/src/main/java/com/zendesk/maxwell/replication/BinlogConnectorDiagnostic.java

public class BinlogConnectorDiagnostic implements MaxwellDiagnostic {

	private final MaxwellContext context;

	public BinlogConnectorDiagnostic(MaxwellContext context) {
			this.context = context;
	}

	@Override
	public String getName() {
		return "binlog-connector";
	}

	@Override
	public CompletableFuture<MaxwellDiagnosticResult.Check> check() {
		return getLatency().thenApply(this::normalResult).exceptionally(this::exceptionResult);
	}

	@Override
	public boolean isMandatory() {
		return true;
	}

	@Override
	public String getResource() {
		MaxwellMysqlConfig mysql = context.getConfig().maxwellMysql;
		return mysql.host + ":" + mysql.port;
	}

	public CompletableFuture<Long> getLatency() {
		HeartbeatObserver observer = new HeartbeatObserver(context.getHeartbeatNotifier(), Clock.systemUTC());
		try {
			context.heartbeat();
		} catch (Exception e) {
			observer.fail(e);
		}

		return observer.latency;
	}

	private MaxwellDiagnosticResult.Check normalResult(Long latency) {
		Map<String, String> info = new HashMap<>();
		info.put("message", "Binlog replication lag is " + latency.toString() + "ms");
		return new MaxwellDiagnosticResult.Check(this, true, info);
	}

	private MaxwellDiagnosticResult.Check exceptionResult(Throwable e) {
		Map<String, String> info = new HashMap<>();
		info.put("error", e.getCause().toString());
		return new MaxwellDiagnosticResult.Check(this, false, info);
	}

	//......

}
複製代碼

• BinlogConnectorDiagnostic實現了MaxwellDiagnostic接口，它定義了MaxwellContext屬性，其getName返回binlog-connector；其isMandatory方法返回true；其getResource方法返回的是mysql的host和port信息；其check方法經過getLatency方法獲取latency的CompletableFuture，成功時經過normalResult轉換，異常經過exceptionResult轉換；getLatency方法建立了HeartbeatObserver並註冊到HeartbeatNotifier中，而後執行context.heartbeat()，若出現異常執行observer.fail(e)，最後返回observer.latency

HeartbeatObserver

maxwell-1.25.1/src/main/java/com/zendesk/maxwell/replication/BinlogConnectorDiagnostic.java

static class HeartbeatObserver implements Observer {
		final CompletableFuture<Long> latency;
		private final HeartbeatNotifier notifier;
		private final Clock clock;

		HeartbeatObserver(HeartbeatNotifier notifier, Clock clock) {
			this.notifier = notifier;
			this.clock = clock;
			this.latency = new CompletableFuture<>();
			this.latency.whenComplete((value, exception) -> close());
			notifier.addObserver(this);
		}

		@Override
		public void update(Observable o, Object arg) {
			long heartbeatReadTime = clock.millis();
			long latestHeartbeat = (long) arg;
			latency.complete(heartbeatReadTime - latestHeartbeat);
		}

		void fail(Exception e) {
			latency.completeExceptionally(e);
		}

		private void close() {
			notifier.deleteObserver(this);
		}
	}
複製代碼

• HeartbeatObserver實現了Observer接口，其構造器將本身註冊到HeartbeatNotifier；其update方法更新latestHeartbeat，同時執行latency.complete；其fail方法執行latency.completeExceptionally(e)

HeartbeatNotifier

maxwell-1.25.1/src/main/java/com/zendesk/maxwell/replication/HeartbeatNotifier.java

public class HeartbeatNotifier extends Observable {

	protected void heartbeat(long heartbeat) {
		setChanged();
		notifyObservers(heartbeat);
	}

}
複製代碼

• HeartbeatNotifier繼承了Observable，其heartbeat方法執行setChanged以及notifyObservers

BinlogConnectorReplicator

maxwell-1.25.1/src/main/java/com/zendesk/maxwell/replication/BinlogConnectorReplicator.java

public class BinlogConnectorReplicator extends RunLoopProcess implements Replicator {

	//......

	private RowMap processHeartbeats(RowMap row) {
		String hbClientID = (String) row.getData("client_id");
		if ( !Objects.equals(hbClientID, this.clientID) )
			return row; // plain row -- do not process.

		long lastHeartbeatRead = (Long) row.getData("heartbeat");
		LOGGER.debug("replicator picked up heartbeat: " + lastHeartbeatRead);
		this.lastHeartbeatPosition = row.getPosition().withHeartbeat(lastHeartbeatRead);
		heartbeatNotifier.heartbeat(lastHeartbeatRead);
		return HeartbeatRowMap.valueOf(row.getDatabase(), this.lastHeartbeatPosition, row.getNextPosition().withHeartbeat(lastHeartbeatRead));
	}

	//......

}
複製代碼

• BinlogConnectorReplicator的processHeartbeats方法會執行heartbeatNotifier.heartbeat(lastHeartbeatRead)方法

小結

BinlogConnectorDiagnostic實現了MaxwellDiagnostic接口，它定義了MaxwellContext屬性，其getName返回binlog-connector；其isMandatory方法返回true；其getResource方法返回的是mysql的host和port信息；其check方法經過getLatency方法獲取latency的CompletableFuture，成功時經過normalResult轉換，異常經過exceptionResult轉換；getLatency方法建立了HeartbeatObserver並註冊到HeartbeatNotifier中，而後執行context.heartbeat()，若出現異常執行observer.fail(e)，最後返回observer.latency

doc

• BinlogConnectorDiagnostic