采用DleayQueue自定义flink的Source function产生测试数据

首先，我们需要导入以下依赖：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-core</artifactId>
    <version>${flink.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>

然后，我们可以定义一个自定义的Source Function，使用DelayQueue产生测试数据。下面是一个简单的示例代码：

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

public class DelayedEventSource implements SourceFunction<DelayedEvent> {

    private volatile boolean running = true;
    private final DelayQueue<DelayedEvent> queue = new DelayQueue<>();

    public void run(SourceContext<DelayedEvent> ctx) throws Exception {
        while (running) {
            DelayedEvent event = queue.take();
            ctx.collect(event);
        }
    }

    public void cancel() {
        running = false;
    }

    public void addEvent(long timestamp, String message) {
        queue.put(new DelayedEvent(timestamp, message));
    }

    public static class DelayedEvent implements Delayed {

        private final long timestamp;
        private final String message;

        public DelayedEvent(long timestamp, String message) {
            this.timestamp = timestamp;
            this.message = message;
        }

        public long getDelay(TimeUnit unit) {
            long diff = timestamp - System.currentTimeMillis();
            return unit.convert(diff, TimeUnit.MILLISECONDS);
        }

        public int compareTo(Delayed other) {
            if (this.timestamp < ((DelayedEvent) other).timestamp) {
                return -1;
            }
            if (this.timestamp > ((DelayedEvent) other).timestamp) {
                return 1;
            }
            return 0;
        }

        public String getMessage() {
            return message;
        }
    }
}

在这个示例中，我们定义了一个DelayQueue，用来存储延迟事件。run()方法在一个while循环中循环获取DelayQueue中的事件，并使用ctx.collect()方法将事件发送给下游算子。addEvent()方法用于向DelayQueue中添加新的延迟事件。

最后，我们可以通过以下代码来使用自定义的Source Function：

DelayedEventSource source = new DelayedEventSource();

// 添加测试数据
source.addEvent(System.currentTimeMillis() + 1000, "Hello");
source.addEvent(System.currentTimeMillis() + 2000, "World");

DataStream<DelayedEventSource.DelayedEvent> stream = env.addSource(source);

在这个示例中，我们创建了一个DelayedEventSource，并向其中添加了两个延迟事件。然后，我们将其作为Source Function添加到Flink的DataStream中。