使用线程在消费流中使用Flink producer in循环运行Flink消费程序

Question

我希望flink consumer stream中的每条消息都能使用flink kafka producer生成多条消息，每条消息都通过一个单独的线程指向Kafka中的某个主题。我正在用Scala编写程序，但用Java就可以了

类似这样：

def thread(x:String): Thread =
    {   
    val thread_ = new Thread {
          override def run {

              val str = some_processing(x)
              flink_producer(str)

      }
    }
    return thread_
 }

val stream = flink_consumer()

stream.map(x =>{

                 var i = 0
                 while(i < 10){

                                val th = thread(x)
                                th.start()
                                i = i+1
                                }

           })

因此，对于flink消费者中的每个输入，我希望使用多线程向其他队列生成10条消息。

Answer 1

大多数Flink操作符都是并行操作符，因此您没有理由在数据管道中创建任何类型的线程，Flink应该是管理一个操作符可以存在多少个并行实例的操作符，如果您想设置该值，应该使用以下API方法。

<代码>。setParallelism（N）//N对您来说是10，

您可以在Fink文档中获得更多信息

你应该这样做：

1. 向群集配置中添加更多任务管理器插槽

您的代码应如下所示：

val stream = flink_consumer()

stream.flatMap((x, out) =>{
                 var i = 0
                 while(i < 10){
                      val valueToCollect = process(x,i)
                      out.collect(valueToCollect)
                 }

           }).setParallelism(10)
           .map(doSomethingWithGeneratedValues)
           .addSink(sinkThatSendsDataToYourDesiredSystem)

如果您知道要有多少个并行任务，请再进行一次尝试

val stream = flink_consumer()

val resultStream = stream.map(process)
val sinkStream = resultStream.union(resultStream,resultStream,resultStream,...) // joins resultStream N times
sinkStream.addSink(sinkThatSendsDataToYourDesiredSystem)

最后，您还可以为一个数据流设置多个接收器

val stream = flink_consumer()

val resultStream = stream.map(process)
resultStream.addSink(sinkThatSendsDataToYourDesiredSystem)
resultStream.addSink(sinkThatSendsDataToYourDesiredSystem)
resultStream.addSink(sinkThatSendsDataToYourDesiredSystem)
...
N
...
resultStream.addSink(sinkThatSendsDataToYourDesiredSystem)

如果要对数据接收器执行并行写入，必须确保所使用的接收器支持此类写入操作。