最近公司的项目需要用到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