聊聊BinlogConnectorReplicator的work

public interface StoppableTask {
    void requestStop() throws Exception;
    void awaitStop(Long timeout) throws TimeoutException;
}

abstract public class RunLoopProcess implements StoppableTask {
    protected volatile StoppableTaskState taskState;
    private Thread thread;

    public RunLoopProcess() {
        this.taskState = new StoppableTaskState(this.getClass().getName());
    }

    public void requestStop() {
        this.taskState.requestStop();

        interrupt();
    }

    public void interrupt() {
        if ( this.thread != null )
            this.thread.interrupt();
    }

    public void awaitStop(Long timeout) throws TimeoutException {
        this.taskState.awaitStop(thread, timeout);
    }

    public void runLoop() throws Exception {
        this.thread = Thread.currentThread();
        this.beforeStart();

        try {
            while (this.taskState.isRunning()) {
                work();
            }
        } finally {
            this.beforeStop();
            this.taskState.stopped();
        }
    }

    protected abstract void work() throws Exception;
    protected void beforeStart() throws Exception { }
    protected void beforeStop() throws Exception { }
}

public class BinlogConnectorReplicator extends RunLoopProcess implements Replicator {

    //......

    public void work() throws Exception {
        RowMap row = null;
        try {
            row = getRow();
        } catch ( InterruptedException e ) {
        }

        if ( row == null )
            return;

        rowCounter.inc();
        rowMeter.mark();

        if ( scripting != null && !isMaxwellRow(row))
            scripting.invoke(row);

        processRow(row);
    }

    protected void processRow(RowMap row) throws Exception {
        if ( row instanceof HeartbeatRowMap) {
            producer.push(row);
            if (stopAtHeartbeat != null) {
                long thisHeartbeat = row.getPosition().getLastHeartbeatRead();
                if (thisHeartbeat >= stopAtHeartbeat) {
                    LOGGER.info("received final heartbeat " + thisHeartbeat + "; stopping replicator");
                    // terminate runLoop
                    this.taskState.stopped();
                }
            }
        } else if ( !shouldSkipRow(row) )
            producer.push(row);
    }

    //......

}

public abstract class AbstractProducer {
    protected final MaxwellContext context;
    protected final MaxwellOutputConfig outputConfig;
    protected final Counter succeededMessageCount;
    protected final Meter succeededMessageMeter;
    protected final Counter failedMessageCount;
    protected final Meter failedMessageMeter;
    protected final Timer messagePublishTimer;
    protected final Timer messageLatencyTimer;
    protected final Counter messageLatencySloViolationCount;

    public AbstractProducer(MaxwellContext context) {
        this.context = context;
        this.outputConfig = context.getConfig().outputConfig;

        Metrics metrics = context.getMetrics();
        MetricRegistry metricRegistry = metrics.getRegistry();

        this.succeededMessageCount = metricRegistry.counter(metrics.metricName("messages", "succeeded"));
        this.succeededMessageMeter = metricRegistry.meter(metrics.metricName("messages", "succeeded", "meter"));
        this.failedMessageCount = metricRegistry.counter(metrics.metricName("messages", "failed"));
        this.failedMessageMeter = metricRegistry.meter(metrics.metricName("messages", "failed", "meter"));
        this.messagePublishTimer = metricRegistry.timer(metrics.metricName("message", "publish", "time"));
        this.messageLatencyTimer = metricRegistry.timer(metrics.metricName("message", "publish", "age"));
        this.messageLatencySloViolationCount = metricRegistry.counter(metrics.metricName("message", "publish", "age", "slo_violation"));
    }

    abstract public void push(RowMap r) throws Exception;

    public StoppableTask getStoppableTask() {
        return null;
    }

    public Meter getFailedMessageMeter() {
        return this.failedMessageMeter;
    }

    public MaxwellDiagnostic getDiagnostic() {
        return null;
    }
}

public class StdoutProducer extends AbstractProducer {
    public StdoutProducer(MaxwellContext context) {
        super(context);
    }

    @Override
    public void push(RowMap r) throws Exception {
        String output = r.toJSON(outputConfig);

        if ( output != null && r.shouldOutput(outputConfig) )
            System.out.println(output);

        this.context.setPosition(r);
    }
}