问题标签 [spring-cloud-function]

目前我有类似这样的 Spring Boot 应用程序。

我想在这里使用反应模式使用 Spring Cloud Function。

此外，我当前的解决方案是非阻塞的吗？我问这个是因为这是我第一次在这种情况下使用 Kotlin 协程。

Java 解决方案也适用于我，因为我只是想理解这里的概念。

我有一个沙箱用于探索 Spring Cloud Stream 中新添加的功能，但是在一个 Spring Cloud Stream 应用程序中使用 Function 和 Supplier 时遇到了问题。

在代码中，我使用了docs中描述的示例。

首先，我在项目中添加Function<String, String>了相应的spring.cloud.stream.bindings和spring.cloud.stream.function.definition属性application.yml。一切正常，我将消息发布到my-fun-inKafka 主题，应用程序执行功能并将结果发送到my-fun-out主题。

然后我将相应的和更新的值添加Supplier<Flux<String>>到同一个项目中。这里奇怪的事情开始发生。当我尝试启动应用程序时，我收到以下错误：spring.cloud.stream.bindingsspring.cloud.stream.function.definitionfun;sup

之后我尝试了几件事：

1. 恢复spring.cloud.stream.function.definition为fun（禁用supbean 绑定到外部目标）。应用程序启动，功能工作，供应商没有工作。一切如预期。
2. 更改spring.cloud.stream.function.definition为sup（禁用funbean 绑定到外部目标）。应用程序启动，功能不起作用，供应商工作（my-sup-out每秒产生到主题的消息）。一切都如预期一样。
3. 将值更新spring.cloud.stream.function.definition为fun;sup。应用程序没有启动，得到相同的 MessageDeliveryException。
4. 将值交换spring.cloud.stream.function.definition为sup;fun。应用程序启动，供应商工作，但功能不起作用（没有向my-fun-out主题发送消息）。

最后一个比错误更让我困惑）所以现在我需要有人帮助解决问题。

我在配置中错过了什么吗？;为什么更改以in分隔的 bean 顺序spring.cloud.stream.function.definition会导致不同的结果？

完整项目上传到GitHub并添加如下：

流应用程序.java：

应用程序.yml

build.gradle.kts：

我尝试Spring Cloud Function在 AWS Lambda 中部署一个具有多个功能的函数。

对于 HTTP 访问，我创建了一个HTTP API Gateway (not a REST API). 我想使用这里描述的函数路由：https ://cloud.spring.io/spring-cloud-static/spring-cloud-function/3.0.1.RELEASE/reference/html/spring-cloud-function.html# _function_routing。

使用以下配置，应根据以下配置RoutingFunction将调用委托给正确的函数function HTTP-Header：

Lambda 处理程序类是：

并且配置的函数名称（FUNCTION_NAME）是：functionRouter。

当我向 API-Gateway 发送请求时，FunctionRouter 得到一个 FluxJust 对象而不是 Message 对象，因为 RequestHandler 似乎是一个发布者。所以我得到以下异常：

我当前的解决方法是在请求被委托给之前拦截请求，RoutingFunction并尝试HashMap使用以下代码从 重构有效负载：

有没有一种方法可以将 HTTP API Gateway 与 in 结合Function Routing使用AWS？

我有三个 Maven 项目：

• demo-api：包含函数 API 的 POJO，即OutputPojo.class, InputPojo.class
• demo-packaged-function：定义一个public OutputPojo apply(InputPojo) {...}依赖于“demo-api”项目的函数。
• demo-deployer：使用 spring-cloud-function-deployer 来运行打包的函数，也依赖于 'demo-api' 项目。

如果我对函数签名使用简单类型（即重构为String apply(String input) {...}然后一切正常。但是，通过上述设置，我得到以下异常：

Exception in thread "main" java.lang.ClassCastException: class org.example.function.api.InputPojo cannot be cast to class org.example.function.api.InputPojo (org.example.function.api.InputPojo is in unnamed module of loader 'app'; org.example.function.api.InputPojo is in unnamed module of loader org.springframework.cloud.function.deployer.FunctionArchiveDeployer$1 @7fedfe27)

这是有道理的，因为类InputPojo是由两个类加载器加载的。如果我不将函数打包为胖 jar，我会ClassNotFoundException在尝试部署 jar 时得到，并且我无法demo-api从部署程序项目中删除依赖项，否则在调用函数时我无法使用 POJO 类。如果没有函数的简单参数类型，这意味着如何工作？

文档没有涵盖这一点，虽然有使用 POJO 的打包函数的示例，但我找不到部署者实际调用它们的任何示例 - 除了一个单元测试，它演示类型转换并将 POJO 转换为消息.

spring-cloud-function-deployer 示例都显示了在启动时加载的已部署函数，即 ApplicationContext 以必要的属性启动，指向要加载的打包 jar。

有没有办法在运行时以编程方式调用部署程序，而不是依赖自动配置？如果我想在应用程序上下文启动后的某个时间部署该功能，或者如果我想从同一个 jar 部署多个功能等。

还有一种方法可以取消部署任何已加载的功能，还是像从目录中删除该功能一样简单？

我正在尝试使用 Spring Cloud Kafka Streams 处理来自包含不同类型消息的 Kafka 主题的消息。例如，我们从主题收到一条 JSON 消息，它可以是 A 类消息，也可以是 B 类消息。生产者在标头中添加消息类型，有没有办法在功能绑定器中读取该标头信息并相应地转换消息？或者还有一个“选择”选项，用于在消息进入时进行分支，将消息路由到正确的转换器？

我正在使用带有弹簧云功能的弹簧云流兔子活页夹并定义监听器，例如：

我还将失败的消息重新路由到 DLQ。问题是发生致命错误org.springframework.messaging.converter.MessageConversionException时。它不会像https://docs.spring.io/spring-amqp/reference/html/#exception-handlingConditionalRejectingErrorHandler中提到的那样得到处理，并且永远循环。

有没有办法让这个工作ConditionalRejectingErrorHandler？

现在我通过@ServiceActivator(inputChannel = "errorChannel")自己使用和处理错误来解决问题。

依赖项：

我对 Spring Cloud Stream 中的新路由功能有一些问题

我试图实现一个简单的场景，我想发送一个带有标头 spring.cloud.function.definition = consume1 或 consume2 的消息

我希望应根据标头上发送的内容调用 consumer1 或 consume2 ，但这些方法是随机调用的。

我使用兔子管理控制台将消息发送给交换消费者

我有以下日志：

应用程序.yml

DemoApplication.java

知道如何修复通往消费者的路线吗？

提前致谢。

演示回购

我不确定这是一个错误还是我尚未找到的一些晦涩的设置，但是对于相同的调用参数，我得到了不同的日志行为。我设置了一个脚本，使用相同的有效负载每 3 秒调用一次端点。在第一次通话时出现冷启动，我看到：

在随后的几个电话中，我只看到：

这是我的host.json：

可能是什么问题呢？采样设置不影响行为。问题是诸如“收到的消息...”之类的消息是重要的消息。

我还查看了该功能的 kudu 应用程序转储，那里的日志记录也不完整。

我从用包含日期时间类型的 spring 云函数编写的天蓝色函数输出 POJO。我已经在它被视为 POJO 的地方进行了即时尝试：

我尝试将其作为 OffsetDateTime 并得到：

我在春季级别尝试了各种事情，例如：

但他们没有任何效果。阅读https://docs.microsoft.com/en-us/azure/azure-functions/functions-reference-java#pojos听起来 azure-functions-java-worker 正在完成将 pojo 转换为 json 和它根本不使用杰克逊。如何通过值的自动序列化获得与 JavaTimeModule 类似的输出？我必须使用字符串输出吗？我找不到与此相关的文档，也没有示例。