kafka流中的延迟函数

Question

正在尝试使用 kafka 流代码进行某些操作，并希望在拆分数据后添加延迟或类似 threads.sleep() 1ms 之类的东西....我很困惑如何做到这一点..有人可以帮我做吗?

KStreamBuilder builder = new KStreamBuilder();
KStream<String, String> textlines = builder.stream("INTOPIC");
KStream<String, String> mstream = textlines
    .mapValues(value -> value.replace("[",""));
    .mapValues(value -> value.replace("]",""));
    .mapValues(value -> value.replaceAll("\\},\\{" ,"\\}\\},\\{\\{"))
    .flatMapValues(value -> Arrays.asList(value.split("\\},\\{")));
mstream.to("OUTTOPIC");
KafkaStreams streams = new KafkaStreams(builder, config);
streams.start();

所以在 .flatmapvalues 语句之后我需要添加一个 thread.sleep() 1ms 那么我的语句可以在那里..？

Answer 1

不确定您想要实现什么，但您似乎想减慢处理速度？比你可以在你的使用代码中加入睡眠。为此，您的 lambda 表达式必须在返回实际结果之前调用“sleep”。作为替代方案，您还可以添加一个额外的.foreach() 或peek() 呼叫并在那里睡觉。