Apache Kafka 支持
概览
Spring Integration 对 Apache Kafka 的支持基于 Spring for Apache Kafka 项目。
您需要在项目中包含此依赖项
-
Maven
-
Gradle
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-kafka</artifactId>
<version>6.4.4</version>
</dependency>
compile "org.springframework.integration:spring-integration-kafka:6.4.4"
它提供了以下组件
出站通道适配器
出站通道适配器用于将消息从 Spring Integration 通道发布到 Apache Kafka Topic。该通道在应用上下文中定义,然后被连接到向 Apache Kafka 发送消息的应用。发送应用可以通过使用 Spring Integration 消息发布到 Apache Kafka,这些消息会由出站通道适配器在内部转换为 Kafka 记录,如下所示
-
Spring Integration 消息的 Payload 用于填充 Kafka 记录的 Payload。
-
默认情况下,Spring Integration 消息的
kafka_messageKey
头用于填充 Kafka 记录的 Key。
您可以通过 kafka_topic
和 kafka_partitionId
头分别自定义消息发布的目标 Topic 和分区。
此外,<int-kafka:outbound-channel-adapter>
提供了通过对出站消息应用 SpEL 表达式来提取 Key、目标 Topic 和目标分区的能力。为此,它支持三对互斥的属性
-
topic
和topic-expression
-
message-key
和message-key-expression
-
partition-id
和partition-id-expression
这些属性允许您在适配器上将 topic
、message-key
和 partition-id
分别指定为静态值,或者在运行时根据请求消息动态评估它们的值。
KafkaHeaders 接口(由 spring-kafka 提供)包含用于与消息头交互的常量。messageKey 和 topic 默认头现在需要 kafka_ 前缀。从使用旧版消息头的早期版本迁移时,您需要在 <int-kafka:outbound-channel-adapter> 上指定 message-key-expression="headers['messageKey']" 和 topic-expression="headers['topic']" 。或者,您可以使用 <header-enricher> 或 MessageBuilder 将上游的消息头更改为 KafkaHeaders 中的新消息头。如果使用常量值,也可以通过 topic 和 message-key 在适配器上进行配置。 |
注意:如果适配器配置了 Topic 或 message key(无论是常量还是表达式),则会使用配置的值,而忽略相应的消息头。如果您希望消息头覆盖配置,则需要在表达式中进行配置,例如以下示例
topic-expression="headers['topic'] != null ? headers['topic'] : 'myTopic'"
适配器需要一个 KafkaTemplate
,而 KafkaTemplate
又需要一个适当配置的 KafkaProducerFactory
。
如果提供了 send-failure-channel
(sendFailureChannel
) 并且接收到 send()
失败(同步或异步),则会向该通道发送一个 ErrorMessage
。Payload 是一个 KafkaSendFailureException
,包含 failedMessage
、record
(ProducerRecord) 和 cause
属性。您可以通过设置 error-message-strategy
属性来覆盖默认的 DefaultErrorMessageStrategy
。
如果提供了 send-success-channel
(sendSuccessChannel
),则在发送成功后会发送一个 Payload 类型为 org.apache.kafka.clients.producer.RecordMetadata
的消息。
如果您的应用使用事务,并且同一个通道适配器用于发布由监听器容器启动事务的消息,以及在没有现有事务的情况下发布消息,则必须在 KafkaTemplate 上配置 transactionIdPrefix ,以覆盖容器或事务管理器使用的前缀。容器启动的事务使用的前缀(生产者工厂或事务管理器属性)在所有应用实例上必须相同。仅由生产者启动的事务使用的前缀在所有应用实例上必须唯一。 |
您可以配置一个 flushExpression
,它必须解析为一个布尔值。在使用 linger.ms
和 batch.size
Kafka 生产者属性时,发送多条消息后进行 flush 可能很有用;表达式在最后一条消息上应评估为 Boolean.TRUE
,不完整的批次将立即发送。默认情况下,表达式在 KafkaIntegrationHeaders.FLUSH
头 (kafka_flush
) 中查找一个 Boolean 值。如果该值为 true
,则会发生 flush;如果为 false
或该头不存在,则不会发生 flush。
KafkaProducerMessageHandler
的 sendTimeoutExpression
默认值已从 10 秒更改为 delivery.timeout.ms
Kafka 生产者属性 + 5000,以便在超时后将实际的 Kafka 错误传播到应用,而不是由该框架生成的超时。此更改是为了保持一致性,因为您可能会遇到意外行为(Spring 可能超时发送,而实际上,最终会成功)。重要提示:该超时默认为 120 秒,因此您可能希望减小它以获得更及时的失败信息。
配置
以下示例展示了如何配置 Apache Kafka 的出站通道适配器
-
Java DSL
-
Java
-
XML
@Bean
public ProducerFactory<Integer, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(embeddedKafka));
}
@Bean
public IntegrationFlow sendToKafkaFlow() {
return f -> f
.splitWith(s -> s.<String>function(p -> Stream.generate(() -> p).limit(101).iterator()))
.publishSubscribeChannel(c -> c
.subscribe(sf -> sf.handle(
kafkaMessageHandler(producerFactory(), TEST_TOPIC1)
.timestampExpression("T(Long).valueOf('1487694048633')"),
e -> e.id("kafkaProducer1")))
.subscribe(sf -> sf.handle(
kafkaMessageHandler(producerFactory(), TEST_TOPIC2)
.timestamp(m -> 1487694048644L),
e -> e.id("kafkaProducer2")))
);
}
@Bean
public DefaultKafkaHeaderMapper mapper() {
return new DefaultKafkaHeaderMapper();
}
private KafkaProducerMessageHandlerSpec<Integer, String, ?> kafkaMessageHandler(
ProducerFactory<Integer, String> producerFactory, String topic) {
return Kafka
.outboundChannelAdapter(producerFactory)
.messageKey(m -> m
.getHeaders()
.get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER))
.headerMapper(mapper())
.partitionId(m -> 10)
.topicExpression("headers[kafka_topic] ?: '" + topic + "'")
.configureKafkaTemplate(t -> t.id("kafkaTemplate:" + topic));
}
@Bean
@ServiceActivator(inputChannel = "toKafka")
public MessageHandler handler() throws Exception {
KafkaProducerMessageHandler<String, String> handler =
new KafkaProducerMessageHandler<>(kafkaTemplate());
handler.setTopicExpression(new LiteralExpression("someTopic"));
handler.setMessageKeyExpression(new LiteralExpression("someKey"));
handler.setSuccessChannel(successes());
handler.setFailureChannel(failures());
return handler;
}
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
@Bean
public ProducerFactory<String, String> producerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddress);
// set more properties
return new DefaultKafkaProducerFactory<>(props);
}
<int-kafka:outbound-channel-adapter id="kafkaOutboundChannelAdapter"
kafka-template="template"
auto-startup="false"
channel="inputToKafka"
topic="foo"
sync="false"
message-key-expression="'bar'"
send-failure-channel="failures"
send-success-channel="successes"
error-message-strategy="ems"
partition-id-expression="2">
</int-kafka:outbound-channel-adapter>
<bean id="template" class="org.springframework.kafka.core.KafkaTemplate">
<constructor-arg>
<bean class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
<constructor-arg>
<map>
<entry key="bootstrap.servers" value="localhost:9092" />
... <!-- more producer properties -->
</map>
</constructor-arg>
</bean>
</constructor-arg>
</bean>
消息驱动通道适配器
KafkaMessageDrivenChannelAdapter
(<int-kafka:message-driven-channel-adapter>
) 使用 spring-kafka 的 KafkaMessageListenerContainer
或 ConcurrentListenerContainer
。
此外,mode
属性可用。它可以接受 record
或 batch
值(默认:record
)。对于 record
模式,每个消息 Payload 都从单个 ConsumerRecord
转换而来。对于 batch
模式,Payload 是一个对象列表,这些对象由消费者轮询返回的所有 ConsumerRecord
实例转换而来。与批处理的 @KafkaListener
一样,KafkaHeaders.RECEIVED_KEY
、KafkaHeaders.RECEIVED_PARTITION
、KafkaHeaders.RECEIVED_TOPIC
和 KafkaHeaders.OFFSET
头也是列表,其位置与 Payload 中的位置相对应。
接收到的消息会填充某些消息头。请参阅KafkaHeaders
类了解更多信息。
Consumer 对象(在 kafka_consumer 头中)不是线程安全的。您只能在适配器内调用监听器的线程上调用其方法。如果您将消息传递给另一个线程,则不得调用其方法。 |
提供 retry-template
后,传递失败会根据其重试策略进行重试。如果同时提供了 error-channel
,则在重试耗尽后,将使用默认的 ErrorMessageSendingRecoverer
作为恢复回调。您也可以使用 recovery-callback
来指定在这种情况下要采取的其他操作,或将其设置为 null
,将最终异常抛给监听器容器,以便在那里进行处理。
构建 ErrorMessage
(用于 error-channel
或 recovery-callback
)时,您可以通过设置 error-message-strategy
属性来自定义错误消息。默认情况下,使用 RawRecordHeaderErrorMessageStrategy
,以便访问已转换的消息以及原始的 ConsumerRecord
。
这种形式的重试是阻塞的,如果所有轮询记录的总重试延迟可能超过 max.poll.interval.ms 消费者属性,则可能导致重新平衡。相反,考虑向监听器容器添加一个 DefaultErrorHandler ,并将其配置为 KafkaErrorSendingMessageRecoverer 。 |
配置
以下示例展示了如何配置消息驱动通道适配器
-
Java DSL
-
Java
-
XML
@Bean
public IntegrationFlow topic1ListenerFromKafkaFlow() {
return IntegrationFlow
.from(Kafka.messageDrivenChannelAdapter(consumerFactory(),
KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1)
.configureListenerContainer(c ->
c.ackMode(AbstractMessageListenerContainer.AckMode.MANUAL)
.id("topic1ListenerContainer"))
.recoveryCallback(new ErrorMessageSendingRecoverer(errorChannel(),
new RawRecordHeaderErrorMessageStrategy()))
.retryTemplate(new RetryTemplate())
.filterInRetry(true))
.filter(Message.class, m ->
m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer.class) < 101,
f -> f.throwExceptionOnRejection(true))
.<String, String>transform(String::toUpperCase)
.channel(c -> c.queue("listeningFromKafkaResults1"))
.get();
}
@Bean
public KafkaMessageDrivenChannelAdapter<String, String>
adapter(KafkaMessageListenerContainer<String, String> container) {
KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenChannelAdapter =
new KafkaMessageDrivenChannelAdapter<>(container, ListenerMode.record);
kafkaMessageDrivenChannelAdapter.setOutputChannel(received());
return kafkaMessageDrivenChannelAdapter;
}
@Bean
public KafkaMessageListenerContainer<String, String> container() throws Exception {
ContainerProperties properties = new ContainerProperties(this.topic);
// set more properties
return new KafkaMessageListenerContainer<>(consumerFactory(), properties);
}
@Bean
public ConsumerFactory<String, String> consumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddress);
// set more properties
return new DefaultKafkaConsumerFactory<>(props);
}
<int-kafka:message-driven-channel-adapter
id="kafkaListener"
listener-container="container1"
auto-startup="false"
phase="100"
send-timeout="5000"
mode="record"
retry-template="template"
recovery-callback="callback"
error-message-strategy="ems"
channel="someChannel"
error-channel="errorChannel" />
<bean id="container1" class="org.springframework.kafka.listener.KafkaMessageListenerContainer">
<constructor-arg>
<bean class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
<constructor-arg>
<map>
<entry key="bootstrap.servers" value="localhost:9092" />
...
</map>
</constructor-arg>
</bean>
</constructor-arg>
<constructor-arg>
<bean class="org.springframework.kafka.listener.config.ContainerProperties">
<constructor-arg name="topics" value="foo" />
</bean>
</constructor-arg>
</bean>
您还可以使用用于 @KafkaListener
注解的容器工厂来创建 ConcurrentMessageListenerContainer
实例用于其他目的。请参阅Spring for Apache Kafka 文档以获取示例。
使用 Java DSL,容器不必配置为 @Bean
,因为 DSL 会将容器注册为一个 bean。以下示例展示了如何做到这一点
@Bean
public IntegrationFlow topic2ListenerFromKafkaFlow() {
return IntegrationFlow
.from(Kafka.messageDrivenChannelAdapter(kafkaListenerContainerFactory().createContainer(TEST_TOPIC2),
KafkaMessageDrivenChannelAdapter.ListenerMode.record)
.id("topic2Adapter"))
...
get();
}
请注意,在这种情况下,适配器被赋予了一个 id
(topic2Adapter
)。容器在应用上下文中注册,名称为 topic2Adapter.container
。如果适配器没有 id
属性,则容器的 bean 名称是容器的完全限定类名加上 #n
,其中 n
为每个容器递增。
入站通道适配器
KafkaMessageSource
提供了一个可轮询的通道适配器实现。
配置
-
Java DSL
-
Kotlin
-
Java
-
XML
@Bean
public IntegrationFlow flow(ConsumerFactory<String, String> cf) {
return IntegrationFlow.from(Kafka.inboundChannelAdapter(cf, new ConsumerProperties("myTopic")),
e -> e.poller(Pollers.fixedDelay(5000)))
.handle(System.out::println)
.get();
}
@Bean
fun sourceFlow(cf: ConsumerFactory<String, String>) =
integrationFlow(Kafka.inboundChannelAdapter(cf,
ConsumerProperties(TEST_TOPIC3).also {
it.groupId = "kotlinMessageSourceGroup"
}),
{ poller(Pollers.fixedDelay(100)) }) {
handle { m ->
}
}
@InboundChannelAdapter(channel = "fromKafka", poller = @Poller(fixedDelay = "5000"))
@Bean
public KafkaMessageSource<String, String> source(ConsumerFactory<String, String> cf) {
ConsumerProperties consumerProperties = new ConsumerProperties("myTopic");
consumerProperties.setGroupId("myGroupId");
consumerProperties.setClientId("myClientId");
retunr new KafkaMessageSource<>(cf, consumerProperties);
}
<int-kafka:inbound-channel-adapter
id="adapter1"
consumer-factory="consumerFactory"
consumer-properties="consumerProperties1"
ack-factory="ackFactory"
channel="inbound"
message-converter="converter"
payload-type="java.lang.String"
raw-header="true"
auto-startup="false">
<int:poller fixed-delay="5000"/>
</int-kafka:inbound-channel-adapter>
<bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
<constructor-arg>
<map>
<entry key="max.poll.records" value="1"/>
</map>
</constructor-arg>
</bean>
<bean id="consumerProperties1" class="org.springframework.kafka.listener.ConsumerProperties">
<constructor-arg name="topics" value="topic1"/>
<property name="groupId" value="group"/>
<property name="clientId" value="client"/>
</bean>
请参阅 javadocs 以获取可用属性。
默认情况下,max.poll.records
必须在消费者工厂中显式设置,或者如果消费者工厂是 DefaultKafkaConsumerFactory
,则会强制设置为 1。您可以将属性 allowMultiFetch
设置为 true
来覆盖此行为。
您必须在 max.poll.interval.ms 内轮询消费者以避免重新平衡。如果您将 allowMultiFetch 设置为 true ,则必须处理所有检索到的记录,并在 max.poll.interval.ms 内再次轮询。 |
此适配器发出的消息包含一个 kafka_remainingRecords
头,其中包含上次轮询后剩余的记录计数。
从 6.2
版本开始,KafkaMessageSource
支持在消费者属性中提供的 ErrorHandlingDeserializer
。DeserializationException
会从记录头中提取并抛给调用方。对于 SourcePollingChannelAdapter
,此异常会被包装到 ErrorMessage
中并发布到其 errorChannel
。请参阅ErrorHandlingDeserializer
文档了解更多信息。
出站网关
出站网关用于请求/响应操作。它与大多数 Spring Integration 网关不同之处在于,发送线程不会在网关中阻塞,并且回复会在回复监听器容器线程上处理。如果您的代码通过同步Messaging Gateway 调用网关,则用户线程会在那里阻塞,直到接收到回复(或发生超时)。
KafkaProducerMessageHandler 的 sendTimeoutExpression
默认值是 delivery.timeout.ms
Kafka 生产者属性 + 5000,以便在超时后将实际的 Kafka 错误传播到应用,而不是由该框架生成的超时。此更改是为了保持一致性,因为您可能会遇到意外行为(Spring 可能超时 send()
,而实际上,最终会成功)。重要提示:该超时默认为 120 秒,因此您可能希望减小它以获得更及时的失败信息。
配置
以下示例展示了如何配置网关
-
Java DSL
-
Java
-
XML
@Bean
public IntegrationFlow outboundGateFlow(
ReplyingKafkaTemplate<String, String, String> kafkaTemplate) {
return IntegrationFlow.from("kafkaRequests")
.handle(Kafka.outboundGateway(kafkaTemplate))
.channel("kafkaReplies")
.get();
}
@Bean
@ServiceActivator(inputChannel = "kafkaRequests", outputChannel = "kafkaReplies")
public KafkaProducerMessageHandler<String, String> outGateway(
ReplyingKafkaTemplate<String, String, String> kafkaTemplate) {
return new KafkaProducerMessageHandler<>(kafkaTemplate);
}
<int-kafka:outbound-gateway
id="allProps"
error-message-strategy="ems"
kafka-template="template"
message-key-expression="'key'"
order="23"
partition-id-expression="2"
reply-channel="replies"
reply-timeout="43"
request-channel="requests"
requires-reply="false"
send-success-channel="successes"
send-failure-channel="failures"
send-timeout-expression="44"
sync="true"
timestamp-expression="T(System).currentTimeMillis()"
topic-expression="'topic'"/>
请参阅 javadocs 以获取可用属性。
请注意,使用的类与出站通道适配器相同,唯一的区别是传递给构造函数的 KafkaTemplate
是 ReplyingKafkaTemplate
。请参阅Spring for Apache Kafka 文档了解更多信息。
出站 Topic、分区、Key 等的确定方式与出站适配器相同。回复 Topic 的确定方式如下
-
名为
KafkaHeaders.REPLY_TOPIC
的消息头(如果存在,它必须是String
或byte[]
值)会根据模板的回复容器订阅的 Topic 进行验证。 -
如果模板的
replyContainer
只订阅了一个 Topic,则使用该 Topic。
您还可以指定 KafkaHeaders.REPLY_PARTITION
头来确定用于回复的特定分区。同样,这也将根据模板的回复容器的订阅进行验证。
或者,您也可以使用类似于以下 bean 的配置
@Bean
public IntegrationFlow outboundGateFlow() {
return IntegrationFlow.from("kafkaRequests")
.handle(Kafka.outboundGateway(producerFactory(), replyContainer())
.configureKafkaTemplate(t -> t.replyTimeout(30_000)))
.channel("kafkaReplies")
.get();
}
入站网关
入站网关用于请求/响应操作。
配置
以下示例展示了如何配置入站网关
-
Java DSL
-
Java
-
XML
@Bean
public IntegrationFlow serverGateway(
ConcurrentMessageListenerContainer<Integer, String> container,
KafkaTemplate<Integer, String> replyTemplate) {
return IntegrationFlow
.from(Kafka.inboundGateway(container, replyTemplate)
.replyTimeout(30_000))
.<String, String>transform(String::toUpperCase)
.get();
}
@Bean
public KafkaInboundGateway<Integer, String, String> inboundGateway(
AbstractMessageListenerContainer<Integer, String>container,
KafkaTemplate<Integer, String> replyTemplate) {
KafkaInboundGateway<Integer, String, String> gateway =
new KafkaInboundGateway<>(container, replyTemplate);
gateway.setRequestChannel(requests);
gateway.setReplyChannel(replies);
gateway.setReplyTimeout(30_000);
return gateway;
}
<int-kafka:inbound-gateway
id="gateway1"
listener-container="container1"
kafka-template="template"
auto-startup="false"
phase="100"
request-timeout="5000"
request-channel="nullChannel"
reply-channel="errorChannel"
reply-timeout="43"
message-converter="messageConverter"
payload-type="java.lang.String"
error-message-strategy="ems"
retry-template="retryTemplate"
recovery-callback="recoveryCallback"/>
请参阅 javadocs 以获取可用属性。
提供 RetryTemplate
后,传递失败会根据其重试策略进行重试。如果同时提供了 error-channel
,则在重试耗尽后,将使用默认的 ErrorMessageSendingRecoverer
作为恢复回调。您也可以使用 recovery-callback
来指定在这种情况下要采取的其他操作,或将其设置为 null
,将最终异常抛给监听器容器,以便在那里进行处理。
构建 ErrorMessage
(用于 error-channel
或 recovery-callback
)时,您可以通过设置 error-message-strategy
属性来自定义错误消息。默认情况下,使用 RawRecordHeaderErrorMessageStrategy
,以便访问已转换的消息以及原始的 ConsumerRecord
。
这种形式的重试是阻塞的,如果所有轮询记录的总重试延迟可能超过 max.poll.interval.ms 消费者属性,则可能导致重新平衡。相反,考虑向监听器容器添加一个 DefaultErrorHandler ,并将其配置为 KafkaErrorSendingMessageRecoverer 。 |
以下示例展示了如何使用 Java DSL 配置一个简单的转大写转换器
或者,您可以使用类似于以下代码的方式配置一个转大写转换器
@Bean
public IntegrationFlow serverGateway() {
return IntegrationFlow
.from(Kafka.inboundGateway(consumerFactory(), containerProperties(),
producerFactory())
.replyTimeout(30_000))
.<String, String>transform(String::toUpperCase)
.get();
}
您还可以使用用于 @KafkaListener
注解的容器工厂来创建 ConcurrentMessageListenerContainer
实例用于其他目的。请参阅Spring for Apache Kafka 文档和消息驱动通道适配器以获取示例。
基于 Apache Kafka Topic 的通道
Spring Integration 提供了基于 Apache Kafka Topic 的 MessageChannel
实现,用于持久化。
每个通道都需要一个用于发送端的 KafkaTemplate
,以及一个监听器容器工厂(用于可订阅通道)或一个 KafkaMessageSource
(用于可轮询通道)。
Java DSL 配置
-
Java DSL
-
Java
-
XML
@Bean
public IntegrationFlow flowWithSubscribable(KafkaTemplate<Integer, String> template,
ConcurrentKafkaListenerContainerFactory<Integer, String> containerFactory) {
return IntegrationFlow.from(...)
...
.channel(Kafka.channel(template, containerFactory, "someTopic1").groupId("group1"))
...
.get();
}
@Bean
public IntegrationFlow flowWithPubSub(KafkaTemplate<Integer, String> template,
ConcurrentKafkaListenerContainerFactory<Integer, String> containerFactory) {
return IntegrationFlow.from(...)
...
.publishSubscribeChannel(pubSub(template, containerFactory),
pubsub -> pubsub
.subscribe(subflow -> ...)
.subscribe(subflow -> ...))
.get();
}
@Bean
public BroadcastCapableChannel pubSub(KafkaTemplate<Integer, String> template,
ConcurrentKafkaListenerContainerFactory<Integer, String> containerFactory) {
return Kafka.publishSubscribeChannel(template, containerFactory, "someTopic2")
.groupId("group2")
.get();
}
@Bean
public IntegrationFlow flowWithPollable(KafkaTemplate<Integer, String> template,
KafkaMessageSource<Integer, String> source) {
return IntegrationFlow.from(...)
...
.channel(Kafka.pollableChannel(template, source, "someTopic3").groupId("group3"))
.handle(..., e -> e.poller(...))
...
.get();
}
/**
* Channel for a single subscriber.
**/
@Bean
SubscribableKafkaChannel pointToPoint(KafkaTemplate<String, String> template,
KafkaListenerContainerFactory<String, String> factory)
SubscribableKafkaChannel channel =
new SubscribableKafkaChannel(template, factory, "topicA");
channel.setGroupId("group1");
return channel;
}
/**
* Channel for multiple subscribers.
**/
@Bean
SubscribableKafkaChannel pubsub(KafkaTemplate<String, String> template,
KafkaListenerContainerFactory<String, String> factory)
SubscribableKafkaChannel channel =
new SubscribableKafkaChannel(template, factory, "topicB", true);
channel.setGroupId("group2");
return channel;
}
/**
* Pollable channel (topic is configured on the source)
**/
@Bean
PollableKafkaChannel pollable(KafkaTemplate<String, String> template,
KafkaMessageSource<String, String> source)
PollableKafkaChannel channel =
new PollableKafkaChannel(template, source);
channel.setGroupId("group3");
return channel;
}
<int-kafka:channel kafka-template="template" id="ptp" topic="ptpTopic" group-id="ptpGroup"
container-factory="containerFactory" />
<int-kafka:pollable-channel kafka-template="template" id="pollable" message-source="source"
group-id = "pollableGroup"/>
<int-kafka:publish-subscribe-channel kafka-template="template" id="pubSub" topic="pubSubTopic"
group-id="pubSubGroup" container-factory="containerFactory" />
消息转换
提供了 StringJsonMessageConverter
。请参阅Spring for Apache Kafka 文档了解更多信息。
将此转换器与消息驱动通道适配器一起使用时,您可以指定希望将入站 Payload 转换为何种类型。这通过在适配器上设置 payload-type
属性 (payloadType
property) 来实现。以下示例展示了如何在 XML 配置中进行设置
<int-kafka:message-driven-channel-adapter
id="kafkaListener"
listener-container="container1"
auto-startup="false"
phase="100"
send-timeout="5000"
channel="nullChannel"
message-converter="messageConverter"
payload-type="com.example.Thing"
error-channel="errorChannel" />
<bean id="messageConverter"
class="org.springframework.kafka.support.converter.MessagingMessageConverter"/>
以下示例展示了如何在 Java 配置中在适配器上设置 payload-type
属性 (payloadType
property)
@Bean
public KafkaMessageDrivenChannelAdapter<String, String>
adapter(KafkaMessageListenerContainer<String, String> container) {
KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenChannelAdapter =
new KafkaMessageDrivenChannelAdapter<>(container, ListenerMode.record);
kafkaMessageDrivenChannelAdapter.setOutputChannel(received());
kafkaMessageDrivenChannelAdapter.setMessageConverter(converter());
kafkaMessageDrivenChannelAdapter.setPayloadType(Thing.class);
return kafkaMessageDrivenChannelAdapter;
}
空 Payload 和日志压缩 '墓碑' 记录
Spring Messaging Message<?>
对象不能有 null Payload。当您使用 Apache Kafka 的端点时,null Payload(也称为墓碑记录)由类型为 KafkaNull
的 Payload 表示。请参阅Spring for Apache Kafka 文档了解更多信息。
Spring Integration 端点的 POJO 方法可以使用真实的 null
值代替 KafkaNull
。为此,请使用 @Payload(required = false)
标记参数。以下示例展示了如何做到这一点
@ServiceActivator(inputChannel = "fromSomeKafkaInboundEndpoint")
public void in(@Header(KafkaHeaders.RECEIVED_KEY) String key,
@Payload(required = false) Customer customer) {
// customer is null if a tombstone record
...
}
从 KStream
调用 Spring Integration 流
您可以使用 MessagingTransformer
从 KStream
调用集成流
@Bean
public KStream<byte[], byte[]> kStream(StreamsBuilder kStreamBuilder,
MessagingTransformer<byte[], byte[], byte[]> transformer) transformer) {
KStream<byte[], byte[]> stream = kStreamBuilder.stream(STREAMING_TOPIC1);
stream.mapValues((ValueMapper<byte[], byte[]>) String::toUpperCase)
...
.transform(() -> transformer)
.to(streamingTopic2);
stream.print(Printed.toSysOut());
return stream;
}
@Bean
@DependsOn("flow")
public MessagingTransformer<byte[], byte[], String> transformer(
MessagingFunction function) {
MessagingMessageConverter converter = new MessagingMessageConverter();
converter.setHeaderMapper(new SimpleKafkaHeaderMapper("*"));
return new MessagingTransformer<>(function, converter);
}
@Bean
public IntegrationFlow flow() {
return IntegrationFlow.from(MessagingFunction.class)
...
.get();
}
当集成流以接口开始时,创建的代理 bean 名称是流 bean 名称加上 ".gateway",因此如果需要,可以使用此 bean 名称作为 @Qualifier
。
read/process/write 场景的性能考量
许多应用程序从一个 topic 消费,执行一些处理,然后写入另一个 topic。在大多数情况下,如果写入操作失败,应用程序会希望抛出异常,以便可以重试传入的请求和/或将其发送到死信 topic。此功能由底层消息监听器容器以及适当配置的错误处理器支持。然而,为了支持这一点,我们需要阻塞监听器线程,直到写入操作成功(或失败),以便可以将任何异常抛给容器。在消费单个记录时,这通过在出站适配器上设置 sync
属性来实现。但是,在消费批量数据时,使用 sync
会导致显著的性能下降,因为应用程序会等待每次发送的结果,然后才发送下一条消息。您也可以执行多次发送,然后在之后等待这些发送的结果。这通过向消息处理器添加 futuresChannel
来实现。要启用此功能,请将 KafkaIntegrationHeaders.FUTURE_TOKEN
添加到出站消息中;然后可以使用它将 Future
与特定发送的消息关联起来。以下是如何使用此功能的一个示例
@SpringBootApplication
public class FuturesChannelApplication {
public static void main(String[] args) {
SpringApplication.run(FuturesChannelApplication.class, args);
}
@Bean
IntegrationFlow inbound(ConsumerFactory<String, String> consumerFactory, Handler handler) {
return IntegrationFlow.from(Kafka.messageDrivenChannelAdapter(consumerFactory,
ListenerMode.batch, "inTopic"))
.handle(handler)
.get();
}
@Bean
IntegrationFlow outbound(KafkaTemplate<String, String> kafkaTemplate) {
return IntegrationFlow.from(Gate.class)
.enrichHeaders(h -> h
.header(KafkaHeaders.TOPIC, "outTopic")
.headerExpression(KafkaIntegrationHeaders.FUTURE_TOKEN, "headers[id]"))
.handle(Kafka.outboundChannelAdapter(kafkaTemplate)
.futuresChannel("futures"))
.get();
}
@Bean
PollableChannel futures() {
return new QueueChannel();
}
}
@Component
@DependsOn("outbound")
class Handler {
@Autowired
Gate gate;
@Autowired
PollableChannel futures;
public void handle(List<String> input) throws Exception {
System.out.println(input);
input.forEach(str -> this.gate.send(str.toUpperCase()));
for (int i = 0; i < input.size(); i++) {
Message<?> future = this.futures.receive(10000);
((Future<?>) future.getPayload()).get(10, TimeUnit.SECONDS);
}
}
}
interface Gate {
void send(String out);
}