在主题的特定分区中发送消息

Question

生产者有什么方法可以将消息发送到代理中的特定主题分区？

截至目前，可以发送具有 2 个分区的主题，但无法控制在特定分区中发送。

Component
@EnableBinding(Source.class)
public class RsvpsKafkaProducer {
  
    private static final int SENDING_MESSAGE_TIMEOUT_MS = 10000;

    private final Source source;

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

    public void sendRsvpMessage(WebSocketMessage<?> message) {
        
         System.out.println("sendRsvpMessage");
         source.output()
                .send(MessageBuilder.withPayload(message.getPayload())
                        .build(),
                        SENDING_MESSAGE_TIMEOUT_MS);   
    }
}

application.properties

spring.cloud.stream.kafka.binder.zkNodes=localhost:2181
spring.cloud.stream.kafka.binder.brokers=localhost:9093

spring.cloud.stream.bindings.output.destination=meetupTopic
spring.cloud.stream.bindings.output.producer.partitionCount=2

spring.cloud.stream.bindings.output.content-type=text/plain
spring.cloud.stream.bindings.output.producer.headerMode=raw

有什么方法可以使用 spring cloud stream 来实现吗？我希望一些消息进入P1 分区，一些消息进入meetupTopic 内的P2 分区。

Answer 1

MessageBuilder.withPayload(message.getPayload())
                        .setHeader(KafkaHeaders.PARTITION_ID, 23)
                        .build()

Answer 2

我还没有尝试过，但是从docs 看来，它可能会起作用。

spring.cloud.stream.bindings.output.producer.partitionSelectorExpression=headers['partitionKey']

然后在发送消息时添加标题partitionKey。