在Flink CEP中，我们如何检测持续一段时间的模式？

import org.apache.flink.cep.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.List;

// 定义事件类
class MyEvent {
    private String type;
    private long timestamp;

    public MyEvent(String type, long timestamp) {
        this.type = type;
        this.timestamp = timestamp;
    }

    public String getType() {
        return type;
    }

    public long getTimestamp() {
        return timestamp;
    }
}

public class FlinkCEPDurationPatternDetection {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 定义模式
        Pattern<MyEvent,?> pattern = Pattern.<MyEvent>begin("start")
               .where(new SimpleCondition<MyEvent>() {
                    @Override
                    public boolean filter(MyEvent event) throws Exception {
                        return event.getType().equals("A");
                    }
                })
               .followedBy("follow")
               .where(new SimpleCondition<MyEvent>() {
                    @Override
                    public boolean filter(MyEvent event) throws Exception {
                        return event.getType().equals("B");
                    }
                })
               .within(Time.seconds(5));
    }
}

// 假设events是一个DataStream<MyEvent>
DataStream<MyEvent> events =...;

PatternStream<MyEvent> patternStream = CEP.pattern(events, pattern);

SingleOutputStreamOperator<String> result = patternStream.select(new PatternSelectFunction<MyEvent, String>() {
    @Override
    public String select(PatternMatch<MyEvent> match) throws Exception {
        MyEvent startEvent = match.get("start");
        MyEvent followEvent = match.get("follow");
        return "Matched: " + startEvent.getType() + " at " + startEvent.getTimestamp() +
                " followed by " + followEvent.getType() + " at " + followEvent.getTimestamp();
    }
});

result.print();

env.execute("Flink CEP Duration Pattern Detection");