spring-cloud-stream-binder-kafka发消息指定 partitionKey

组件版本

<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-kafka</artifactId><version>2.1.4.RELEASE</version></dependency>

背景信息

stream-binder 通过绑定 channel 发送消息很方便，
但 MessageChannel中 send 方法只接收 Message 参数无法指定 partitionKey；

default boolean send(Message<?> message) {return send(message, INDEFINITE_TIMEOUT);
}boolean send(Message<?> message, long timeout);

网友推荐

方法如下

// 指定
partitionKeyExtractorClasspartitionSelectorClasspartitionKeyExpression

需要自己定义Bean并且实现对应的接口，略显麻烦。

自查源码

可以在构造 Message 时，设置 header信息来指定 messageKey，后续会用作选取partition。

// partitionKey 为自定义的 业务key （String类型）
setHeader(KafkaHeaders.MESSAGE_KEY, partitionKey.getBytes())

理论依据

在 KafkaProducerMessageHandler 中有相应的处理逻辑：

//  org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler.handleRequestMessage(); L348protected Object handleRequestMessage(final Message<?> message) {MessageHeaders messageHeaders = message.getHeaders();String topic = this.topicExpression != null ?this.topicExpression.getValue(this.evaluationContext, message, String.class): messageHeaders.get(KafkaHeaders.TOPIC, String.class);Assert.state(StringUtils.hasText(topic), "The 'topic' can not be empty or null");Integer partitionId = this.partitionIdExpression != null ?this.partitionIdExpression.getValue(this.evaluationContext, message, Integer.class): messageHeaders.get(KafkaHeaders.PARTITION_ID, Integer.class);// 优先级低于 messageKeyExpressionObject messageKey = this.messageKeyExpression != null? this.messageKeyExpression.getValue(this.evaluationContext, message): messageHeaders.get(KafkaHeaders.MESSAGE_KEY);Long timestamp = this.timestampExpression != null? this.timestampExpression.getValue(this.evaluationContext, message, Long.class): messageHeaders.get(KafkaHeaders.TIMESTAMP, Long.class);...
}

DefaultPartitioner 根据 keyBytes 选取 partition

         // org.apache.kafka.clients.producer.internals.DefaultPartitioner.partition L69// hash the keyBytes to choose a partitionreturn Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;

spring-cloud-stream-binder-kafka发消息指定 partitionKey相关推荐

一. spring cloud gateway集成 spring cloud stream binder kafka，实现“动态路由“刷新与加载之采坑记录
一.前言 Spring Cloud Stream是用于构建消息驱动的微服务应用程序的框架. 本文主要介绍如何集成 Spring Cloud Stream,以 Kafka发布订阅模式(topic),实现 ...
Spring Cloud Stream 与 Kafka 整合
重要注意事项: 1. 尽管 Spring Cloud Stream Binder 中存在 Kafka 的整合,然而Spring Kafka 和 Spring Cloud Stream Kafka 在处 ...
Spring Cloud Stream Binder 实现
Spring Cloud Stream Binder 实现 JMS 实现 ActiveMQ 1.增加Maven依赖 <!-- 整合 Sprig Boot Starter ActiveMQ --& ...
Spring Cloud Stream与RabbitMQ 消费者消息分组
Spring Cloud Stream 其实是发布订阅模型,如果一个topic有多个订阅实例 ,消息就会被这些消息消费者接收到,这样就会带来一个问题,那就是消息的重复消费,这种问题在很多业务场景下是不 ...
Spring Cloud Stream同一通道根据消息内容分发不同的消费逻辑
应用场景有的时候,我们对于同一通道中的消息处理,会通过判断头信息或者消息内容来做一些差异化处理,比如:可能在消息头信息中带入消息版本号,然后通过if判断来执行不同的处理逻辑,其代码结构可能是这样的: ...
Spring Cloud【Finchley】- 21 Spring Cloud Stream 构建消息驱动微服务
文章目录概述添加依赖配置文件配置RabbitMQ的地址信息接口定义接收方 @EnableBinding @StreamListener 测试消费组发送复杂对象消息回执代码概述官网 ...
Spring Cloud Stream消息中间件通过RabbitMQ实现消息推送
一.Spring Cloud Stream是什么 Spring Cloud Stream 是一个构建消息驱动微服务的框架. Spring Cloud Stream解决了开发人员无感知的使用消息中间件的 ...
SpringCloud学习笔记 - 消息驱动 - Spring Cloud Stream
1. stream为什么被引入常见MQ(消息中间件): ActiveMQ RabbitMQ RocketMQ Kafka 有没有一种新的技术诞生,让我们不再关注具体MQ的细节,我们只需要用一种适配绑 ...
Kafka的Spring Cloud Stream
总览该示例项目演示了如何使用事件驱动的体系结构 , Spring Boot ,Spring Cloud Stream, Apache Kafka和Lombok构建实时流应用程序. 在本教程结束时,您 ...