Reactor Kafka中未创建订阅时出错
Reactor Kafka是一个基于Reactor的响应式流处理库，用于处理Kafka消息队列。当在Reactor Kafka中未创建订阅时出错，可能是由于以下原因：
- 订阅不存在：在使用Reactor Kafka时，首先需要创建一个订阅，以便接收Kafka消息。如果未正确创建订阅，尝试订阅时就会出错。解决方法是确保在订阅之前正确创建订阅。
- 主题不存在：在订阅时，需要指定一个有效的Kafka主题来接收消息。如果指定的主题不存在，就会出现订阅错误。解决方法是确认所使用的主题存在，并确保正确指定主题名称。
- Kafka连接问题：Reactor Kafka需要与Kafka集群建立连接才能进行消息传递。如果连接出现问题，例如网络故障或配置错误，就会导致订阅错误。解决方法是检查Kafka连接配置，确保连接参数正确，并确保网络连接正常。
- 权限问题：如果没有足够的权限来访问Kafka主题或执行相关操作，也可能导致订阅错误。解决方法是检查所使用的凭据和权限，确保具有足够的权限来执行所需的操作。
推荐的腾讯云相关产品： 腾讯云提供了一系列与云计算和消息队列相关的产品，以下是一些推荐的产品和其介绍链接：
- 云服务器（CVM）：提供可扩展的计算能力，用于部署和运行应用程序。链接：https://cloud.tencent.com/product/cvm
- 云数据库MySQL版（CDB）：提供高可用性和可扩展性的MySQL数据库服务。链接：https://cloud.tencent.com/product/cdb_mysql
- 云原生容器服务（TKE）：用于部署、管理和扩展容器化应用程序的容器服务。链接：https://cloud.tencent.com/product/tke
- 消息队列CMQ：提供高可靠性、高可用性的消息队列服务，用于异步通信和解耦。链接：https://cloud.tencent.com/product/cmq
请注意，以上链接仅供参考，具体产品选择应根据实际需求和情况进行评估和决策。
相关·内容
2回答
并且订阅是针对单个主题进行的。log.info("onPartitionsRevoked : {}", partitions));}消费者代码:我在消费者代码中的日志将订阅的主题打印为空(ReceiverOptions.java:385) ~[reactor-kafka-1.3.4.jar:1.3.4]
at reactor.kafka.receiver.internals.C
浏览 51提问于2021-06-11得票数 0
回答已采纳
1回答
有谁有一个长期运行的反应式Kafka消费者的例子吗？https://github.com/reactor/reactor-kafka.git中的SampleConsumer在订阅后退出，即expected.But我需要持续接收来自我们的主题的消息，并且在出现故障(特别是连接问题)时应重新连接到该主题。我确实通过https://github.com/CollaborationInEncapsulatio
浏览 20提问于2020-10-11得票数 0
我面临一个问题，而没有订阅从卡夫卡消费者，请注意，我们正在使用反应堆卡夫卡API。它在开始时确实成功地取消订阅，但随后它立即加入到组中，并被分配给该主题的分区，因此本质上它始终保持订阅状态，并且一直在使用来自主题的消息，即使它不应该这样做！;import reactor.kafka.receiver.ReceiverOptions;
impor
浏览 1提问于2021-08-04得票数 0
回答已采纳
我们有一个在spring-boot-starter-webflux:jar:2.1.5.RELEASE和reactor-netty:jar:0.8.8.RELEASE上运行的spring-webflux当响应式客户端在请求完成之前离开(k8s pod被杀死或客户端订阅被释放)，我们会看到服务器简单地停止处理请求，并且不会再看到与该请求相关的应用程序日志。2020-03-18T16:46:40.372Z TRACE --- [reactor-http-epoll-2] r.n.c.ChannelOperations : [id: 0x4cad
浏览 221提问于2020-03-19得票数 1
1回答
主要的一点是，kafka的消费者在试图订阅一个主题时都会阻塞，即使他们只在最初订阅一个主题或当连接断开时也是如此。这可能会导致潜在的大量阻塞调用，从而耗尽线程池的有限数量。我正在使用Reactor-kafka库，ReactiveKafkaConsumerTemplate。
浏览 2提问于2019-11-27得票数 0
1回答
我试图用Spring集成和项目反应堆创建一个简单的流程，在这里我使用了Kafka反应堆的记录，将它们传递到一个通道，从那里它将产生信息到反应堆Kafka的另一个主题中。reactiveKafkaProducerTemplate.send("topic2", s.getPayload().toString())) }我想知道我应该如何和在什么地方执行订阅(ReceiverOptions.java:423) ~[reactor-
浏览 18提问于2022-01-24得票数 1
回答已采纳
我们的应用程序使用springBootVersion = 2.0.4.RELEASE和compile('io.projectreactor.kafka:reactor-kafka:1.0.1.RELEASE我们拥有的Kafka代理的版本是1.0.1。当我们通过创建reactor.kafka.sender.SenderRecord间歇性地向Kafka发送消息时，在响应Kafka时，当查找<e
浏览 154提问于2019-02-13得票数 1
回答已采纳
我正在尝试reactor-kafka消费消息。其他一切都很好，但我想为失败的消息添加一个retry(2)。默认情况下，spring-kafka已经重试了3次失败的记录，我想用reactor-kafka来实现同样的效果。 我使用spring-kafka作为reactive-kafka的包装器。message){
return Mono.error(new RuntimeException("test retry"); //sa
浏览 38提问于2021-10-01得票数 3
回答已采纳
我正在使用reactor-kafka库和spring webflux SSE流消费来自kafka主题的数据。当来自主题的所有消息都被消费完时，我需要返回一个特殊的ServerSentEvent，即最大主题偏移量get等于订阅时从0偏移量开始消耗的当前偏移量。以便客户端知道在kafka topic中没有更多的消息。
有没有可能使用web flux来实现这样的目标？也就是说，如果我说在每100个元素从任何有限的元素列表中消费并作为Serv
浏览 7提问于2021-06-28得票数 1
2回答
我正在使用scala订阅kafka中的一个主题，下面是我正在执行的命令：每当我执行命令时，我都会出错，如下所示：
25:错误:找不到:价值火花还有其他
浏览 0提问于2018-09-12得票数 0
回答已采纳
就像@KafkaListener在阻止Kafka一样-它正在等待新的消息public class KafkaConsumerstatic final String BOOTSTRAP_SERVERS = "localhost:9092";
private static final String TOPIC = "Kafka_Example
浏览 4提问于2019-04-09得票数 3
1回答
当尝试在kafka上生成消息时，不断获得log:reactor.core.Exceptions$ErrorCallbackNotImplemented: org.apache.kafka.common.errors.TimeoutExceptionCaused by: org.apache.kafka.common.errors.TimeoutException: Topic topic not present in metadata after我已经确保我在生产者项目中有杰克逊核心，杰克逊数据库和<em
浏览 1068提问于2021-07-09得票数 0
1回答
我试着找出反应堆项目，我正在寻找一种方法来取消订阅。例如，我知道在订阅Flux之后，我可以获得可用于发送onCancel信号的取消对象的引用，但这只是在进行了订阅之后，我需要将该引用保存在某种集合中。
有没有更好的方法来获取取消对象？或者只是取消订阅。也许是某种包含所有活跃订阅的引用的地方-是的，那将是很棒的……
浏览 29提问于2016-09-02得票数 6
回答已采纳
根据我到目前为止所读到的内容，可以使用ConnectableFlux将Flux多播到多个订阅者，方法类似于以下内容：
Flux<Integer> integerFlux = Flux.range(1,我正在处理一个Mono有多个订阅者的场景。我怎样才能从Mono中获得一个热门的出版商呢？
浏览 14提问于2021-10-21得票数 1
我们通过在消息头和发布消息中设置spring.cloud.stream.sendto.destination来动态设置目标主题。我们正在寻求处理错误情况，如kafka服务器间歇性停机或主题在发布时不可用。我们已经实现了@ServiceActivator来处理所有错误。当主题是动态设置时，生产者错误不会被ServiceActivator捕获，只有消费者错误会命中ServiceActivator。log.error("Error caught" + em.toString());启动了spring
浏览 258提问于2021-05-13得票数 0
我有一个基于拉取的流数据源(就像Kafka一样)。我想把reactor应用到这个事件处理应用程序上。 目前，我使用EmitterProcessor创建了一个无限的事件序列。并且它在初始时间被订阅一次，并且从未被取消。 下面的代码表明了我所做的事情。subscriberContext不是在我的Event级别，而是在订阅级别。因此，在我的代码中将只有一个上下文。它不起作用。问题是，我应该将事件流视为一个Flux<Event>还是多个Mono<Event>，并对每个事件进行订阅<
浏览 28提问于2020-01-09得票数 1
回答已采纳
我找不到一种方法来使用kafka-python库中的KafkaConsumer并使用treq触发对elasticsearch的post请求。我可以将问题分解成小块:创建一个kafka使用者迭代器并从中读取数据(主题可能很大) consumer = KafkaConsumer(bootstrap_servers="kafka:9092", auto_offset_reset='earliest')
consume
浏览 2提问于2016-10-02得票数 0
回答已采纳
处理失败的最好方法是断开并重新连接kafka，并在失败的情况下进行指数回退。这个想法是让消息取消提交，并在进程在特定时间后重新连接时重新处理它们。在akka stream中，我可以这样做： // Restart source will restart the stream if it faildone.whenComplete((d, e) -> {
// Disconnect from
浏览 0提问于2021-02-22得票数 0
抛出大量读取超时(每秒加载1000个请求)。我正在使用Springboot版本2.1.1.RELEASE。请找到下面的代码，如果我缺少任何配置，请告诉我：public WebClient webClient() {}
.contentT
浏览 0提问于2018-12-14得票数 4
回答已采纳
我想要做的是在Webflux中的单声道上有条件地重复，webclient.The的情况如下：
我们有一些业务rest服务返回一个生成的文档。此文档的生成是从在此服务之前调用的另一个服务触发的。
浏览 1提问于2019-04-30得票数 18
回答已采纳
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频
腾讯云开发者
