Spring cloud stream binder kafka 常用配置

Spring Cloud Stream 集成 Kafka 示例

话不多说直接上配置：

spring:
  cloud:
    stream:
      kafka:
        binder:
          auto-add-partitions: true
          consumer-properties:
            max:
              poll:
                interval:
                  ms: 900000
                records: 50
      bindings:
        order-out-channel: #生产者通道
          destination: order-topic
          content-type: application/json
          group: order-group
        order-in-channel: #消费者通道
          destination: order-topic
          content-type: application/json
          group: order-group
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

1.其中配置Kafka消费者超时配置：

max.poll.records：单次消费者拉取的最大数据条数，默认值500。
1

max.poll.interval.ms：表示若在阈值时间之内消费者没有消费完上一次poll的消息，consumer client会主动向coordinator发起LeaveGroup请求，触发Rebalance；然后consumer重新发送JoinGroup请求。默认值300s，5分钟
1

也可以使用Spring集成Kakfa一样配置：

spring:
  kafka:
    consumer:
      max-poll-records: 5
      properties:
        max:
          poll:
            interval:
              ms: 900000 -- 每次消费的处理时间 15分钟
1
2
3
4
5
6
7
8
9

因为binder会合并配置：
org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties#mergedConsumerConfiguration

2.取消重试

order-in-channel: #消费者通道
destination: order-topic
content-type: application/json
group: order-group
consumer:
  max-attempts: 1
1
2
3
4
5
6