Spring Cloud Stream KStream Consumer Concurrency 没有效果？

Question

我的消费者配置记录在Spring cloud stream consumer properties documentation.

spring-cloud-dependencies:Finchley.SR1
springBootVersion = '2.0.5.RELEASE'

我有 4 个分区用于 kstream_test 主题，它们充满了来自生产者的消息，如下所示：

root@kafka:/# kafka-run-class kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic kstream_test --time -1
kstream_test:2:222
kstream_test:1:203
kstream_test:3:188
kstream_test:0:278

我基于spring cloud stream kafka binder的配置是：

spring.cloud.stream.bindings.input:
  destination: kstream_test
  group: consumer-group-G1_test
  consumer:
    useNativeDecoding: true
    headerMode: raw
    startOffset: latest
    partitioned: true
    concurrency: 3

KStream 监听类

    @StreamListener
    @SendTo(MessagingStreams.OUTPUT)
    public KStream<?, ?> process(@Input(MessagingStreams.INPUT) KStream<?, ?> kstreams) {
        ......
        log.info("Got a message");
        ......
        return kstreams;
    }

我的生产者在 1 次运行中发送了 100 条消息。但是日志似乎只有 1 个线程 StreamThread-1 处理消息，尽管我的并发性为 3。这里可能有什么问题？ 100 条消息是否不足以查看并发性？

2018-10-18 11:50:01.923  INFO 10228 --- [-StreamThread-1] c.c.c.s.KStreamHandler     : Got a message
2018-10-18 11:50:01.923  INFO 10228 --- [-StreamThread-1] c.c.c.s.KStreamHandler     : Got a message
2018-10-18 11:50:01.945  INFO 10228 --- [-StreamThread-1] c.c.c.s.KStreamHandler     : Got a message
2018-10-18 11:50:01.956  INFO 10228 --- [-StreamThread-1] c.c.c.s.KStreamHandler     : Got a message
2018-10-18 11:50:01.972  INFO 10228 --- [-StreamThread-1] c.c.c.s.KStreamHandler     : Got a message

更新：

根据答案，以下num.stream.threads 配置在活页夹级别有效。

spring.cloud.stream.kafka.streams.binder.configuration:
 num.stream.threads: 3

Answer 1

看来需要设置num.stream.threads来增加并发...

/** {@code num.stream.threads} */
@SuppressWarnings("WeakerAccess")
public static final String NUM_STREAM_THREADS_CONFIG = "num.stream.threads";
private static final String NUM_STREAM_THREADS_DOC = "The number of threads to execute stream processing.";

...默认为 1。

活页夹应该根据...consumer.concurrency 属性进行真正的设置；请针对 the binder 打开一个 github 问题。

与此同时，您可以直接在...consumer.configuration 中设置该属性。

更正

我刚刚被告知...consumer.configuration 目前也没有应用于流绑定器；您必须在活页夹级别设置它。