Flink Kafka生产者元素乱序

Question

我试图使用FlinkKafkaProducer010生成元素，但是当我打开消费者控制台窗口时，元素似乎出现了故障。

我使用kafka-topics.bat创建了主题

消费者是使用：kafka console consumer创建的。蝙蝠——zookeeper本地主机：2181——主题mytopic

我使用的Kafka制作人代码是：

public static void main(String[] args) throws Exception {
    ParameterTool parameterTool = ParameterTool.fromArgs(args);
    if(parameterTool.getNumberOfParameters() < 2) {
        System.out.println("Missing parameters!");
        System.out.println("Usage: Kafka --topic <topic> --bootstrap.servers <kafka brokers>");
        return;
    }

    StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();
    env.getConfig().disableSysoutLogging();
    env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));

    DataStream<String> messageStream = env.addSource(getSourceFunction());

    FlinkKafkaProducer010<String> producer = new FlinkKafkaProducer010<>(parameterTool.getRequired("topic"), new SimpleStringSchema(), parameterTool.getProperties());
    messageStream.addSink(producer);
    env.execute("Kafka Producer");
}

public static SourceFunction<String> getSourceFunction() {
    return new SourceFunction<String>() {
        private static final long serialVersionUID = 6369260225318862378L;
        public boolean running = true;

        @Override
        public void run(SourceContext<String> ctx) {
            int counter = 0;
            while (this.running && counter < 500) {
                String data = "item " + Integer.toString(counter);
                ctx.collect(data);

                counter++;
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    };
}

当我查看Kafka日志文件时，我看到一个. log文件，其中的元素也是无序的。元素的顺序会跳转大约10个值。在我的用例中，拥有正确的顺序是必不可少的。我一直在搜索如何确保元素按顺序到达，但到目前为止没有任何运气。我是否错过了修复排序的东西？

提前感谢您的帮助！

Answer 1

我猜你在使用并行