聊聊maxwell的BinlogConnectorEventListener

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);
        }
    }
}

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;
    }
}

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);
    }
​
    //......
​
}