Блог

Есть ли какой-либо способ, которым производитель может отправлять сообщения в определенный раздел темы в брокере?

На данный момент я могу отправить тему, имеющую 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 раздел, а некоторые — в P2 раздел внутри meetupTopic .

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

Я не пробовал это, но из документов похоже, что это может сработать.

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

А затем вы добавляете заголовок partitionKey при отправке сообщения.