聊聊maxwell的BootstrapController

序

本文主要研究一下maxwell的BootstrapController

BootstrapController

maxwell-1.25.1/src/main/java/com/zendesk/maxwell/bootstrap/BootstrapController.java

public class BootstrapController extends RunLoopProcess  {
	static final Logger LOGGER = LoggerFactory.getLogger(BootstrapController.class);
	private final long MAX_TX_ELEMENTS = 10000;

	private final ConnectionPool maxwellConnectionPool;
	private final SynchronousBootstrapper bootstrapper;
	private final AbstractProducer producer;
	private final String clientID;
	private final boolean syncMode;
	private Long currentSchemaID;

	public BootstrapController(
		ConnectionPool maxwellConnectionPool,
		AbstractProducer producer,
		SynchronousBootstrapper bootstrapper,
		String clientID,
		boolean syncMode,
		Long currentSchemaID
	) {
		this.maxwellConnectionPool = maxwellConnectionPool;
		this.producer = producer;
		this.bootstrapper = bootstrapper;
		this.clientID = clientID;
		this.syncMode = syncMode;
		this.currentSchemaID = currentSchemaID;
	}

	// this mutex is used to block rows from being produced while a "synchronous"
	// bootstrap is run
	private Object bootstrapMutex = new Object();

	// this one is used to protect against races in an async producer.
	private Object completionMutex = new Object();
	private BootstrapTask activeTask;
	private RowMapBuffer skippedRows = new RowMapBuffer(MAX_TX_ELEMENTS);

	@Override
	protected void work() throws Exception {
		try {
			doWork();
		} catch ( InterruptedException e ) {
		} catch ( SQLException e ) {
			LOGGER.error("got SQLException trying to bootstrap", e);
		}
	}

	private void doWork() throws Exception {
		List<BootstrapTask> tasks = getIncompleteTasks();
		synchronized(bootstrapMutex) {
			for ( BootstrapTask task : tasks ) {
				LOGGER.debug("starting bootstrap task: {}", task.logString());
				synchronized(completionMutex) {
					activeTask = task;
				}

				bootstrapper.startBootstrap(task, producer, getCurrentSchemaID());

				synchronized(completionMutex) {
					pushSkippedRows();
					activeTask = null;
				}
			}
		}

		Thread.sleep(1000);
	}

	private synchronized Long getCurrentSchemaID() {
		return this.currentSchemaID;
	}

	public synchronized void setCurrentSchemaID(long schemaID) {
		this.currentSchemaID = schemaID;
	}

	private List<BootstrapTask> getIncompleteTasks() throws SQLException {
		ArrayList<BootstrapTask> list = new ArrayList<>();
		try ( Connection cx = maxwellConnectionPool.getConnection() ) {
			PreparedStatement s = cx.prepareStatement("select * from bootstrap where is_complete = 0 and client_id = ? order by id");
			s.setString(1, this.clientID);

			ResultSet rs = s.executeQuery();

			while (rs.next()) {
				list.add(BootstrapTask.valueOf(rs));
			}
		}
		return list;
	}

	public boolean shouldSkip(RowMap row) throws IOException {
		// The main replication thread skips rows of the currently bootstrapped
		// table and the tables that are queued for bootstrap. The bootstrap thread replays them at
		// the end of the bootstrap.

		if ( syncMode )
			synchronized(bootstrapMutex) { return false; }
		else {
			synchronized (completionMutex) {
				if (activeTask == null)
					return false;

				// async mode with an active task
				if (activeTask.matches(row)) {
					skippedRows.add(row);
					return true;
				} else
					return false;
			}
		}
	}

	private void pushSkippedRows() throws Exception {
		skippedRows.flushToDisk();
		while ( skippedRows.size() > 0 ) {
			RowMap row = skippedRows.removeFirst();
			producer.push(row);
		}
	}

}
複製代碼

• BootstrapController繼承了RunLoopProcess，其work方法執行doWork方法；其doWork方法經過getIncompleteTasks獲取tasks，而後遍歷task挨個執行bootstrapper.startBootstrap(task, producer, getCurrentSchemaID())及pushSkippedRows方法；getIncompleteTasks從數據庫中查詢bootstrap表中is_complete爲0的指定client_id的記錄；pushSkippedRows方法先執行skippedRows.flushToDisk()，而後遍歷移除並執行producer.push(row)

SynchronousBootstrapper

maxwell-1.25.1/src/main/java/com/zendesk/maxwell/bootstrap/SynchronousBootstrapper.java

public class SynchronousBootstrapper {
	static final Logger LOGGER = LoggerFactory.getLogger(SynchronousBootstrapper.class);
	private static final long INSERTED_ROWS_UPDATE_PERIOD_MILLIS = 250;
	private final MaxwellContext context;

	private long lastInsertedRowsUpdateTimeMillis = 0;

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


	public void startBootstrap(BootstrapTask task, AbstractProducer producer, Long currentSchemaID) throws Exception {
		performBootstrap(task, producer, currentSchemaID);
		completeBootstrap(task, producer);
	}

	private Schema captureSchemaForBootstrap(BootstrapTask task) throws SQLException {
		try ( Connection cx = getConnection(task.database) ) {
			CaseSensitivity s = MaxwellMysqlStatus.captureCaseSensitivity(cx);
			SchemaCapturer c = new SchemaCapturer(cx, s, task.database, task.table);
			return c.capture();
		}
	}

	public void performBootstrap(BootstrapTask task, AbstractProducer producer, Long currentSchemaID) throws Exception {
		LOGGER.debug("bootstrapping requested for " + task.logString());

		Schema schema = captureSchemaForBootstrap(task);
		Database database = findDatabase(schema, task.database);
		Table table = findTable(task.table, database);

		producer.push(bootstrapStartRowMap(task, table));
		LOGGER.info(String.format("bootstrapping started for %s.%s", task.database, task.table));

		try ( Connection streamingConnection = getStreamingConnection(task.database)) {
			setBootstrapRowToStarted(task.id);
			ResultSet resultSet = getAllRows(task.database, task.table, table, task.whereClause, streamingConnection);
			int insertedRows = 0;
			lastInsertedRowsUpdateTimeMillis = 0; // ensure updateInsertedRowsColumn is called at least once
			while ( resultSet.next() ) {
				RowMap row = bootstrapEventRowMap("bootstrap-insert", table.database, table.name, table.getPKList(), task.comment);
				setRowValues(row, resultSet, table);
				row.setSchemaId(currentSchemaID);

				Scripting scripting = context.getConfig().scripting;
				if ( scripting != null )
					scripting.invoke(row);

				if ( LOGGER.isDebugEnabled() )
					LOGGER.debug("bootstrapping row : " + row.toJSON());

				producer.push(row);
				Thread.sleep(1);
				++insertedRows;

				updateInsertedRowsColumn(insertedRows, task.id);
			}
			setBootstrapRowToCompleted(insertedRows, task.id);
		} catch ( NoSuchElementException e ) {
			LOGGER.info("bootstrapping aborted for " + task.logString());
		}
	}

	public void completeBootstrap(BootstrapTask task, AbstractProducer producer) throws Exception {
		producer.push(bootstrapEventRowMap("bootstrap-complete", task.database, task.table, new ArrayList<>(), task.comment));
		LOGGER.info("bootstrapping ended for " + task.logString());
	}

	//......

}	
複製代碼

• SynchronousBootstrapper的startBootstrap方法執行performBootstrap及completeBootstrap方法；其中performBootstrap方法執行producer.push(bootstrapStartRowMap(task, table))，以後根據task指定的數據庫、表及條件查詢記錄，而後遍歷結果集producer.push(row)；completeBootstrap方法執行producer.push(bootstrapEventRowMap("bootstrap-complete", task.database, task.table, new ArrayList<>(), task.comment))

小結

BootstrapController繼承了RunLoopProcess，其work方法執行doWork方法；其doWork方法經過getIncompleteTasks獲取tasks，而後遍歷task挨個執行bootstrapper.startBootstrap(task, producer, getCurrentSchemaID())及pushSkippedRows方法；getIncompleteTasks從數據庫中查詢bootstrap表中is_complete爲0的指定client_id的記錄；pushSkippedRows方法先執行skippedRows.flushToDisk()，而後遍歷移除並執行producer.push(row)

doc

• BootstrapController