如何将消息发布到基于条件的2个kafka主题-- spring云流

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.support.MessageBuilder;

@EnableBinding(Source.class)
public class MessageProducer {

    private final Source source;

    public MessageProducer(Source source) {
        this.source = source;
    }

    public void sendMessage(String message, boolean condition) {
        source.output().send(MessageBuilder.withPayload(message).setHeader("condition", condition).build());
    }
}

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;

@EnableBinding(Sink.class)
public class MessageConsumer {

    @StreamListener(Sink.INPUT)
    public void receiveMessage(String message) {
        // 处理接收到的消息
    }
}

@Autowired
private MessageProducer messageProducer;

public void publishMessage(String message, boolean condition) {
    if (condition) {
        messageProducer.sendMessage(message, true);
    } else {
        messageProducer.sendMessage(message, false);
    }
}