springcloudstreambinderkafka属性配置

序

本文简单介绍下spring-cloud-stream-binder-kafka的一些属性配置。

maven

org.springframework.cloudspring-cloud-stream-binder-kafka1.0.3.RELEASE

stream属性

spring-cloud-stream-1.0.3.RELEASE-sources.jar!/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java

spring:cloud:stream:instanceIndex: 0 ##支持环境变量INSTANCE_INDEX## The instance index of the application: a number from 0 to instanceCount-1. Used for partitioning and with KafkainstanceCount: 1 ## The number of deployed instances of an application. Must be set for partitioning and if using Kafka.## used to partition data across different consumers.

Topic在逻辑上可以被认为是一个queue。每条消费都必须指定它的topic&＃xff0c;可以简单理解为必须指明把这条消息放进哪个queue里。为了使得 Kafka的吞吐率可以水平扩展&＃xff0c;物理上把topic分成一个或多个partition&＃xff0c;每个partition在物理上对应一个文件夹&＃xff0c;该文件夹下存储这个partition的所有消息和索引文件。partiton命名规则为topic名称&＃43;有序序号&＃xff0c;第一个partiton序号从0开始&＃xff0c;序号最大值为partitions数量减1。

同一个partition内的消息只能被同一个组中的一个consumer消费。

当消费者数量多于partition的数量时&＃xff0c;多余的消费者空闲。

消费者少于和等于partition的数量时&＃xff0c;会出现多个partition对应一个消费者的情况&＃xff0c;个别消费者消费量会比其他的多。

instanceCount主要是consumer用的&＃xff0c;一般小于或等于topic的partition数量&＃xff0c;主要用作消费者的消费分区用。

bingdings属性

spring-cloud-stream-1.0.3.RELEASE-sources.jar!/org/springframework/cloud/stream/config/BindingProperties.java

spring:cloud:stream:bindings:output:destination: event-democontent-type: text/plain#group: test ##consumer属性#producer:#consumer:

producer

spring-cloud-stream-1.0.3.RELEASE-sources.jar!/org/springframework/cloud/stream/binder/ProducerProperties.java

spring:cloud:stream:bindings:output:destination: event-democontent-type: text/plainproducer:partitionCount: 1headerModepartitionKeyExtractorClass: org.springframework.cloud.stream.partitioning.CustomPartitionKeyExtractorClasspartitionSelectorClass: org.springframework.cloud.stream.partitioning.CustomPartitionSelectorClassheaderMode: raw

kafka producer扩展属性
spring-cloud-stream-binder-kafka-1.0.3.RELEASE-sources.jar!/org/springframework/cloud/stream/binder/kafka/KafkaProducerProperties.java

spring:cloud:stream:bindings:output:destination: event-democontent-type: text/plainproducer:bufferSize: 16384maxRequestSize: 1048576sync: truebatchTimeout: 0

consumer

spring-cloud-stream-1.0.3.RELEASE-sources.jar!/org/springframework/cloud/stream/binder/ConsumerProperties.java

spring:cloud:stream:bindings:input:destination: event-democontent-type: text/plainconsumer:concurrency: 1 ## The concurrency of the inbound consumer.partitioned: false ## Whether the consumer receives data from a partitioned producer.Default: false.headerMode: raw

kafka consumer扩展属性
spring-cloud-stream-binder-kafka-1.0.3.RELEASE-sources.jar!/org/springframework/cloud/stream/binder/kafka/KafkaConsumerProperties.java

spring:cloud:stream:bindings:input:destination: event-democontent-type: text/plainconsumer:autoCommitOffset: falseresetOffsets: truestartOffset: earliestenableDlq: falserecoveryInterval: 5000

原生api

ConsumerConfig consumerConfig &＃61; new kafka.consumer.ConsumerConfig(props);ConsumerConnector consumerConnector &＃61; kafka.consumer.Consumer.createJavaConsumerConnector(consumerConfig);Map topicCountMap &＃61; new HashMap();topicCountMap.put(topic, consumerCount);Map>> consumerMap &＃61; consumerConnector.createMessageStreams(topicCountMap);

这里头topicCountMap告诉Kafka我们在Consumer中将用多少个线程来消费该topic。topicCountMap的key是topic name&＃xff0c;value针对该topic是线程的数量。

小结

整体的话&＃xff0c;spring cloud stream自己抽象了一部分&＃xff0c;但是有个硬伤就是spring.cloud.stream.instanceIndex这个不大友好&＃xff0c;这样就造成服务的实例是有状态的了&＃xff0c;在基于docker部署起来比较麻烦&＃xff0c;还不如直接原生api。如果partition不多&＃xff0c;或者每个consumer性能强悍的话&＃xff0c;那么至少部署两个&＃xff0c;配置起来也还可以接受。