我正在使用Spring Cloud Stream 3.0.6(Cloud:hoxton.sr6,Boot 2.3.0.release)和Solace PubSub+。我不能让并发消费者工作。无论我配置什么,总是有一个线程依次执行每个传入消息。 以下是我的代码: 这里会有什么问题? 安慰pubsub+活页夹 本地运行的Solace PubSub+实例的Docker组合文件:
假设我的使用者从一个代理轮询,该代理有多个主题,每个主题有多个分区。我在同一个消费群体中总共有5个消费者。如果我的每个消费者都进行投票,将返回的数据顺序是什么? topicD-分区5 我的问题是,在这个单一的1轮询中,在按顺序移动到下一个主题/分区之前,我会收到来自该主题/分区的所有可用消息吗?意思例如: 在一次投票循环中,我收到了这个... 或者在那个单一的1轮询循环中,有可能接收到这个消息顺序
null 编辑:好的,所以我取得了一些进步(如果我错了请纠正我): 每个消费者都将获得所有消息。 租约被分配了一个EventProcessorHost,所以它需要一个唯一的名称,所以这里的使用者组名称实际上并不相关。 仍然不能百分之百确定context.checkpointasync,但我相信它只适用于ConsumerGroup?
我有三根线。线程1(T1)是生成器,它生成数据。线程2和线程3(T2和T3)分别等待T1的数据在单独的循环中处理。我正在考虑在线程之间共享BlockingQueue,并通过调用“Take”让T2和T3等待。
我有一个Kafka流应用程序,它从几个主题中获取数据,并将数据加入另一个主题。 Kafka配置: 注意:我在运行Kafka Brokers的机器上运行Kafka Streams应用程序。 每小时消耗/产生数百万条记录。每当我让Kafka经纪人倒下时,都会进入再平衡阶段,再平衡大约需要30分钟,有时甚至更长时间。 有人知道如何解决Kafka消费者的再平衡问题吗?而且,很多时候,它在重新平衡时抛出异常
我的应用程序由一个带有POST方法的REST控制器组成,用于提交我必须使用生产者发送到主题的数据。 这是控制器 使用Spring-Cloud-Stream版本 从3.1版开始,和注释被弃用,所以我尝试切换到新的方式来设置生产者,我就是这样工作的 最后在应用程序中。yaml我有这个 现在的问题是,当我启动应用程序时,方法被无限调用(我在主题中看到消息)。然后使用供应商似乎我被迫在供应商内部定义消息数
我是的新手,正在尝试使用组件。 我在这里简化了代码,但基本上我有一个参与者,它从endpoint使用。 因此,我尝试这种“直接”方法,但在endpoint:endpoint[direct://mycamelendpoint]上没有可用的消费者。Exchange[message:My test message]异常。 我在这里缺少什么配置?
我的问题是关于Kafka在爪哇的消费者 > 已启动Kafka服务器 创建的主题 创作者 创建的消费者 我在终端中做的所有这些事情,工作正常,能够在消费者处正确接收日志。运行下面的消费者(在Java),但没有收到任何记录。它继续汇集在 也没有收到任何记录。 下面给出的我的 java 消费者代码 请告诉我在java消费者类中接收消息,我在配置中做错了什么吗?属性中的“group.id”是怎么回事? 下
我正在尝试使用NodeJS从远程机器连接到远程Apache Kafka服务器。我无法从nodejs代码中生成所需的kafka主题的消息。我也无法消费任何数据从主题以及。 我使用的是Apache-kafka版本2.122.2.1和Java8。我也在使用节点版本8.11.0。我还启动了zookeeper服务器和kafka服务器。我在ubuntu机器上本地创建了一个主题和一个生产者和消费者,以检查apa
最近,我们与Kafka消费者和生产商之间出现了一些性能问题。我们在scala中使用Kafka Java API。打开和关闭消费者和生产者对象的良好做法是什么?我认为这是一个非常开放的问题,正确的答案总是,但我正在尝试对此进行推理。 消费者可以长时间运行连接并保持开放吗? 当我们完成信息生成时,生产者是否应该关闭?
我有一个处于RPC模式的消费者(RabbitListner),我想知道是否有可能引发发布者可以处理的异常。 为了更清楚地说明我的情况如下: 发布者以RPC模式发送消息 消费者收到消息,检查消息的有效性,如果由于缺少参数,消息无法计数,那么我想抛出异常。异常可以是特定的业务异常,也可以是特定的AmqpException,但我希望发布者可以在未进入超时状态时处理该异常 我尝试使用AmqpRejectA
目前,我每5分钟调用一次消费api来消费数据。在这种情况下,不必要地调用了消费者api,但我希望在Kafka中有新数据可用时调用该api。
那么我如何确保我的队列中只有一个消费者呢?
我成功地建立了一个话题交换,并且能够同时向几个消费者传递消息。 我还想向竞争对手传递信息,并继续使用主题交换。我了解到,使用相同的队列名称可以让消费者竞争消息。然而,我可能弄错了,因为我无法使它工作。 为同一主题的多个侦听器设置: < li >申报话题交流 < li >对于每个侦听器,用自动生成的名称声明一个新队列 < li >用给定的主题路由关键字将此队列绑定到上面的交换 如何将相互竞争的消费者
我正在处理一个kafka用例,在这个用例中,我需要在生产者和消费者端具有事务性语义...我可以使用kafka transaction API 0.11将事务性消息发布到kafka集群,但在消费者端,我面临着一个问题...我在属性文件中设置了但我不能使用它...我可以看到消息被使用但这不是希望的... 生产者代码 ProducerTX.Properties 消费者 感谢你的帮助..谢谢