异步写入 spring 云 kafka 流

Question

我想异步写入 spring 云 kafka 流。例如

Class SomeClass{

@StreamLister(Processor.INPUT)
public void receiveEvents(String e){

  class ThreadExecutor implements Runnable {
  private String message;

    public ThreadExecutor(message){
        this.message = message;
    }

    public void run(){
        //after processing the string I will publish it
         message = message + "done";
         writeToStream(message);
     }
   }

   Executors.newCachedThreadPool().execute(new ThreadExecutor(e));
}

@SendTo //not sure how to write it back
public Message<String> writeToStream(String message){
   //this is what I want to know
  }

}

所以在上面的例子中。我想知道如何调用 writeToStream 方法，以便它写回 kafka。与轮询不同，基本上我想在任务完成时写入流。请帮忙

Answer 1

@Autowired
private MessageChannel output;

...

    output.send(MessageBuilder.withPayload(data).build());

但是，不清楚您为什么需要这个；默认情况下，kafka 发送已经是异步的。

异步写入 spring 云 kafka 流

Asynchronously write to spring cloud kafka stream

spring-cloud-stream

spring-kafka