0

我正在尝试使用 Spring Integration 和 Project Reactor 创建一个简单的流程，在其中我使用 Reactor Kafka 使用记录，将它们传递到一个通道，从那里它将使用 Reactor Kafka 将消息生成到另一个主题。

消费流程为：

@Service
public class ReactiveConsumerService {
    public ReactiveKafkaConsumerTemplate<String, String> reactiveKafkaConsumerTemplate;

    @Qualifier("directChannel")
    @Autowired
    public MessageChannel directChannel;

    public ReactiveConsumerService(ReactiveKafkaConsumerTemplate<String, String> reactiveKafkaConsumerTemplate) {
        this.reactiveKafkaConsumerTemplate = reactiveKafkaConsumerTemplate;
    }

    @Bean
    public IntegrationFlow readFromKafka() {
        return IntegrationFlows.from(reactiveKafkaConsumerTemplate.receiveAutoAck()
                .map(GenericMessage::new))
            .<ConsumerRecord<String, String>, String>transform(ConsumerRecord::value)
            .<String, String>transform(String::toUpperCase)
            .channel(directChannel)
            .get();
    }
}

生产流程为：

@Service
public class ReactiveProducerService {
    private final ReactiveKafkaProducerTemplate<String, String> reactiveKafkaProducerTemplate;

    @Qualifier("directChannel")
    @Autowired
    public MessageChannel directChannel;

    public ReactiveProducerService(ReactiveKafkaProducerTemplate<String, String> reactiveKafkaProducerTemplate) {
        this.reactiveKafkaProducerTemplate = reactiveKafkaProducerTemplate;
    }

    @Bean
    public IntegrationFlow kafkaProducerFlow() {
        return IntegrationFlows.from(directChannel)
            .handle(s -> reactiveKafkaProducerTemplate.send("topic2", s.getPayload().toString()))
            .get();
    }
}

我想知道我应该如何以及在哪里执行订阅。

编辑：

我添加了一个.subscripe()，它仍然不起作用：

2022-01-25 20:36:59.570  INFO 1804 --- [ration-sample-1] o.a.kafka.common.utils.AppInfoParser     : App info kafka.consumer for consumer-reactive-kafka-spring-integration-sample-1 unregistered
2022-01-25 20:36:59.573 ERROR 1804 --- [oundedElastic-1] reactor.core.publisher.Operators         : Operator called default onErrorDropped

reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.IllegalStateException: No subscriptions have been created
Caused by: java.lang.IllegalStateException: No subscriptions have been created
    at reactor.kafka.receiver.ReceiverOptions.subscriber(ReceiverOptions.java:423) ~[reactor-kafka-1.3.9.jar:1.3.9]
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Assembly trace from producer [reactor.core.publisher.FluxPeekFuseable] :
    reactor.core.publisher.Flux.doOnRequest
Caused by: java.lang.IllegalStateException: No subscriptions have been created

    reactor.kafka.receiver.internals.ConsumerHandler.receive(ConsumerHandler.java:110)
Error has been observed at the following site(s):
    *________Flux.doOnRequest ? at reactor.kafka.receiver.internals.ConsumerHandler.receive(ConsumerHandler.java:110)
    |_            Flux.filter ? at reactor.kafka.receiver.internals.DefaultKafkaReceiver.lambda$receiveAutoAck$6(DefaultKafkaReceiver.java:70)
    |_         Flux.publishOn ? at reactor.kafka.receiver.internals.DefaultKafkaReceiver.lambda$receiveAutoAck$6(DefaultKafkaReceiver.java:71)
    |_               Flux.map ? at reactor.kafka.receiver.internals.DefaultKafkaReceiver.lambda$receiveAutoAck$6(DefaultKafkaReceiver.java:72)
    *______________Flux.using ? at reactor.kafka.receiver.internals.DefaultKafkaReceiver.lambda$withHandler$19(DefaultKafkaReceiver.java:137)
    *__________Flux.usingWhen ? at reactor.kafka.receiver.internals.DefaultKafkaReceiver.withHandler(DefaultKafkaReceiver.java:129)
    |_                        ? at reactor.kafka.receiver.internals.DefaultKafkaReceiver.receiveAutoAck(DefaultKafkaReceiver.java:68)
    |_                        ? at reactor.kafka.receiver.KafkaReceiver.receiveAutoAck(KafkaReceiver.java:124)
    |_         Flux.concatMap ? at org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate.receiveAutoAck(ReactiveKafkaConsumerTemplate.java:69)
    |_               Flux.map ? at reactor.kafka.spring.integration.samples.service.ReactiveConsumerService.readFromKafka(ReactiveConsumerService.java:38)
    |_              Flux.from ? at org.springframework.integration.channel.FluxMessageChannel.subscribeTo(FluxMessageChannel.java:118)
    |_ Flux.delaySubscription ? at org.springframework.integration.channel.FluxMessageChannel.subscribeTo(FluxMessageChannel.java:119)
    |_         Flux.publishOn ? at org.springframework.integration.channel.FluxMessageChannel.subscribeTo(FluxMessageChannel.java:120)
    |_          Flux.doOnNext ? at org.springframework.integration.channel.FluxMessageChannel.subscribeTo(FluxMessageChannel.java:121)
Original Stack Trace:
        at reactor.kafka.receiver.ReceiverOptions.subscriber(ReceiverOptions.java:423) ~[reactor-kafka-1.3.9.jar:1.3.9]
        at reactor.kafka.receiver.internals.ConsumerEventLoop$SubscribeEvent.run(ConsumerEventLoop.java:207) ~[reactor-kafka-1.3.9.jar:1.3.9]
        at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68) ~[reactor-core-3.4.14.jar:3.4.14]
        at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28) ~[reactor-core-3.4.14.jar:3.4.14]
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[na:na]
        at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) ~[na:na]
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[na:na]
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[na:na]
        at java.base/java.lang.Thread.run(Thread.java:833) ~[na:na]

2022-01-25 20:36:59.772  INFO 1804 --- [           main] o.s.b.web.embedded.netty.NettyWebServer  : Netty started on port 8090
2022-01-25 20:36:59.853  INFO 1804 --- [           main] o.a.c.impl.engine.AbstractCamelContext   : Routes startup summary (total:0 started:0)
2022-01-25 20:36:59.853  INFO 1804 --- [           main] o.a.c.impl.engine.AbstractCamelContext   : Apache Camel 3.12.0 (camel-1) started in 149ms (build:84ms init:59ms start:6ms)
2022-01-25 20:36:59.866  INFO 1804 --- [           main] ReactorKafkaSpringIntegrationApplication : Started ReactorKafkaSpringIntegrationApplication in 4.246 seconds (JVM running for 4.616)

示例代码：

@Service
public class ReactiveProducerService {
    private final ReactiveKafkaProducerTemplate<String, String> reactiveKafkaProducerTemplate;

    @Qualifier("directChannel")
    @Autowired
    public MessageChannel directChannel;

    public ReactiveProducerService(ReactiveKafkaProducerTemplate<String, String> reactiveKafkaProducerTemplate) {
        this.reactiveKafkaProducerTemplate = reactiveKafkaProducerTemplate;
    }

    @Bean
    public IntegrationFlow kafkaProducerFlow() {
        return IntegrationFlows.from(directChannel)
            .handle(s -> reactiveKafkaProducerTemplate.send("topic2", s.getPayload().toString()).subscribe(System.out::println))
            .get();
    }
}
4

1 回答 1

1

当应用程序上下文自动启动reactiveKafkaConsumerTemplate端点时，立即发生订阅。.<ConsumerRecord<String, String>, String>transform(ConsumerRecord::value)

将此视为替代方案：

/**
 * Represent an Integration Flow as a Reactive Streams {@link Publisher} bean.
 * @param autoStartOnSubscribe start message production and consumption in the flow,
 * when a subscription to the publisher is initiated.
 * If this set to true, the flow is marked to not start automatically by the application context.
 * @param <T> the expected {@code payload} type
 * @return the Reactive Streams {@link Publisher}
 * @since 5.5.6
 */
@SuppressWarnings(UNCHECKED)
protected <T> Publisher<Message<T>> toReactivePublisher(boolean autoStartOnSubscribe) {

尽管我认为您的意思是出站方面的订阅。从你的问题中不清楚，但它reactiveKafkaProducerTemplate有一个类似的合同：

public Mono<SenderResult<Void>> send(String topic, V value) {

因此，您需要订阅返回Mono的内容以启动流程。

注意：你也有混乱的论点send()。你不是这个意思吗：reactiveKafkaProducerTemplate.send("test", "topic2")？

要让它订阅那个Mono，你只需要自己做那个handle()：

.handle(s -> reactiveKafkaProducerTemplate.send("topic2", "test").subscribe())

更新 2

java.lang.IllegalStateException: No subscriptions have been created类似的错误reactor.kafka.receiver.ReceiverOptions.subscriber()意味着您没有分配要收听的主题、模式或分区。

见ReceiverOptions.subscription()或ReceiverOptions.assignment()。

于 2022-01-24T16:05:16.943 回答