首先，我已经在这里发现了这个问题：flink程序在并行性中的行为不同，它看起来和我现在面临的问题相同，但我认为在我的场景中确实需要cep，因为我每小时有超过1000.000条属于不同用户密钥的记录需要分析。
所以当我用parallelism 1运行cep时，一切都很好，即使是对于不同的用户密钥，但是有点慢，因为flink需要在一个线程中逐个用户地分析，而且这个操作需要足够快，以便识别某些模式，然后在不到1分钟的时间内发送通知，例如，这就是为什么我需要一个以上的并行线程。
在我的例子中，我使用richflatmapfunction保持前一个模式来识别下一个模式，然后发送通知，下面是我的代码：

final DataStream<EventPush> eventsStream = RabbitMQConnector.eventStreamObject(env)
                .flatMap(new RabbitMQPushConsumer())
                .keyBy(k -> k.id);

private static SingleOutputStreamOperator<String> getPushToSend(KeyedStream<EventPush, String> stream) {
        return stream.flatMap(new WebPushFlatMapFunction())
                .map(x -> new ObjectMapper().writeValueAsString(x));
    }

/*the code below belongs to WebPushFlatMapFunction class, which is the RichFlatMapFunction using ValueState*/

 private boolean inTime(long start, long end) {
        final long difference = (end > start) ? (end - start) : (start - end);
        long time_frame = 120000L;
        return difference > 0 && time_frame >= difference;
    }

    @Override
    public void flatMap(EventPush value, Collector<EventPush> out) {
        final String pageName= value.pageName.trim();
        Tuple4<String, String, Long, Timestamp> prev;
        try {
            prev = previous.value();
            if (b_pageName.equalsIgnoreCase(pattern)) {
                LOG.info("umid " + value.idsUmid + " match (" + pattern + ") at: " + value.timestamp);
                previous.update(new Tuple4<>(value.idsUmid, pageName, value.timestamp.getTime(), value.timestamp));
            }
            if (prev != null) {
                if (inTime(value.timestamp.getTime(), prev.f2)) {
                    if ((prev.f1 != null && !prev.f1.equals("")) && prev.f1.equals(full_pattern) && pageName.equals(home) && prev.f3.before(value.timestamp)) {
                        if (PropertyFileReader.isWebPushLoggerActivated())
                            LOG.info("umid " + value.idsUmid + " match (" + home + ")" + "triggered at: " + value.timestamp);
                        prev.f1 = "";
                        out.collect(value);
                    }
                    if ((prev.f1 != null && !prev.f1.equals("")) && prev.f1.equals(pattern) && pageName.equals(full_pattern) && prev.f3.before(value.timestamp)) {
                        LOG.info("umid " + value.idsUmid + " match (" + full_pattern + ") at: " + value.timestamp);
                        prev.f3 = value.timestamp;
                        prev.f1 = pageName;
                        previous.update(prev);
                    }
                }
            }
        } catch (IOException e) {
            CatchHandler.generalCatchHandler(e);
        }
    }

用平行度1，我得到正确的顺序：1，2，3。我可以在一个线程中接收1个，在另一个线程中接收3个，因为所有的状态都属于同一个用户密钥，这3个状态将在不同的线程中进行分区。我的问题是：有没有任何方法可以用更多的并行性来做到这一点？谨致问候。