spring-cloud-stream 相关问题

Spring Cloud Stream允许用户使用Spring Integration开发和运行消息传递微服务,并在本地或云中或甚至在Spring Cloud Data Flow上运行它们。只需添加@EnableBinding并将您的应用程序作为Spring Boot应用程序(单个应用程序上下文)运行。您只需要连接到总线的物理代理,如果类路径上有相关的总线实现,则这是自动的。

Spring Cloud Stream Kafka Binder 中的消费者和生产者并发配置?

spring.cloud.stream.binding.myInputProcess.consumer.concurrency:3 我的应用程序中有这样的配置,它从 kafka 获取输入并处理它,然后放回另一个 kafka。 那么,我...

回答 1 投票 0

Spring Cloud Streams 错误处理不起作用

Spring Cloud Streams 中错误处理的不同方法的示例很少,并且通过文档部分提供的少数示例似乎也不起作用。 我有一个测试仓库...

回答 1 投票 0

Spring Cloud Stream 使用 Kafka DLT 处理毒丸

spring-boot 2.5.2 弹簧云 Hoxton.SR12 spring-kafka 2.6.7(因问题降级:https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1079) 我正在按照这个食谱来...

回答 1 投票 0

GlobalKTable 作为带有 Kafka Streams Binder 的 QueryableStore

我正在使用 spring-boot-starter-parent 版本 3.1.2 和 spring-cloud-stream-binder-kafka-streams 版本 4.0.3 绝大多数在线示例都显示使用 @Input 注释创建 GlobalKTable...

回答 1 投票 0

Spring Cloud Stream Rabbit 使用 ssl 启动时出错

我正在尝试使用Spring cloud Stream从aws(AmazonMQ)连接rabbitmq,并使用ssl连接spring-cloud-stream-binder-rabbit,但是在启动应用程序时出现错误,我不需要

回答 1 投票 0

将 JdbcMessageStore 添加到 Spring Kafka 聚合器

如何将 JdbcMessageStore 设置为 Aggregator,以便它使用 RDBMS 而不是内存中消息存储? 目前AggregatorAnnotationPostProcessor直接设置new SimpleMessageStore(...

回答 1 投票 0

`spring-kafka` 和 `spring-cloud-stream-binder-kafka` 的现有转换器是否不同?

在spring-kafka提供的KafkaProperties中,生产者和消费者的反序列化器是String{Des/S}序列化器。 但是,使用 spring-cloud-stream-binder-kafka 时,改为 ByteArray{...

回答 1 投票 0

具有多个键的绑定RoutingKey不起作用

我想使用指定的路由键过滤rabbit mq队列消息,以下是我在application.properties中的配置: spring.cloud.stream.rabbit.bindings.subscription-changed-input。

回答 1 投票 0

将 Spring Cloud Stream 代码转换为基于 Spring boot 3 的功能

我必须使用基于功能的bean将使用Spring Cloud Stream发送RabbitMQ的代码从spring boot 2更新到spring boot 3 有人可以帮忙转换这个简化版本吗? @EnableBinding(M...

回答 1 投票 0

Spring云流中的后处理

Spring Cloud Stream 中引入了对后处理的支持 https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream.html#

回答 1 投票 0

Spring 云流:反应式管道中的异常导致流崩溃,并显示错误调度程序没有订阅者

我有以下消费者功能。如果 WebClient 调用成功则有效 @豆 有趣的消费者()=消费者> { 通量-> Flux.concatMap { s -> WebCli...

回答 1 投票 0

在 Spring 上下文中找不到 StreamBridge bean

StreamBridge 无法注入(使用@Autowired)到服务中并导致以下错误: 导致:org.springframework.beans.factory.NoSuchBeanDefinitionException:没有符合条件的 t...

回答 2 投票 0

将 Spring Cloud Connector 迁移到 Cloudfoundry 中的 java-cfenv 导致 Rabbit-MQ 服务的 TLS 失败

在将 Spring Cloud Services 从 2.4.1 升级到 3.5.5 时,我们发现应用程序无法连接到 Cloud Foundry 中的 Rabbit MQ 服务。它抛出以下异常:- c.r.客户...

回答 1 投票 0

如何在 Spring Cloud Stream 中为生产者专用事务(db + kafka)启用数据库 + kafka 事务?

我们有一个事件驱动的分布式架构,具有使用 Spring Cloud Stream 的独立生产者和消费者微服务,应用程序需要执行以下操作:在生产者中,一个

回答 1 投票 0

无法使用 spring-cloud-stream-test-binder:4.0.4 发布和使用消息

我正在使用 spring-cloud-stream-test-binder 版本 4.0.4 但是在发布/使用时我遇到了以下错误 java.lang.NullPointerException:无法调用“org.springframework.messaging。

回答 1 投票 0

Spring云流和异步流程

我有以下简单的消费者应用程序。我想知道如何使消费者仅在当前异步进程完成后才拉取下一条消息 @SpringBootApplication 公开课应用程序{...

回答 0 投票 0

EmitterProcessor 已弃用,我们如何替换它?

在提出问题之前,我浏览了很多链接,例如:Spring WebFlux (Flux): 如何动态发布,但还无法解决。 我有下面的代码,但是类型 EmitterProcessor 在提出问题之前,我浏览了很多链接,例如:Spring WebFlux (Flux): 如何动态发布,但还无法解决。 我有下面的代码,但是 EmitterProcessor> 类型已弃用,我如何将此代码转换为 Sinks.many().multicast().onBackpressureBuffer()? 配置.java @Configuration public class DynamicDestinationConfig { @Bean public EmitterProcessor<Message<?>> emitterProcessor(){ return EmitterProcessor.create(); } @Bean public Supplier<Flux<Message<?>>> supplier() { return () -> emitterProcessor(); } } 控制器.java @RestController public class DynamicDestinationController { @Autowired private EmitterProcessor<Message<?>> emitterProcessor; @Autowired private ObjectMapper jsonMapper; @SuppressWarnings("unchecked") @PostMapping("/") @ResponseStatus(HttpStatus.ACCEPTED) public void handleRequest(@RequestBody String body, @RequestHeader(HttpHeaders.CONTENT_TYPE) Object contentType) throws Exception { Map<String, String> payload = jsonMapper.readValue(body, Map.class); String destinationName = payload.get("id"); Message<?> message = MessageBuilder.withPayload(payload) .setHeader("spring.cloud.stream.sendto.destination", destinationName) .build(); emitterProcessor.onNext(message); } }

回答 0 投票 0

消息<String>和消息<byte[]>

我正在使用Spring Cloud Stream功能模型 如果有人可以帮助了解哪个性能更好 功能 或者 功能 或者 功能 我正在使用 Spring Cloud Stream 功能模型 如果有人可以帮助了解哪个性能更好 功能 或者 功能 或者 功能、消息>

回答 0 投票 0

消费者函数和服务函数访问同一个kafka流的状态存储是线程安全的吗?

我正在使用 kotlin + Spring Boot + Kafka Streams 和 Spring Cloud Stream。 我有一个服务功能,可以验证客户端的请求并将其发送到 kafka 主题。 要验证请求,需要...

回答 0 投票 0

Spring Cloud Stream:Spring Boot 3.x:JsonProperty、JsonIgnoreProperties 在消息转换器中与 Jackson 的行为异常

我目前正在开发 Spring Boot 应用程序(版本 3.0.6)并使用 Spring Cloud(版本 2022.0.2)。我有两个不同的端点(“/mvc”和“/message”),它们使用两个

回答 1 投票 0

© www.soinside.com 2019 - 2024. All rights reserved.