如何从Knative Broker调用同步获取Funq函数返回结果？

为什么会收到202状态码？ #

Knative Broker是异步事件路由组件，它的核心职责是接收事件后异步分发给订阅的服务，而非同步返回服务的处理结果。返回202 Accepted仅表示Broker已成功接收事件，后续的处理流程是异步执行的，因此无法通过Broker请求直接拿到fakeEventListener的返回值。

方案1：直接调用Funq HTTP端点（同步获取返回结果） #

Funq默认会将@Funq注解的函数暴露为HTTP POST端点，绕过Broker直接调用该端点即可同步获取函数返回值。

调整后的Curl命令 #

假设你的Quarkus服务域名/IP为your-quarkus-service，函数fakeEventListener对应的HTTP路径为/funq/fakeEventListener，执行以下命令：

curl -v -X POST http://your-quarkus-service/funq/fakeEventListener \
  -H "Content-Type: application/json" \
  -d '"{}"'

执行后会直接收到函数返回的"received"，状态码为200。

端点路径说明 #

Funq的默认路由规则为：/funq/{函数名}，如果需要自定义路径，可以通过@Funq(path = "/custom-path")注解修改。

方案2：基于Knative Broker的异步结果获取（无法同步，但可实现请求-响应模式） #

如果业务必须依赖Knative Broker的事件路由能力，无法绕开Broker，则无法实现同步获取结果，但可以通过请求-响应模式间接获取处理结果：

1. 发送事件时，在CloudEvent中添加自定义扩展字段（如Ce-Reply-To），指定结果的接收地址（比如Kafka主题、另一个Knative服务端点）。
2. 处理函数完成后，将结果发送到指定的回复地址。
3. 客户端监听回复地址，获取处理结果。

修改后的示例代码 #

import io.quarkus.funq.Funq
import io.quarkus.knative.eventing.CloudEventMapping
import io.smallrye.reactive.messaging.kafka.KafkaRecord
import org.eclipse.microprofile.reactive.messaging.Outgoing
import org.jboss.logging.Logger
import io.cloudevents.CloudEvent

@Funq
@CloudEventMapping(trigger = "fakeEvent")
fun fakeEventListener(payload: String, cloudEvent: CloudEvent) {
    val log = Logger.getLogger(javaClass)
    log.info("$payload")
    
    // 从原始事件中读取回复地址（需客户端发送时携带该扩展字段）
    val replyTo = cloudEvent.getExtension("reply-to") ?: "event-replies"
    
    // 发送处理结果到指定回复通道
    sendReply(replyTo, "received")
}

@Outgoing("reply-producer")
fun sendReply(replyTopic: String, result: String): KafkaRecord<String, String> {
    return KafkaRecord.of(replyTopic, result)
}

配置说明 #

在application.properties中配置回复通道的消息中间件（以Kafka为例）：

mp.messaging.outgoing.reply-producer.connector=smallrye-kafka
mp.messaging.outgoing.reply-producer.bootstrap.servers=kafka-cluster:9092

客户端发送事件时携带回复地址 #

curl -v "http://broker-ingress.knative-eventing.svc.cluster.local/brokerName/default" \
  -X POST \
  -H "Ce-Id: 1234" \
  -H "Ce-Specversion: 1.0" \
  -H "Ce-Type: fakeEvent" \
  -H "Ce-Source: curl" \
  -H "Ce-Reply-To: event-replies" \
  -H "Content-Type: application/json" \
  -d '"{}"'

之后客户端通过监听event-replies主题即可获取处理结果。

内容的提问来源于stack exchange，提问作者Sherlock Wayne