如何在夸尔库斯注射Kafka模板

Question

我正在尝试注入一个KafkaTemplate来发送一条消息。我正在开发一个位于反应式方法之外的小函数。

我只能找到使用@Ingoing的例子，@Outgoing来自斯莫利，但我不需要KafkaStream。

我尝试使用Kafka-CDI，但无法注入SimpleKafkaProducer。

有什么想法吗？

对于克莱门特的回答

这似乎是正确的方向，但执行订单。.send（“你好”）;我收到此错误：

(vert.x-eventloop-thread-3) Unhandled exception:java.lang.IllegalStateException: Stream not yet connected

我通过命令行使用我的主题，Kafka已经启动并运行，如果我手动生成，我可以看到使用的消息。

好像是相对于doc的这句话:

要将Emitter用于流hello，您需要在代码（或配置）中的某个位置使用@Incoming（“hello”）。

我班上有这样的代码:

    @Incoming("orders")
    public CompletionStage<Void> consume(KafkaMessage<String, String> msg) {
        log.info("Received message (topic: {}, partition: {}) with key {}: {}", msg.getTopic(), msg.getPartition(), msg.getKey(), msg.getPayload());
        return msg.ack();
    }

也许我忘记了一些配置？

Answer 1

自从Clement的回答之后，< code>@Stream注释就被弃用了。必须改用< code>@Channel批注。

您可以使用由< code > quar kus-small rye-reactive-messaging-kafka 依赖项提供的< code >发射器来生成针对Kafka主题的消息。

一个简单的Kafka生产者实现：

public class MyKafkaProducer {

    @Inject
    @Channel("my-topic")
    Emitter<String> myEmitter;

    public void produce(String message) {
      myEmitter.send(message);
    }
}

并且必须将以下配置添加到应用程序中。属性文件：

mp.messaging.outgoing.my-topic.connector=smallrye-kafka
mp.messaging.outgoing.my-topic.bootstrap.servers=localhost:9092
mp.messaging.outgoing.my-topic.value.serializer=org.apache.kafka.common.serialization.StringSerializer

这将向名为my topic的kafka主题生成字符串序列化消息。

请注意，默认情况下，通道的名称也是将在其中生成数据的kafka主题的名称。这种行为可以通过配置进行更改。在反应式消息传递文档中描述了支持的配置属性

Answer 2

因此，您只需要使用发射器：

@Inject
@Stream("orders") // Emit on the channel 'orders'
Emitter<String> orders;

// ...
orders.send("hello");

在你的application.properties中，声明：

## Orders topic (WRITE)
mp.messaging.outgoing.orders.type=io.smallrye.reactive.messaging.kafka.Kafka
mp.messaging.outgoing.orders.topic=orders
mp.messaging.outgoing.orders.bootstrap.servers=localhost:9092
mp.messaging.outgoing.orders.key.serializer=org.apache.kafka.common.serialization.StringSerializer
mp.messaging.outgoing.orders.value.serializer=org.apache.kafka.common.serialization.StringSerializer
mp.messaging.outgoing.orders.acks=1

要避免“流尚未连接”异常，如文档所示：”

要将Emitter用于流hello，您需要在代码（或配置）中的某个位置使用@Incoming（“hello”）。

假设您的应用程序属性中有类似的内容：

# Orders topic (READ)
smallrye.messaging.source.orders-r-topic.type=io.smallrye.reactive.messaging.kafka.Kafka
smallrye.messaging.source.orders-r-topic.topic=orders
smallrye.messaging.source.orders-r-topic.bootstrap.servers=0.0.0.0:9092
smallrye.messaging.source.orders-r-topic.key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
smallrye.messaging.source.orders-r-topic.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
smallrye.messaging.source.orders-r-topic.group.id=my-group-id

添加类似如下的内容：

@Incoming("orders-r-topic")
public CompletionStage<Void> consume(KafkaMessage<String, String> msg) {
    log.info("Received message (topic: {}, partition: {}) with key {}: {}", msg.getTopic(), msg.getPartition(), msg.getKey(), msg.getPayload());
    return msg.ack();
}