有人能帮我理解什么时候，怎样在Flink窗口（会议）发生？或者样本是如何处理的？
例如，如果我有一个连续的事件流流入，事件就是应用程序中的请求和应用程序提供的响应。作为flink处理的一部分，我们需要了解服务请求需要多少时间。
我知道有时间滚动窗口，每n秒触发一次，一旦时间过去，那么所有事件在该时间窗口将被聚合。
例如：假设定义的时间窗口是30秒，如果一个事件到达t时间，另一个事件到达t+30，那么这两个事件都将被处理，但是到达t+31的事件将被忽略。
如果我说的不正确，请改正。
上面的问题是：如果说一个事件在t时间到达，另一个事件在t+3时间到达，它还会等待整个30秒来聚合和最终确定结果吗？
在会话窗口的情况下，这是如何工作的？如果单独处理事件，并且在反序列化时将代理时间戳用作单个事件的会话id，那么将为每个事件创建会话窗口吗？如果是，那么我们是否需要区别对待请求和响应事件，因为如果不这样做，那么响应事件将获得自己的会话窗口？
我将尝试张贴我的例子（在java中），我正在玩在短时间内，但任何输入以上几点将是有益的！

过程函数

dto的：

public class IncomingEvent{
    private String id;
    private String eventId;
    private Date timestamp;
    private String component;
    //getters and setters
}
public class FinalOutPutEvent{
    private String id;
    private long timeTaken;
    //getters and setters
}

==========================================================传入事件的反序列化：
公共类incomingeventdeserializationscheme实现kafkadeserializationschema{

private ObjectMapper mapper;

public IncomingEventDeserializationScheme(ObjectMapper mapper) {
    this.mapper = mapper;
}

@Override
public TypeInformation<IncomingEvent> getProducedType() {
    return TypeInformation.of(IncomingEvent.class);
}

@Override
public boolean isEndOfStream(IncomingEvent nextElement) {
    return false;
}

@Override
public IncomingEvent deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
    if (record.value() == null) {
        return null;
    }
    try {
        IncomingEvent event = mapper.readValue(record.value(), IncomingEvent.class);
        if(event != null) {
            new SessionWindow(record.timestamp());
            event.setOffset(record.offset());
            event.setTopic(record.topic());
            event.setPartition(record.partition());
            event.setBrokerTimestamp(record.timestamp());
        }
        return event;
    } catch (Exception e) {
        return null;
    }
}

}

主要逻辑

public class MyEventJob {

private static final ObjectMapper mapper = new ObjectMapper();

public static void main(String[] args) throws Exception {

    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    MyEventJob eventJob = new MyEventJob();

    InputStream inStream = eventJob.getFileFromResources("myConfig.properties");
    ParameterTool parameter = ParameterTool.fromPropertiesFile(inStream);
    Properties properties = parameter.getProperties();
    Integer timePeriodBetweenEvents = 120;
    String outWardTopicHostedOnServer = localhost:9092";
    DataStreamSource<IncomingEvent> stream = env.addSource(new FlinkKafkaConsumer<>("my-input-topic", new IncomingEventDeserializationScheme(mapper), properties));
    SingleOutputStreamOperator<IncomingEvent> filteredStream = stream
        .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<IncomingEvent>() {
            long eventTime;
            @Override
            public long extractTimestamp(IncomingEvent element, long previousElementTimestamp) {
                return element.getTimestamp();
            }
            @Override
            public Watermark getCurrentWatermark() {
                return new Watermark(eventTime); 
            }
        })
        .map(e -> { e.setId(e.getEventId()); return e; });
    SingleOutputStreamOperator<FinalOutPutEvent> correlatedStream = filteredStream
        .keyBy(new KeySelector<IncomingEvent, String> (){
            @Override
            public String getKey(@Nonnull IncomingEvent input) throws Exception {
                return input.getId();
            }
        })
        .window(GlobalWindows.create()).allowedLateness(Time.seconds(defaultSliceTimePeriod))
        .trigger( new Trigger<IncomingEvent, Window> (){
            private final long sessionTimeOut;
            public SessionTrigger(long sessionTimeOut) {
                this.sessionTimeOut = sessionTimeOut;
            }
            @Override
            public TriggerResult onElement(IncomingEvent element, long timestamp, Window window, TriggerContext ctx)
                    throws Exception {
                ctx.registerProcessingTimeTimer(timestamp + sessionTimeOut); 
                return TriggerResult.CONTINUE;
            }
            @Override
            public TriggerResult onProcessingTime(long time, Window window, TriggerContext ctx) throws Exception {
                return TriggerResult.FIRE_AND_PURGE;
            }
            @Override
            public TriggerResult onEventTime(long time, Window window, TriggerContext ctx) throws Exception {
                    return TriggerResult.CONTINUE;
            }
            @Override
            public void clear(Window window, TriggerContext ctx) throws Exception {
                //check the clear method implementation
            }
        })
        .process(new ProcessWindowFunction<IncomingEvent, FinalOutPutEvent, String, SessionWindow>() {
        @Override
        public void process(String arg0,
                ProcessWindowFunction<IncomingEvent, FinalOutPutEvent, String, SessionWindow>.Context arg1,
                Iterable<IncomingEvent> input, Collector<FinalOutPutEvent> out) throws Exception {
            List<IncomingEvent> eventsIn = new ArrayList<>();
            input.forEach(eventsIn::add);
            if(eventsIn.size() == 1) {
                //Logic to handle incomplete request/response events
            } else if (eventsIn.size() == 2) {
                //Logic to handle the complete request/response and how much time it took
            }
        }
    } );
        FlinkKafkaProducer<FinalOutPutEvent> kafkaProducer = new FlinkKafkaProducer<>(
                outWardTopicHostedOnServer,            // broker list
                "target-topic",            // target topic
                new EventSerializationScheme(mapper));
    correlatedStream.addSink(kafkaProducer);
    env.execute("Streaming");
}

}
谢谢vicky