聊聊maxwell的BinlogConnectorEventListener

序

本文主要研究一下maxwell的BinlogConnectorEventListener

BinlogConnectorEventListener

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

class BinlogConnectorEventListener implements BinaryLogClient.EventListener {
    private static final Logger LOGGER = LoggerFactory.getLogger(BinlogConnectorEventListener.class);

    private final BlockingQueue<BinlogConnectorEvent> queue;
    private final Timer queueTimer;
    protected final AtomicBoolean mustStop = new AtomicBoolean(false);
    private final BinaryLogClient client;
    private final MaxwellOutputConfig outputConfig;
    private long replicationLag;
    private String gtid;

    public BinlogConnectorEventListener(
        BinaryLogClient client,
        BlockingQueue<BinlogConnectorEvent> q,
        Metrics metrics,
        MaxwellOutputConfig outputConfig
    ) {
        this.client = client;
        this.queue = q;
        this.queueTimer =  metrics.getRegistry().timer(metrics.metricName("replication", "queue", "time"));
        this.outputConfig = outputConfig;

        final BinlogConnectorEventListener self = this;
        metrics.register(metrics.metricName("replication", "lag"), (Gauge<Long>) () -> self.replicationLag);
    }

    public void stop() {
        mustStop.set(true);
    }

    @Override
    public void onEvent(Event event) {
        long eventSeenAt = 0;
        boolean trackMetrics = false;

        if (event.getHeader().getEventType() == EventType.GTID) {
            gtid = ((GtidEventData)event.getData()).getGtid();
        }

        BinlogConnectorEvent ep = new BinlogConnectorEvent(event, client.getBinlogFilename(), client.getGtidSet(), gtid, outputConfig);

        if (ep.isCommitEvent()) {
            trackMetrics = true;
            eventSeenAt = System.currentTimeMillis();
            replicationLag = eventSeenAt - event.getHeader().getTimestamp();
        }

        while (mustStop.get() != true) {
            try {
                if ( queue.offer(ep, 100, TimeUnit.MILLISECONDS ) ) {
                    break;
                }
            } catch (InterruptedException e) {
                return;
            }
        }

        if (trackMetrics) {
            queueTimer.update(System.currentTimeMillis() - eventSeenAt, TimeUnit.MILLISECONDS);
        }
    }
}

• BinlogConnectorEventListener實現了BinaryLogClient.EventListener接口，其onEvent方法建立BinlogConnectorEvent，而後將該event放入queue中

BinlogConnectorEvent

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

public class BinlogConnectorEvent {
    public static final String BEGIN = "BEGIN";
    public static final String COMMIT = "COMMIT";
    public static final String SAVEPOINT = "SAVEPOINT";
    private final MaxwellOutputConfig outputConfig;
    private BinlogPosition position;
    private BinlogPosition nextPosition;
    private final Event event;
    private final String gtidSetStr;
    private final String gtid;

    public BinlogConnectorEvent(Event event, String filename, String gtidSetStr, String gtid, MaxwellOutputConfig outputConfig) {
        this.event = event;
        this.gtidSetStr = gtidSetStr;
        this.gtid = gtid;
        EventHeaderV4 hV4 = (EventHeaderV4) event.getHeader();
        this.nextPosition = new BinlogPosition(gtidSetStr, gtid, hV4.getNextPosition(), filename);
        this.position = new BinlogPosition(gtidSetStr, gtid, hV4.getPosition(), filename);
        this.outputConfig = outputConfig;
    }

    public Event getEvent() {
        return event;
    }

    public WriteRowsEventData writeRowsData() {
        return (WriteRowsEventData) event.getData();
    }

    public UpdateRowsEventData updateRowsData() {
        return (UpdateRowsEventData) event.getData();
    }

    public DeleteRowsEventData deleteRowsData() {
        return (DeleteRowsEventData) event.getData();
    }

    public QueryEventData queryData() {
        return (QueryEventData) event.getData();
    }

    public XidEventData xidData() {
        return (XidEventData) event.getData();
    }

    public TableMapEventData tableMapData() {
        return (TableMapEventData) event.getData();
    }

    public BinlogPosition getPosition() {
        return position;
    }

    public BinlogPosition getNextPosition() {
        return nextPosition;
    }

    public EventType getType() {
        return event.getHeader().getEventType();
    }

    public Long getTableID() {
        EventData data = event.getData();
        switch ( event.getHeader().getEventType() ) {
            case EXT_WRITE_ROWS:
            case WRITE_ROWS:
                return ((WriteRowsEventData) data).getTableId();
            case EXT_UPDATE_ROWS:
            case UPDATE_ROWS:
                return ((UpdateRowsEventData) data).getTableId();
            case EXT_DELETE_ROWS:
            case DELETE_ROWS:
                return ((DeleteRowsEventData) data).getTableId();
            case TABLE_MAP:
                return ((TableMapEventData) data).getTableId();
        }
        return null;
    }

    public boolean isCommitEvent() {
        EventType eventType = getType();
        if (eventType == EventType.XID) {
            return true;
        } else if (eventType == EventType.QUERY) {
            // MyISAM will output a "COMMIT" QUERY_EVENT instead of a XID_EVENT.
            // There's no transaction ID but we can still set "commit: true"
            return COMMIT.equals(queryData().getSql());
        }

        return false;
    }

    private void writeData(Table table, RowMap row, Serializable[] data, BitSet includedColumns) {
        int dataIdx = 0, colIdx = 0;

        for ( ColumnDef cd : table.getColumnList() ) {
            if ( includedColumns.get(colIdx) ) {
                Object json = null;
                if ( data[dataIdx] != null ) {
                    json = cd.asJSON(data[dataIdx], outputConfig);
                }
                row.putData(cd.getName(), json);
                dataIdx++;
            }
            colIdx++;
        }
    }

    private void writeOldData(Table table, RowMap row, Serializable[] oldData, BitSet oldIncludedColumns) {
        int dataIdx = 0, colIdx = 0;

        for ( ColumnDef cd : table.getColumnList() ) {
            if ( oldIncludedColumns.get(colIdx) ) {
                Object json = null;
                if ( oldData[dataIdx] != null ) {
                    json = cd.asJSON(oldData[dataIdx], outputConfig);
                }

                if (!row.hasData(cd.getName())) {
                    /*
                       If we find a column in the BEFORE image that's *not* present in the AFTER image,
                       we're running in binlog_row_image = MINIMAL.  In this case, the BEFORE image acts
                       as a sort of WHERE clause to update rows with the new values (present in the AFTER image),
                       In this case we should put what's in the "before" image into the "data" section, not the "old".
                     */
                    row.putData(cd.getName(), json);
                } else {
                    if (!Objects.equals(row.getData(cd.getName()), json)) {
                        row.putOldData(cd.getName(), json);
                    }
                }
                dataIdx++;
            }
            colIdx++;
        }
    }

    private RowMap buildRowMap(String type, Position position, Position nextPosition, Serializable[] data, Table table, BitSet includedColumns, String rowQuery) {
        RowMap map = new RowMap(
            type,
            table.getDatabase(),
            table.getName(),
            event.getHeader().getTimestamp(),
            table.getPKList(),
            position,
            nextPosition,
            rowQuery
        );

        writeData(table, map, data, includedColumns);
        return map;
    }

    public List<RowMap> jsonMaps(Table table, long lastHeartbeatRead, String rowQuery) {
        ArrayList<RowMap> list = new ArrayList<>();

        Position position     = Position.valueOf(this.position, lastHeartbeatRead);
        Position nextPosition = Position.valueOf(this.nextPosition, lastHeartbeatRead);

        switch ( getType() ) {
            case WRITE_ROWS:
            case EXT_WRITE_ROWS:
                for ( Serializable[] data : writeRowsData().getRows() ) {
                    list.add(buildRowMap("insert", position, nextPosition, data, table, writeRowsData().getIncludedColumns(), rowQuery));
                }
                break;
            case DELETE_ROWS:
            case EXT_DELETE_ROWS:
                for ( Serializable[] data : deleteRowsData().getRows() ) {
                    list.add(buildRowMap("delete", position, nextPosition, data, table, deleteRowsData().getIncludedColumns(), rowQuery));
                }
                break;
            case UPDATE_ROWS:
            case EXT_UPDATE_ROWS:
                for ( Map.Entry<Serializable[], Serializable[]> e : updateRowsData().getRows() ) {
                    Serializable[] data = e.getValue();
                    Serializable[] oldData = e.getKey();

                    RowMap r = buildRowMap("update", position, nextPosition, data, table, updateRowsData().getIncludedColumns(), rowQuery);
                    writeOldData(table, r, oldData, updateRowsData().getIncludedColumnsBeforeUpdate());
                    list.add(r);
                }
                break;
        }

        return list;
    }
}

• BinlogConnectorEvent提供了jsonMaps方法，該方法根據event.getHeader().getEventType()類型來作不一樣處理，主要經過buildRowMap、writeOldData這幾個方法來實現；buildRowMap方法返回RowMap類型，writeOldData方法主要判斷是否putOldData

BinlogConnectorReplicator

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

public class BinlogConnectorReplicator extends RunLoopProcess implements Replicator {

    //......

    public RowMap getRow() throws Exception {
        BinlogConnectorEvent event;

        if ( stopOnEOF && hitEOF )
            return null;

        if ( !replicatorStarted )
            throw new ReplicatorNotReadyException("replicator not started!");

        while (true) {
            if (rowBuffer != null && !rowBuffer.isEmpty()) {
                RowMap row = rowBuffer.removeFirst();

                if ( row != null && isMaxwellRow(row) && row.getTable().equals("heartbeats") )
                    return processHeartbeats(row);
                else
                    return row;
            }

            event = pollEvent();

            if (event == null) {
                if ( stopOnEOF ) {
                    if ( client.isConnected() )
                        continue;
                    else
                        return null;
                } else {
                    try {
                        ensureReplicatorThread();
                    } catch ( ClientReconnectedException e ) {}
                    return null;
                }
            }

            switch (event.getType()) {
                case WRITE_ROWS:
                case EXT_WRITE_ROWS:
                case UPDATE_ROWS:
                case EXT_UPDATE_ROWS:
                case DELETE_ROWS:
                case EXT_DELETE_ROWS:
                    LOGGER.warn("Started replication stream inside a transaction.  This shouldn't normally happen.");
                    LOGGER.warn("Assuming new transaction at unexpected event:" + event);

                    queue.offerFirst(event);
                    rowBuffer = getTransactionRows(event);
                    break;
                case TABLE_MAP:
                    TableMapEventData data = event.tableMapData();
                    tableCache.processEvent(getSchema(), this.filter, data.getTableId(), data.getDatabase(), data.getTable());
                    break;
                case QUERY:
                    QueryEventData qe = event.queryData();
                    String sql = qe.getSql();
                    if (BinlogConnectorEvent.BEGIN.equals(sql)) {
                        try {
                            rowBuffer = getTransactionRows(event);
                        } catch ( ClientReconnectedException e ) {
                            // rowBuffer should already be empty by the time we get to this switch
                            // statement, but we null it for clarity
                            rowBuffer = null;
                            break;
                        }
                        rowBuffer.setServerId(event.getEvent().getHeader().getServerId());
                        rowBuffer.setThreadId(qe.getThreadId());
                        rowBuffer.setSchemaId(getSchemaId());
                    } else {
                        processQueryEvent(event);
                    }
                    break;
                case ROTATE:
                    tableCache.clear();
                    if ( stopOnEOF && event.getPosition().getOffset() > 0 ) {
                        this.binlogEventListener.mustStop.set(true);
                        this.client.disconnect();
                        this.hitEOF = true;
                        return null;
                    }
                    break;
                default:
                    break;
            }

        }
    }

    protected BinlogConnectorEvent pollEvent() throws InterruptedException {
        return queue.poll(100, TimeUnit.MILLISECONDS);
    }

    //......

}

• BinlogConnectorReplicator提供了getRow方法，它先從rowBuffer取row，取不到則經過pollEvent方法從queue中取event，以後根據不一樣的event.getType()來作不一樣處理，有的會經過getTransactionRows(event)來給rowBuffer賦值

小結

BinlogConnectorEventListener實現了BinaryLogClient.EventListener接口，其onEvent方法建立BinlogConnectorEvent，而後將該event放入queue中；BinlogConnectorReplicator提供了getRow方法，它會經過pollEvent從queue中拉取BinlogConnectorEvent，解析爲RowMap

doc

• BinlogConnectorEventListener