实时计算 Flink版产品使用合集之Akka RPC 压力过大，除了增大心跳超时，是否有其他解决方法

import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
public class CustomChangelogProducer<T> implements SourceFunction<ChangeEvent<T>> {
    private final Supplier<T> dataSupplier;
    private final long windowSize;
    private T currentData;
    public CustomChangelogProducer(Supplier<T> dataSupplier, long windowSize) {
        this.dataSupplier = dataSupplier;
        this.windowSize = windowSize;
    }
    @Override
    public void run(SourceContext<ChangeEvent<T>> ctx) throws Exception {
        while (true) {
            long currentTime = System.currentTimeMillis();
            T newData = dataSupplier.get();
            if (newData != null) {
                currentData = newData;
                ctx.collect(new ChangeEvent<>(currentTime, currentTime + windowSize, currentData));
            }
            Thread.sleep(1000);
        }
    }
    @Override
    public void cancel() {
        // No-op
    }
    @Override
    public boolean isBounded() {
        return false;
    }
    @Override
    public void onElement(ChangeEvent<T> element, Context context, Collector<ChangeEvent<T>> collector) throws Exception {
        // Do nothing
    }
    @Override
    public void onTimeout(long timeout, TimeWindow window, Context context) throws Exception {
        // Do nothing
    }
    @Override
    public void onWatermark(Watermark mark) throws Exception {
        // Do nothing
    }
}

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.WatermarkStrategy;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
public class Main {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 使用自定义的ChangelogProducer
        DataStream<String> inputStream = env.addSource(new CustomChangelogProducer<>(() -> "example", 1000L));
        // 对输入数据进行处理，例如按照主键进行对比，生成insert和delete的数据
        // ...
        env.execute("Flink Changelog Example");
    }
}

job.setRestartStrategy(RestartStrategies.fixedDelayRestart(
    int restartAttempts,
    long delayBetweenAttempts));
job.setFailoverStrategy(FailoverStrategies.standbyMode());