Spring集成- Kafka消息驱动通道-自动确认
Spring集成是指使用Spring框架来集成不同的技术或组件，以实现更高效、更便捷的开发方式。Kafka消息驱动通道是Spring集成中的一种方式，它利用Kafka作为消息中间件，实现了异步消息传递和解耦的目的。自动确认是指消费者在成功处理完消息后，自动向Kafka发送确认消息，告知Kafka该消息已被消费。
Kafka是一个高吞吐量的分布式发布订阅消息系统，具有持久化、可扩展、高可靠性等特点。它适用于构建实时流数据管道、大数据处理、日志收集和传输等场景。
使用Kafka消息驱动通道可以带来以下优势：
- 异步通信：生产者和消费者之间通过消息队列进行异步通信，提高系统的响应速度和并发能力。
- 解耦：生产者和消费者之间通过消息队列解耦，使得系统各个模块之间的依赖性降低，提高系统的可维护性和可扩展性。
- 可靠性：Kafka具有高可靠性和持久化特性，能够保证消息的可靠传输和存储。
- 扩展性：Kafka支持分布式部署，可以根据业务需求进行水平扩展，满足高并发场景的需求。
在实际应用中，Kafka消息驱动通道可以应用于以下场景：
- 日志收集和分析：通过将日志消息发送到Kafka消息队列，实现实时的日志收集和分析，帮助开发人员快速定位和解决问题。
- 异步任务处理：将耗时的任务封装成消息，通过Kafka消息队列异步处理，提高系统的并发能力和吞吐量。
- 分布式系统解耦：将不同模块之间的通信通过Kafka消息队列进行解耦，提高系统的可维护性和可扩展性。
- 实时数据处理：将实时产生的数据通过Kafka消息队列传输到各个消费者进行实时处理，如实时统计、实时监控等。
腾讯云提供了一系列与Kafka相关的产品和服务，包括云原生消息队列 CMQ、消息队列 CKafka 等。您可以通过以下链接了解更多信息：
通过Spring集成Kafka消息驱动通道，可以实现高效、可靠的消息传递和解耦，提升系统的性能和可维护性。
相关·内容
我正在使用入站网关使用SOAP web服务，我们需要将消息放在kafka主题中，并使用spring集成的声明性方式向请求者返回同步确认。这是可能的吗？
public Acknowledgement process(@RequestPayload MessagePayload payload) {
// perform validation & logic
// need to send message to kafka topic using declarative way
// sending synchronous ack to request o
浏览 0提问于2017-01-06得票数 0
1回答
：入站通道适配器与消息驱动适配器
、、、、
我刚开始使用春季集成卡夫卡。我当时正在读取文档，在入站通道适配器和消息驱动的通道适配器之间出现了混淆。有人能解释他们之间的区别吗。同样，卡夫卡的spring集成也有一些错误处理机制，比如amqp/rmq(使用dlq)的spring集成机制。
PS :我们正试图从兔子mq转移到kafka，作为in集成框架中的消息来源。
谢谢,
浏览 4提问于2019-10-05得票数 0
回答已采纳
我有以下几点：
[inbound channel adapter] -> ... -> foo -> [outbound channel adapter] -> bar
我如何编写我的spring集成应用程序，以便foo可以使用一个额外的对象，该对象不是[outbound channel adapter]要消费的消息的一部分，这样bar就可以得到它？
我的应用程序基本上是从AWS SQS接收消息(使用spring-integration-aws)，执行一些过滤/转换，然后将消息发布到Apache Kafka (使用spring-integration-kafka)，并且
浏览 54提问于2018-12-15得票数 1
回答已采纳
1回答
我目前正在尝试编写一个适配器，它将使用来自ActiveMQ的消息并将其发布到Kafka。
我正在考虑使用spring集成来集成这两个消息传递系统。
我的问题是，我的应用程序不会维护模型的注册表，许多应用程序将使用这些模型将记录发布到activeMQ。我想要接收这些javax消息，并想执行一些转换，比如将jmscorrelationId添加到kafka消息中。
ALso，另一个要求是仅当kafka发送/发布成功时才向活动mq发送确认。
是否可以使用spring集成将ack发送回activemq？
spring集成是个不错的选择吗？
请注意，我的技术架构师不赞成使用Camel/Mule。此外，他不
浏览 10提问于2017-07-05得票数 0
回答已采纳
我正在使用spring -cloud-stream:1.3.0开发spring启动应用程序。messages，spring-cloud-stream-binder-kafka:1.3.0.RELEASE.我使用spring集成dsl来拆分文件中的行，并使用beanio将行转换为json，要求将成功的json消息写入kafka主题，并将错误消息写入不同的kafka主题。下面是application.yml中的配置。
spring:
cloud:
stream:
kafka:
binder:
autoAddPartitions: true
浏览 2提问于2017-10-19得票数 1
我使用和功能接口来处理我的kafka处理。
如果我按顺序进行处理，如果我杀死了应用程序，它将返回用于处理的消息，该消息可以按预期的方式工作，因为没有消息下降。然而，如果我试图做平行，它似乎是承认卡夫卡，这是可以理解的，因为它现在是一个单独的线程，因此想要转向手动承认。
我的守则：
application.yml (相关部分)
spring:
cloud:
stream:
kafka:
binder:
brokers: localhost:9092
autoAddPartitions: true
浏览 1提问于2020-04-22得票数 0
回答已采纳
我正在尝试在spring集成中实现一个自定义的入站通道适配器，以使用来自的消息。基于spring集成示例，我发现我需要创建一个实现MessageSource接口的类，并实现从kafka返回消费消息的receive()方法。但是基于，KafkaStream中的消息迭代器是由BlockingQueue支持的。因此，如果队列中没有消息，线程将被阻塞。
那么，实现接收()方法的最佳方式是什么，因为该方法可能会阻塞，直到有东西可用为止。？
在更一般的意义上，我们如何为流消息源实现一个自定义的入站通道，该通道会一直阻塞，直到有东西可以使用？
浏览 1提问于2013-02-01得票数 0
回答已采纳
2回答
我正在使用下面的配置来集成activemq和kafka。我收到了activemq的消息，并把它转发给了kafka。但是，我注意到消息正在从JMS队列中退出队列，但是消息不会传递给kafka。
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:j
浏览 1提问于2017-07-08得票数 1
回答已采纳
我正在使用spring集成框架，并使用一个转换器
inputChannel ->卡夫卡消费者
outputChannel ->数据库jdbc编写器
@Bean
public DirectChannel inboundChannel() {
return new DirectChannel();
}
@Bean
public DirectChannel outboundChannel() {
return new DirectChannel();
}
@Bean
@Transformer(inputChannel="inbou
浏览 1提问于2021-04-21得票数 0
回答已采纳
我正在努力配置一个“自定义”ObjectMapper，以供变压器使用。我收到一个java.time.Instant json表示，我希望解析为对象属性。即：
{“类型”：“测试”，“源”：“测试”，“时间戳”：{“epochSecond”：1454503381，“纳米”：335000000}}
这条消息是一条卡夫卡消息，它提出了一个问题:我是否应该编写一个实现kafka编码器/解码器的自定义序列化程序，以便能够将Kafka消息转换为正确的对象，或者弹簧集成必须自动完成这一任务？
fw/依赖关系和版本：
弹簧启动-1.3.2
Spring集成Java Dsl - 1.1.1.RELEASE
Fa
浏览 1提问于2016-02-03得票数 2
回答已采纳
我正在尝试设置一个场景，监听器将手动向RabbitMQ发送确认。我的application.properties里有这个
spring.cloud.stream.rabbit.bindings.input.consumer.acknowledge-mode=MANUAL
如何从监听程序发送ack？我从中看到，对于Kafka，我们创建了一个确认对象，并在其中调用object ()。RabbitMQ有没有这样做的例子？
浏览 3提问于2016-08-20得票数 4
1回答
我有集成项目的wit设置如下。
Application1(Spring集成) --> AMQ --> Mirth --> Internet --> Mirth --> AMQ --> (Spring集成)应用2
上面的流消息是以两种方式传递的，我想实现这样的东西。
当我从应用程序1发起请求消息(出站)时，它会发送到应用程序2(作为入站)，并将零对多消息(出站)发送到应用程序1。我的要求是保持等待所有响应并执行某些操作。
在愉悦的spring中有支持这一点的特性吗？
浏览 1提问于2017-07-18得票数 0
1回答
线程配置
、
我使用了SpringIntegration-Kafka1.1.0和下面的配置。我不太理解streams配置。当我增加这个值时，Spring会自动生成更多的线程来处理消息吗？例如，当我有streams=2时，相关的转换器和服务激活器是否都在两个线程中运行？我想错过一些线程执行器配置，但不确定怎么做。任何暗示都是感激的。谢谢。
<int:poller default="true" fixed-delay="10"/>
<int:channel id="tag.track">
</int:channel>
&
浏览 2提问于2015-04-17得票数 2
1回答
我正在做一个与spring启动，spring批处理和spring集成有关的项目。
我已经配置了spring集成，以便在新消息到达时启动一个春季批处理作业。
我向附加到JobLaunchingGateway的spring集成通道发送一条消息，对于每一条消息，JobLaunchingGateway尝试启动一个新的TaskExecutor。
让通道由一个持久队列(例如ActiveMQ )支持，让任务执行器池大小等于2。
我想配置系统，以便在已经使用了执行器池大小时，新消息不会被JobLaunchingGateway使用，而是保留在持久队列上。
有可能吗？有没有什么最佳做法。
如有任何反馈，将不胜感激
浏览 2提问于2016-10-28得票数 0
回答已采纳
3回答
我用一个分区构建了一个kafka主题。
kafka-topics --create --topic files.write --if-not-exists --zookeeper zookeeper:32181 --partitions 1 --replication-factor 1
可以在这个主题中推送许多信息。
但是，我希望单个使用者(对于给定的组)逐一处理这些消息。
spring:
application:
name: file-consumer
cloud:
stream:
kafka:
binder:
type
浏览 0提问于2019-01-30得票数 1
回答已采纳
2回答
我有以下简化的spring集成流程：
int-ws:inbound-gateway ----> int:transformer ----> int-kafka:outbound-channel-adapter
基本上：
使用int-ws:inbound-gateway公开web服务端点。
来自此端点的消息被放入input通道(第一个--->)。
自定义转换器转换有效载荷JSON格式并添加MESSAGE_KEY报头(卡夫卡需要)
该消息被放置到inputToKafka通道(第二个--->)上。
int-kafka:outbound-channel-ad
浏览 13提问于2015-06-26得票数 2
回答已采纳
我想使用spring集成来消费来自activemq的消息。然后根据我从activemq收到的消息调用作业。
有谁能帮帮我吗？
浏览 0提问于2017-05-30得票数 0
2回答
我希望只在消息处理成功时使用spring云流手动提交偏移量。这是我的代码- application.yml & Handler类
public void process(Message<?> message) {
System.out.println(message.getPayload());
Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
浏览 3提问于2021-02-08得票数 2
回答已采纳
我们使用Spring Cloud Stream Kafka Binder (带项目反应堆集成，即Flux streams)和manual offset commits (即autoCommitOffset = false)。 我们正在尝试编写一个与Embedded Kafka from spring-kafka-test的集成测试，通过手动读取消费者组偏移量，使用admin client，在测试向我们的主题发送消息之前和之后，断言所有这些都是有效的。 测试间歇性地失败。使用awaitility，我们现在等待10秒来轮询偏移量，这似乎绕过了我们的大多数问题，因为偏移量将在大约7秒后发生变化-但
浏览 15提问于2019-07-01得票数 1
回答已采纳
1回答
我有一个REST控制器，它调用带有@MessagingGateway(errorChannel = ERROR_CHANNEL)注释的网关
这样，无论在网关启动的集成流下游发生什么错误，都会流到一个由另一个集成流处理的错误通道中，这是按预期工作的。
现在，还有另一个场景，集成流从Kafka读取消息，将这些消息路由到另一个通道，再一个集成流处理这些消息，另一个流向远程服务发送HTTP请求。
public IntegrationFlowBuilder attachmentEventTenantRouter(String tenantId) {
return attachmentEventB
浏览 1提问于2020-08-12得票数 0
回答已采纳
我试图使用手动确认从一个主题消费多个消息，但如果所有消息都只有一次确认，则确认工作。
@KafkaListener(
id = "${kafka.buyers.product-sales-pricing.id}",
topics = "${kafka.buyers.product-sales-pricing.topic}",
groupId = "${kafka.buyers.group-id}",
concurrency = "${kafka.buyers.concurrency}"
)
publ
浏览 4提问于2019-08-01得票数 1
1回答
我有一个简单的Spring应用程序，它从Kafka主题读取并将消息保存到一些缓存中。我想添加一个集成测试，它将启动我的原始应用程序，生成一些来自嵌入式Kafka的消息，然后断言缓存内容。
我在挣扎于“启动我的原始应用程序”部分。如何在Spring集成测试中做到这一点？我试过这样做：
@RunWith(SpringRunner.class)
@SpringBootTest(classes = OriginalApplication.class)
@EmbeddedKafka
public class OriginalApplicationIntegrationTest {
@Test
浏览 1提问于2020-01-08得票数 0
回答已采纳
1回答
泉云流卡夫卡误码信道
、、、
我正在尝试设置一个绑定，以便将Kafka消息从Spring转发到自定义通道(用于集中化错误处理)。
错误消息被发送到配置的通道，但它们是以GenericMessage的形式到达的，byte[]有效负载由异常详细信息和失败消息组成。
我的配置：
spring:
cloud:
stream:
kafka:
bindings:
accountOut.producer:
sync: true
binder:
autoCreateTopics: false
head
浏览 0提问于2018-03-09得票数 0
回答已采纳
我正在发送有关主题的消息，并且可以使用kafka控制台和Java消费者代码使用main方法读取相同的消息。但是消息并没有从使用者配置xml中读取。请帮助解决这个问题。我正在使用消息int-kafka:消息驱动-通道适配器。春季-卡夫卡1.3.0版
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://ww
浏览 2提问于2015-12-26得票数 1
回答已采纳
1回答
我需要实现以下架构：
我有大量订单，必须拆分成较小的顺序(并行)，并发送到下游异步rest端点。
在使用相关ids完成订单(失败或成功)之后，向下流排序API将消息发布到应答队列( kafka/rabbitmq)。
需要有聚合侦听器来收集所有的响应，并将最终输出发送给调用者。
我正在考虑使用spring集成、散射聚集模式和其他有用的Spring特性。
你能帮我举个例子，说明如何在Spring集成的帮助下实现这样的架构吗？
浏览 5提问于2021-12-12得票数 0
我们使用spring-integration-kafka version3.1.2.RELEASE和int- kafka :message-driven-channel-adapter来消费来自远程kafka主题的消息。生产者发送加密的消息，我们使用反序列化程序解密实际的消息。我们可以使用该主题中发布的所有消息。我们已经使用了自动提交作为false。我们想知道如何在成功处理消息后提交或确认来自我们的服务的消息。有人可以帮助我们如何提交从消息驱动通道读取的消息，并提供一些参考实现吗？
当我们将自动提交设置为true时，我们假设它将在提交间隔之后提交消息，但我们希望在我们的服务中处理它。我在下面的
浏览 1提问于2019-05-19得票数 0
1回答
我正在使用与kafka经纪公司的春云流进行微服务互传。作为其中的一部分，stream bridge将用于发送消息，这是很好的。
--但是在使用所述消息期间，不需要立即使用该消息，而是当满足条件时，只需要使用。
从文档中，我了解到我需要在这方面使用Polled Consumers(如果我弄错了就纠正我)。
这就是我对的理解所做的尝试。
application.properties
spring.cloud.stream.pollable-source = consumeResponse
spring.cloud.stream.function.definition = consumeRespon
浏览 7提问于2022-03-10得票数 0
回答已采纳
目前我在同一个Spring Boot应用程序中配置了生产者和消费者，但奇怪的是，触发消息的Spring云流并没有通过Kafka (我使用kafka-console- consumer来监控消息)，但是消费者仍然接收消息(使用与生产者相同的线程)。 如果我移除应用程序中的consumerHandler (@StreamListener)，生产者会成功地将消息发送到Kafka。 对此有什么配置吗？我需要Spring云流默认向Kafka发送消息。 生产者和消费者配置： @Component
public interface NotificationProcessor {
Strin
浏览 30提问于2021-03-21得票数 1
回答已采纳
这是I need current easy to follow instructions for configuring spring integration kafka from XML的后续问题 Spring-integration-kafka在过去的几次迭代中发展了很多，许多旧的例子不再起作用。 特别是，这个从spring-batch世界到spring集成世界的bean将不会实例化，因为KafkaTemplate类没有实现MessagingTemplate。当前推荐的完成此集成的方式是什么？ <bean id="partitionHandler" class=&#
浏览 57提问于2019-03-21得票数 0
回答已采纳
是否有方法在使用amqp发布消息后继续使用spring集成流:出站-通道-适配器？-使用确认- ack -通道，我只能在amqp ack之后才能继续该流程，这需要太长时间。我在试着看看我是否不在乎这些人，我能不能继续这样的流动
浏览 1提问于2016-11-15得票数 1
回答已采纳
我的春季集成应用程序需要能够通过简单的配置更改在Kafka和遗留消息传递库(tibco rendezvous，spring集成不提供任何默认出站网关实现)之间按需切换。
遗留消息传递库提供了一些基本的请求/应答方法。
Class LegacyTransport {
Object request(Object query,String topic);
}
我正在试图找出什么是抽象消息出站网关(Kafka和旧版)的最佳方法，这样我就可以在我的主IntegrationFlow中(通过简单的配置更改)将其中一个替换为另一个。
我目前的想法是使用下面的方法作为我的主要IntegrationFlo
浏览 3提问于2020-06-03得票数 0
回答已采纳
在使用@StreamListener收听kafka主题之后，在RuntimeException、全局erroChannel或主题特定errorChannel (topic.group.errors)上没有收到任何错误消息。@ServiceActivator没有收到任何信息。
POM依赖项: Greenwich.RELEASE
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactI
浏览 1提问于2019-06-07得票数 0
回答已采纳
我在我的应用程序中使用Spring Boot Java。应用程序正在使用MessageChannel向Kafka发送一些消息。例如:this.messageChannel.send(消息)。 我想在集成测试中模拟这个消息通道调用，这样消息就不会发送到Kafka。目前正在使用Wiremock进行Spring集成测试。 请给我建议模拟MessageChannel发送方法的方法
浏览 27提问于2021-08-06得票数 0
使用Spring for Apache Kafka或Spring AMQP，我可以实现消息发布/订阅。Spring Cloud Bus使用kafka/rabbitmq来做大致相同的事情，它们之间有什么区别？
浏览 11提问于2018-08-01得票数 4
在基于服务的应用程序中，我们使用<int-amqp:publish-subscribe-channel/>作为一种事件总线。send方法和消息处理程序基于spring消息传递中的Message类(从SpringIntegration4.0 (+)开始)。事件是对需要被其他服务获取的实体的更改。
问题是:spring将spring消息传递Message类视为任意对象有效负载，因为它不被识别为spring Message。这造成以下问题：
默认消息格式是序列化的Java对象。Spring不仅序列化了原始的有效负载对象，而且还序列化了Spring消息传递Message，它在Sprin
浏览 3提问于2015-02-26得票数 1
回答已采纳
1回答
弹簧靴卡夫卡手册失败
、、、
我有一个Spring消费者的下面的配置。我在尝试手动确认而不是自动提交。在手动确认后，我开始出错。Spring版本为2.7.2。
kafka.consumer.groupId=mcs-ccp-event
message.topic.name=mcs_ccp_test
kafka.bootstrapAddress=kafka-dev-app-a1.com:9092
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.listener.ack-mode=MANUAL_IMMEDIATE
和这种使用者配置
@EnableKafk
浏览 26提问于2022-08-06得票数 0
回答已采纳
有一个简单的设置。
接口
public interface MyKafkaBinding {
@Output(PUBLISHER)
MessageChannel publisher();
@Input("subscriber")
SubscribableChannel subscriber();
}
绑定
@EnableBinding(MyKafkaBinding.class)
听者
@StreamListener(MyKafkaBinding.PUBLISHER)
public void listen(MyEvent message) {
浏览 1提问于2020-01-03得票数 2
回答已采纳
我已经使用了spring io文档中列出的示例配置，它工作得很好。
<int-kafka:message-driven-channel-adapter
id="kafkaListener"
listener-container="container1"
auto-startup="false"
phase="100"
send-timeout="5000"
channel="nullChannel"
浏览 4提问于2017-08-07得票数 0
1回答
我正在努力实现以下目标：
使用SpringCloudStream2.1.3.RELEASE与Kafka绑定器一起向输入通道发送消息，并实现发布订阅行为，其中每个用户都将得到通知，并能够处理发送给Kafka主题的消息。
我了解到，在卡夫卡，如果每个消费者都属于自己的消费群体，就能从一个话题中读出每一条信息。在我的示例中，spring为运行中的spring引导应用程序的每个实例创建了一个匿名的唯一使用者组。spring引导应用程序只有一个流侦听器配置为侦听输入通道。
测试示例：
使用一个与Kafka主题绑定的输入通道配置了一个示例应用程序。使用spring控制器向输入通道发送消息，希望该消息将传递
浏览 0提问于2019-08-22得票数 1
我正在尝试将读取消息从Quickfix(读取修复消息)配置到spring集成中。我知道我可以使用inbound-channel-adapter从外部源读取数据。您能提供一个如何编写事件驱动的入站通道适配器的示例吗？我有以下不起作用的配置
<bean id="QuickFixClient" class="com.limebrokerage.quickfix.QuickFixClient" >
<constructor-arg index="0" value= "/home/sbansal/workspace/bo/ta
浏览 1提问于2014-04-17得票数 0
我想在向kafka topic发出消息后，获取偏移量和分区信息。我通读了spring cloud stream kafka绑定文档，发现这可以通过使用RECORD_METADATA kafka头来实现。
来自Spring文档：()
recordMetadataChannel成功发送结果应发送到的MessageChannel的bean名称；该bean必须存在于应用程序上下文中。发送到通道的消息是具有附加标头KafkaHeaders.RECORD_METADATA的已发送消息(转换后，如果有)。头部包含一个由Kafka客户端提供的RecordMetadata对象；它包括在主题中写入记录的分区和偏移
浏览 0提问于2020-09-23得票数 1
我的要求是，当应用程序无法处理从通过spring boot java应用程序使用的kafka主题的当前偏移量读取的消息时，重置kafka主题的偏移量。在手动重置偏移量或发送否定确认之后，需要通过spring boot java应用程序的kafka消费者再次从不合适的偏移量值轮询消息。它能实现吗？
浏览 3提问于2019-01-09得票数 1
我最近才开始学习，因为我需要替换MDB(J2EE)应用程序。
该应用程序主要由MDB组成，实现了MDB的拆分、聚合和调度。我认为这是使用spring集成的完美标准。
我尝试了一些JMS示例，并试图部署它，但无法理解如何使用jms-入站网关来替换MDB。
有办法这样做吗？还是唯一的选择仍然是使用MDB并从MDB的onMessage调用spring集成服务？
浏览 1提问于2014-02-28得票数 0
回答已采纳
1回答
我正在使用maprstream与spring集成，并试图创建一个发布者来发送消息到maprstream主题。我正在使用下面提到的的Jar版本兼容性矩阵。
弹簧-集成-kafka-2.0.1.
Spring-Kafka -1.0.3
Kafka-客户- 0.9.0.0-mapr-1607
正如在spring integration 中提到的，如果我使用的是，我应该能够在KafkaProducerMessageHandler中设置属性‘同步’，但是我会遇到模式验证问题，说明“同步”在KafkaProducerMessageHandler中是不需要的。
有人能帮我一下吗？
浏览 6提问于2017-12-01得票数 0
回答已采纳
kafka集成项目版本1.15由于兼容的issues.How使用发送消息到kerberos的kafka，我们没有使用spring boot .how来设置属性来发送消息给kafka?我试图查找，但在ProducerConfig中找不到与kerberos相关的属性。
浏览 1提问于2017-06-07得票数 1
2回答
使用Kafka作为微服务体系结构中的消息传递系统，使用spring相对于spring云流+spring-云-初学者流-kafka有什么好处？
spring云流框架支持更多的消息传递系统，因此具有更模块化的设计。但是功能呢？泉-卡夫卡的功能和弹簧云流+弹簧云-启动-流-卡夫卡的功能之间有差距吗？哪个API设计得更好？
期待阅读你的观点
浏览 5提问于2017-11-15得票数 56
1回答
我是卡夫卡的新手。我们正在与我们的应用程序进行Kafka集成。问题是，在使用时，假设在处理特定记录时发生错误，那么如何处理这种情况。
我使用的是spring kafka。
如果重试该特定记录是解决方案，那么我想单独进行。例如，如果我有偏移量1到10，偏移量4的记录给出错误，那么偏移量5到10被正常消耗，10分钟后在偏移量4的记录被重试。
我检查了一些链接，但它们没有什么帮助。
任何线索/解决方案都将受到高度赞赏。
浏览 0提问于2018-01-20得票数 4
我正在尝试测试从消息驱动的通道适配器开始的Spring集成流，该适配器配置为：
<int-jms:message-driven-channel-adapter id="myAdapter" ... />
我的测试结果如下：
@SpringJUnitConfig(locations = {"my-app-context.xml"})
@SpringIntegrationTest(noAutoStartup = {"myAdapter"})
public class MyIntegrationFlowTest {
@Autowire
浏览 15提问于2022-01-18得票数 0
回答已采纳
2回答
在Spring集成中，是否有可能保持通道同步(发送消息后获得确认)，但同时处理更多消息(并行处理)，而不使用线程(即ExecutorService执行和提交工作人员)创建自己的代码并等待它们？我想上传文件通过FTP使用，但同时上传更多，而不是在代码中创建自己的线程。我需要知道何时上传所有文件(这就是为什么我希望它是同步的)。是否有可能通过Spring集成配置，如果是的话，如何实现？
浏览 1提问于2014-09-09得票数 5
回答已采纳
可以使用原生Kafka KafkaConsumer.poll ()批量获取消息。集成spring cloud starter stream Kafka (3.0.10.发布)？我不懂公文希望举个例子，谢谢
浏览 56提问于2020-12-17得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
