最近公司的项目需要用到kafka，因为整个项目基于Spring Cloud，所以想着不如用Spring Cloud Stream来集成。Spring Cloud Stream封装了一层抽象的接口，底层实现可以用kafka，也可以基于其他消息中间件。
环境
- Spring Cloud：Edgware SR5
- kafka-clients：0.10.1.1 这个是spring-cloud-stream-binder-kafka依赖的kafka-clients.jar的版本号
- kafka：版本号未知
kafka binder连接本地zookeeper
kafka binder在启动时会尝试连接本地zookeeper，如果本地没有zookeeper服务的话就会报错导致启动失败。解决方法是加入如下配置
spring.cloud.stream.kafka.binder.auto-create-topics=false
具体可以参考github issue#37
消费者接收数据异常
消费者接收数据用的是@StreamListener注解，主要参考如下示例
@SpringBootApplication
@EnableBinding(Sink.class)
public class VoteRecordingSinkApplication {
public static void main(String[] args) {
SpringApplication.run(VoteRecordingSinkApplication.class, args);
}
@StreamListener(Sink.INPUT)
public void processVote(Vote vote) {
votingService.recordVote(vote);
}
}
本以为照着官方示例写就万事大吉了，结果解析数据的时候抛出了StringIndexOutOfBoundException。因为之前用Greenwich版本的Spring Cloud Stream运行过同样的程序，那个时候是没有任何问题的。仔细研究了下官方文档，发现Edgware版本生产者和消费者的headerMode的默认配置为embeddedHeaders，而Greenwich版本则默认依赖于binder的实现。虽然不知道Spring Cloud Stream针对embeddedHeaders到底做了什么处理，不过可以想见应该是这里的问题。果然把headerMode改为raw之后就正常了。
生产者发送到kafka的指定分区
关键的配置如下
spring.cloud.stream.bindings.<channelName>.producer.partitionKeyExpression
spring.cloud.stream.default.producer.partitionCount
- 除了
partitionKeyExpression，还包括partitionKeyExtractorClass、partitionSelectorClass和partitionSelectorExpression等。这些都是用来定制更为复杂的发送策略的。 -
partitionCount是取模的基数，可以和kafka实际的分区数不一致。比如说如果配置为1的话，那么所有的数据都会发送到kafka的第0个分区。 - 注意不能用
spring.cloud.stream.default.producer.partitionKeyExpression，否则会提示
Failed to convert property value of type 'java.lang.String' to required type 'org.springframework.expression.Expression' for property 'producer.partitionKeyExpression'
具体原因可以参考github issue#1040和github pull#1041
消费者从kafka的指定分区接收数据
默认情况下kafka会自动平衡每个消费者对应的分区。比如说在只有一个消费者的情况下，所有的分区数据都会发送给这个消费者。这个时候如果再启动另一个消费者，kafka会自动进行调整，把一部分分区的数据发送给新启动的消费者。如果我们希望固定分区和消费者的对应关系，比如说处理的数据都是有状态的，这个时候我们可以采取如下方式
- 禁用kafka自动平衡
spring.cloud.stream.kafka.bindings.input.consumer.autoRebalanceEnabled
- 设置消费节点的分区信息
spring.cloud.stream.bindings.input.consumer.partitioned
spring.cloud.stream.instanceCount
spring.cloud.stream.instanceIndex
需要注意的是每台机器要有不同的instanceIndex
