文档建议反馈控制台
首页
学习
活动
专区
圈层
工具
MCP广场
文章/答案/技术大牛
发布
社区首页 >问答首页 >Stream使用KafkaHeaders.Message_KEY向不同分区发送/使用消息

问Stream使用KafkaHeaders.Message_KEY向不同分区发送/使用消息
EN

Stack Overflow用户
提问于 2022-06-30 17:37:01
回答 1查看 347关注 0票数 0
代码语言：javascript
复制
I am trying to implement a prototype for implementing messaging system using Spring Cloud Stream. I selected Apache Kafka as binder. I created a topic with 2 partitions for scalability.  Then I tried to  send different messages to different partitions using following rest api method.

我为两个分区设置了两个不同的消息键。

代码语言：javascript
复制
@PostMapping("/publish")
public void publish(@RequestParam String message) {
    log.debug("REST request the message : {} to send to Kafka topic ", message);
    Message message1 = MessageBuilder.withPayload("Hello from a")
        .setHeader(KafkaHeaders.MESSAGE_KEY, "node1")
        .build();
    Message message2 = MessageBuilder.withPayload("Hello from b")
        .setHeader(KafkaHeaders.MESSAGE_KEY, "node1")
        .build();
    Message message3 = MessageBuilder.withPayload("Hello from c")
        .setHeader(KafkaHeaders.MESSAGE_KEY, "node1")
        .build();
    Message message4 = MessageBuilder.withPayload("Hello from d")
        .setHeader(KafkaHeaders.MESSAGE_KEY, "node2")
        .build();
    Message message5 = MessageBuilder.withPayload("Hello from e")
        .setHeader(KafkaHeaders.MESSAGE_KEY, "node2")
        .build();
    Message message6 = MessageBuilder.withPayload("Hello from f")
        .setHeader(KafkaHeaders.MESSAGE_KEY, "node2")
        .build();
    output.send("simulatePf-out-0", message1);
    output.send("simulatePf-out-0", message2);
    output.send("simulatePf-out-0", message3);
    output.send("simulatePf-out-0", message4);
    output.send("simulatePf-out-0", message5);
    output.send("simulatePf-out-0", message6);


}

这是我的生产者应用程序的application.yml

代码语言：javascript
复制
  cloud:
stream:
  kafka:
    binder:
      replicationFactor: 2
      auto-create-topics: true
      brokers: localhost:9092,localhost:9093,localhost:9094
      auto-add-partitions: true
    bindings:
      simulatePf-out-0:
        producer:
          configuration:
            key.serializer: org.apache.kafka.common.serialization.StringSerializer
            value.serializer: org.springframework.kafka.support.serializer.JsonSerializer
  bindings:
    simulatePf-out-0:
      producer:
        useNativeEncoding: true
        partition-count: 3
      destination: pf-topic
      content-type: text/plain
      group: dsa-back-end

为了测试并行性，我创建了一个使用者应用程序，该应用程序从pf-topic中读取消息。这是来自使用者应用程序的配置。

代码语言：javascript
复制
  cloud:
stream:
  kafka:
    binder:
      replicationFactor: 2
      auto-create-topics: true
      brokers: localhost:9092, localhost:9093, localhost:9094
      min-partition-count: 2
    bindings:
      simulatePf-in-0:
          consumer:
              configuration:
                key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
                value.deserializer: org.springframework.kafka.support.serializer.JsonDeserializer

  bindings:
    simulatePf-in-0:
      destination: pf-topic
      content-type: text/plain
      group: powerflowservice
      consumer:
        use-native-decoding: true

。我在消费者应用程序中创建了一个函数来使用消息。

代码语言：javascript
复制
   @Bean
public Consumer<Message> simulatePf() {
    return message -> {
        log.info("header " + message.getHeaders());
        log.info("received " + message.getPayload());
    };
}

现在是测试的时候了。为了测试并行性，我运行了两个spring引导使用者应用程序实例。我本来希望看到一个消费者使用来自一个分区的消息，而另一个消费者使用来自其他分区的信息。所以我希望消息a，消息b，信息被消费者消费。消息d、消息e和消息f是其他使用者的使用者。因为我设置了不同的消息键来分配不同的分区。但是，所有消息仅由一个应用程序使用。

代码语言：javascript
复制
 2022-06-30 20:34:48.895  INFO 11860 --- [container-0-C-1] c.s.powerflow.config.AsyncConfiguration  : header {deliveryAttempt=1, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=node1, kafka_receivedTopic=pf-topic, skip-input-type-conversion=true, kafka_offset=270, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@1eaf51df, source-type=streamBridge, id=a77d12f2-f184-0f2f-6a76-147803dd43f3, kafka_receivedPartitionId=0, kafka_receivedTimestamp=1656610488838, kafka_groupId=powerflowservice, timestamp=1656610488890}
2022-06-30 20:34:48.901  INFO 11860 --- [container-0-C-1] c.s.powerflow.config.AsyncConfiguration  : received Hello from a
2022-06-30 20:34:48.929  INFO 11860 --- [container-0-C-1] c.s.powerflow.config.AsyncConfiguration  : header {deliveryAttempt=1, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=node1, kafka_receivedTopic=pf-topic, skip-input-type-conversion=true, kafka_offset=271, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@1eaf51df, source-type=streamBridge, id=2e89f9b7-b6e7-482f-3c46-f73b2ad0705c, kafka_receivedPartitionId=0, kafka_receivedTimestamp=1656610488840, kafka_groupId=powerflowservice, timestamp=1656610488929}
2022-06-30 20:34:48.932  INFO 11860 --- [container-0-C-1] c.s.powerflow.config.AsyncConfiguration  : received Hello from b
2022-06-30 20:34:48.933  INFO 11860 --- [container-0-C-1] c.s.powerflow.config.AsyncConfiguration  : header {deliveryAttempt=1, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=node1, kafka_receivedTopic=pf-topic, skip-input-type-conversion=true, kafka_offset=272, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@1eaf51df, source-type=streamBridge, id=15640532-b57f-b58e-62e7-c2bc9375fdf0, kafka_receivedPartitionId=0, kafka_receivedTimestamp=1656610488841, kafka_groupId=powerflowservice, timestamp=1656610488933}
2022-06-30 20:34:48.934  INFO 11860 --- [container-0-C-1] c.s.powerflow.config.AsyncConfiguration  : received Hello from c
2022-06-30 20:34:48.935  INFO 11860 --- [container-0-C-1] c.s.powerflow.config.AsyncConfiguration  : header {deliveryAttempt=1, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=node2, kafka_receivedTopic=pf-topic, skip-input-type-conversion=true, kafka_offset=273, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@1eaf51df, source-type=streamBridge, id=590f0fb7-042f-e134-d214-ead570e42fe3, kafka_receivedPartitionId=0, kafka_receivedTimestamp=1656610488842, kafka_groupId=powerflowservice, timestamp=1656610488934}
2022-06-30 20:34:48.938  INFO 11860 --- [container-0-C-1] c.s.powerflow.config.AsyncConfiguration  : received Hello from d
2022-06-30 20:34:48.940  INFO 11860 --- [container-0-C-1] c.s.powerflow.config.AsyncConfiguration  : header {deliveryAttempt=1, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=node2, kafka_receivedTopic=pf-topic, skip-input-type-conversion=true, kafka_offset=274, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@1eaf51df, source-type=streamBridge, id=9a67e68b-95d4-a02e-cc14-ac30c684b639, kafka_receivedPartitionId=0, kafka_receivedTimestamp=1656610488842, kafka_groupId=powerflowservice, timestamp=1656610488940}
2022-06-30 20:34:48.941  INFO 11860 --- [container-0-C-1] c.s.powerflow.config.AsyncConfiguration  : received Hello from e
2022-06-30 20:34:48.943  INFO 11860 --- [container-0-C-1] c.s.powerflow.config.AsyncConfiguration  : header {deliveryAttempt=1, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=node2, kafka_receivedTopic=pf-topic, skip-input-type-conversion=true, kafka_offset=275, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@1eaf51df, source-type=streamBridge, id=333269af-bbd5-12b0-09de-8bd7959ebf08, kafka_receivedPartitionId=0, kafka_receivedTimestamp=1656610488843, kafka_groupId=powerflowservice, timestamp=1656610488943}
2022-06-30 20:34:48.943  INFO 11860 --- [container-0-C-1] c.s.powerflow.config.AsyncConfiguration  : received Hello from f

你能帮我一下我错过了什么吗。

apache-kafka
jhipster
spring-cloud-stream
java
EN

回答 1

Stack Overflow用户

回答已采纳

发布于 2022-06-30 22:39:41

您只在发送时将消息键设置为标题。您可以在消息中添加KafkaHeaders.PARTITION头以强制使用特定的分区。

如果不希望通过标头添加硬编码分区，可以在应用程序中设置分区键SpEL表达式或分区密钥提取器bean。这两种机制都是特定于的。如果您提供这两种方法中的任何一种，仍然需要告诉如何选择分区。为此，可以使用分区选择器SpEL表达式或分区选择器策略。如果您不提供它们，那么它将使用默认的选择器策略，获取消息键的主题分区数%的hashCode。

我想你昨天问了另一个相关的问题，我把这个博客链接到了我的答案中。在该博客的最后几节中，将解释所有这些细节。

引用博客的话：

如果您不提供分区密钥表达式或分区密钥提取器bean，那么将完全不参与为您做出任何分区决策的事务。在这种情况下，如果主题有多个分区，那么Kafka的默认分区机制将被触发。默认情况下，Kafka使用一个DefaultPartitioner，如果消息有一个键(见上面)，然后使用这个键的散列来计算分区。

我认为你在你的应用程序中看到了卡夫卡的默认行为。

票数 0
EN
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：

https://stackoverflow.com/questions/72819902

复制
相关文章

相似问题

领券

腾讯云开发者

扫码关注腾讯云开发者

扫码关注腾讯云开发者

领取腾讯云代金券

热门产品

热门推荐

更多推荐

Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有 

深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 粤公网安备44030502008569号

腾讯云计算（北京）有限责任公司 京ICP证150476号 |  京ICP备11018762号

问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档