Spring集成Kafka线程配置

Question

我正在使用带有以下配置的spring集成kafka 1.1.0。我不太了解流的配置。当我增加这个值时，Spring会自动生成更多线程来处理消息吗？e、 g.当我有流=2时，相关的转换器和服务激活器是否都在2个线程中运行？我想错过一些线程执行器配置，但不知道如何配置。任何提示都将不胜感激。谢谢

<int:poller default="true" fixed-delay="10"/>

<int:channel id="tag.track">
</int:channel>

<int-kafka:inbound-channel-adapter id="kafkaInboundChannelAdapterForTagTrack" kafka-consumer-context-ref="consumerContextForTagTrack" auto-startup="true" channel="tag.track">
</int-kafka:inbound-channel-adapter>

<int-kafka:consumer-context id="consumerContextForTagTrack"
	consumer-timeout="${kafka.consumer.timeout}" zookeeper-connect="zookeeperConnect">
	<int-kafka:consumer-configurations>
		<int-kafka:consumer-configuration group-id="${kafka.consumer.group.track}" max-messages="200">
			<int-kafka:topic id="tag.track" streams="2" />
		</int-kafka:consumer-configuration>
	</int-kafka:consumer-configurations>
</int-kafka:consumer-context>

<int:channel id="tag.track.transformed">
	<int:interceptors>
		<int:wire-tap channel="event.logging" />
	</int:interceptors>
</int:channel>

<int:transformer id="kafkaMessageTransformerForTagTrack"
	ref="kafkaMessageTransformer" input-channel="tag.track" method="transform"
	output-channel="tag.track.transformed" />

<int:service-activator input-channel="tag.track.transformed" ref="tagTrackMessageHandler" method="handleTagMessage">
	<int:request-handler-advice-chain>
		<ref bean="userTagRetryAdvice" />
	</int:request-handler-advice-chain>
</int:service-activator>

已尝试使用消息驱动通道适配器，但无法使其工作，以下配置未拾取任何消息。还尝试了组织。springframework。集成。Kafka。侦听器。KafkaTopicOffsetManager，它抱怨偏移管理主题不能有多个分区。此外，在此适配器中，如何配置使用者组？是否有关于如何使用消息驱动通道适配器的详细示例？项目页面上的说明非常高。

<int:channel id="tag.track">
	<int:queue capacity="100"/>
</int:channel>

<bean id="kafkaConfiguration" class="org.springframework.integration.kafka.core.ZookeeperConfiguration">
	<constructor-arg ref="zookeeperConnect"/>
</bean>

<bean id="connectionFactory" class="org.springframework.integration.kafka.core.DefaultConnectionFactory">
	<constructor-arg ref="kafkaConfiguration"/>
</bean>

<bean id="decoder" class="org.springframework.integration.kafka.serializer.common.StringDecoder"/>

<int-kafka:message-driven-channel-adapter
		id="adapter"
		channel="tag.track"
		connection-factory="connectionFactory"
		key-decoder="decoder"
		payload-decoder="decoder"
		max-fetch="100"
		topics="tag.track"
		auto-startup="true"
		/>

Answer 1

streams属性与Spring本身无关；调用ConsumerConnector时，它只会传递给Kafka。createMessageStreams（）（每个主题/流条目都在map参数中传递）。

请参阅Kafka文档。

编辑：

使用高级使用者时，会轮询kafka入站通道适配器，因此下游集成流运行的线程与kafka客户端线程无关；它们在轮询器配置中进行管理。

您可以考虑改用消息驱动的通道适配器。