拆分后，所有消息都放到Kafka上，如何执行方法？ 我试过在channel之后使用handle()，但在第二条消息中没有发布到Kafka上。.); flowBuilder.channel(messageChannel); 我想在将所有的拆分消息写到Kafka之后<e
浏览 6提问于2019-05-09得票数 2
回答已采纳
我们让多个应用程序使用者收听相同的kafka主题，并且生产者在向主题发送消息时设置消息头，以便特定实例能够评估标头并处理消息。TestObj message) {}
在SpringCloudStream3.0.0中，@StreamListener被废弃了，我无法在函数中找到与
浏览 11提问于2021-09-16得票数 5
我正在开发一个Spring Boot应用程序，它使用以Kafka主题为源的Spring集成流。这些配置为通过云配置和spring.cloud.stream.kafka.bindings从Kafka读取。
当应用程序第一次启动时，它会立即开始读取Kafka主题。这是一个问题，因为应用程序需要初始化一些本地的、不可持久的数据库，然后才能开始正确处理传入的Kafka消息
浏览 0提问于2018-04-24得票数 1
messages，spring-cloud-stream-binder-kafka:1.3.0.RELEASE.我使用spring集成dsl来拆分文件中的行，并使用beanio将行转换为json，要求将成功的json消息写入kafka主题，并将错误消息写入不同的kafka主题。:1.3.0.RELEASE，中有一个拉取请求，它在errorChannelEnabled设置为true时
浏览 2提问于2017-10-19得票数 1
我有一个从DB读取记录并传递到spring网关(Spring集成)的主类，从那里我将消息拆分成一个带有队列的多线程应用程序。我的Sping应用程序正在关闭上下文，而消息仍在由spring集成适配器处理。
需要一个机制，只关闭上下文后，所有的消息都得到了处理。作为一种临时解决办法，我在while循环中使用thre
浏览 33提问于2018-01-11得票数 0
我有一个模型对象，它是经过几次转换和解析后填充的。现在，我需要使用spring集成将模型中的message属性发送给kafka。我可以使用messageKey方法构造密钥，但如何从m.getPayload()、.getMessage()等模型获取实际消息并将其发送给kafka。.bridge(e -> e.order(Ordered.HIGHEST_PRECEDENCE))
浏览 4提问于2019-02-20得票数 0
1回答
修复了任务执行器的延迟
、、、、
我的项目有一个-spring集成固定延迟轮询器(5分钟)其获得一串记录-将其发送到拆分器，拆分器将记录拆分并在输出通道将它们发送到池大小为2的任务执行器路由器通道，并且发送每个记录以便在服务激活器处进一步处理并在聚合器处接收流程运行得很好。
然而，固定延迟是将先前的执行时间视为将记录提交给任务执行器的时间，而不是所有记录处理完成并在聚合器通道接收
浏览 6提问于2015-06-06得票数 1
2回答
我对spring集成非常陌生，我不完全理解它是如何工作的，也不完全理解它是否正确。到目前为止，我有这个代码asynchronously --我能够阅读所有的电子邮件，但我不确定它们是否被处理为，或者它是否是正确的方式？SpringApplicationBuilder(Main.class); }
( javax.mail 2)春季启动中是否也有可能移动/删除电子邮件、创建文件夹以及使用电子邮件帐户执行</e
浏览 4提问于2020-01-13得票数 3
Kafka使用者为它处理的每个消息调用一个web服务，如果web服务不可用，那么我预计使用者将尝试处理该消息3次，然后再继续处理下一条消息。因此，在我的测试中，我禁用了the服务，因此没有一条消息可以被正确处理。从日志中，我可以看到这种情况正在发生。 过了一段时间后，我停止并重新启动了Kafka消费者(webservice仍然是禁用的)。我期望在Kafka
浏览 29提问于2019-03-18得票数 2
在我们的SI流中，我们使用Splitter将接收到的有效载荷分解为不同的业务消息(可能在其中最多有5000条业务消息)，如果有效负载包含多个业务消息，则在拆分列表开始时使用Splitter还包括响应业务消息拆分后，我们有一个Router将列表中的第一个消息(响应业务消息)路由到一个通道(res
浏览 4提问于2014-11-25得票数 1
回答已采纳
场景/用例：我有一个Spring应用程序，使用Spring向主题发送消息。在完成特定事件(由http请求触发)后，将创建一个新线程(通过Spring @异步)，该线程调用kafkatemplate.send()，并对它返回的ListenableFuture进行回调。问题行为:但是，在执行负载测试时，如果我将所有kafka代理关闭3到5分
浏览 1提问于2019-10-03得票数 1
回答已采纳
1回答
我试图实现一个spring集成流，如果输入变量为true，则需要多线程调用。如果此变量为真，则执行多线程调用的流和主线程将继续其流。
然后，在最后，它将需要等待两个流完成，然后返回一个响应。我已经成功地使用拆分器实现了多线程spring集成流，但是拆分器会导致所有消息进入同一个通道，这是不同的，因为多线程调用需要调用与执行的主线
浏览 4提问于2022-10-20得票数 2
回答已采纳
1回答
我已经将PlatformTransactionManager设置为Kafka事务作为一个bean。我在方法级别使用了@Transactional注解。我们将在3个主题中发布消息。在第一个主题中发布消息后，我放入Thread.sleep(5000)，如果执行时间超过6秒，当前线程将从过滤器中断。因此，这里的通话被打断了，但消息正在发布到Kafka。
浏览 49提问于2021-09-05得票数 0
我们使用spring-integration-kafka version3.1.2.RELEASE和int- kafka :message-driven-channel-adapter来消费来自远程kafka生产者发送加密的消息，我们使用反序列化程序解密实际的消息。我们可以使用该主题中发布的所有消息。我们已经使用了自动提交作为false。我们想知道如何在成功处理消息后提交或确认来自我们的服务的消息。
浏览 1提问于2019-05-19得票数 0
在我们的项目中，我们使用spring-cloud-stream-binder-kafka版本1.1.2与kafka集成。最近，我们遇到了这样一种情况:我们的一个服务在启动后使用了某个主题中的旧消息(已经使用过)。该主题有2个分区和2个消费者分组在一个消费者组下。我们不确定偏移量是否正确提交给zookeeper。在启动期间，每条消息都会抛出下面的错误消息</
浏览 14提问于2017-06-28得票数 0
我有一个巨大的xml，它作为Spring集成流的输入有效负载。因此，我在transformer中使用claim而不是header enricher来保留我的有效负载。我使用的是内存中的消息库。在我的SI流的后面，我有一个拆分器，它将有效负载拆分成多个线程，每个线程将根据其中一个属性payload调用不同的通道。我正在使用路由器来实现这一点。每个流或每个线程都使用一个</em
浏览 0提问于2016-10-05得票数 0
在我们的项目中，我们需要通过许多线程定期从第三方获取数据，然后将这些数据推送给Kafka。如果目前无法使用Kafka服务器，则应该终止流，在下一次预定执行过程中应该丢失并重新获取获取的数据。此外，还需要使用事务管理，因为我们需要将这批消息发送给Kafka到不同的主题。如果没有发送一条消息，则所有其他消息都应该回滚。
如果Kafka Server不可用，我们就遇到了终止执行<
浏览 1提问于2018-07-16得票数 2
2回答
我有一个客户端从一个由几个分区组成的Kakfa主题中读取。客户端为它读取的每一条Kafka消息调用一个webservice。如果few服务在几次重试后无法使用，我想阻止消费者从Kafka阅读。参考以前的Stackoverflow问题()，我自动调用了，并调用changeState()方法试图阻止消费者，但是日志显示，在调用changeState()之后，用户将继续读取来自Kafka
浏览 2提问于2019-04-05得票数 1
云服务器
ICP备案
实时音视频
即时通信 IM
云直播
腾讯云开发者
