这是一个java问题，而不是flink问题。简而言之，您永远不能依赖thread.sleep（x）来睡眠x毫秒。正确地支持中断也很重要，否则您就不能优雅地关闭您的工作。

public class custom_source implements SourceFunction<String> {
    private static final Duration SLEEP_DURATION = Duration.ofHours(1);
    private volatile boolean isCanceled = false;

    public void run(SourceContext<String> ctx) {
        while (!isCanceled) {
            // 1 hour wait time
            LocalTime end = LocalTime.now().plusHours(1);
            // this loop ensures that random interruption is not prematurely closing the source
            while (LocalTime.now().compareTo(end) < 0) {
                try {
                    Thread.sleep(Duration.between(LocalTime.now(), end).toMillis());
                } catch (InterruptedException e) {
                    // swallow interruption unless source is canceled
                    if (isCanceled) {
                        Thread.interrupted();
                        return;
                    }
                }
            }
            ctx.collect("string");
        }
    }

    @Override
    public void cancel() {
        isCanceled = true;
    }
}