《中邮消费金融有限公司》专题

关于破坏者，我有以下问题： 消费者（事件处理器）没有实现他们实现EventHandler的任何可调用或可运行接口，那么他们如何能够并行运行，因此，例如，我有一个disruptor实现，其中有这样一个菱形模式 其中c1到c3可以在p1之后并联工作，C4和C5在p1之后工作。 所以通常我会有这样的东西（P1和C1-C5是可运行/可调用的） 但是在Disruptor的情况下，我的事件处理程序都没有实现R

我使用的是高级消费者，如所述：https://cwiki.apache.org/confluence/display/KAFKA/Consumer组示例 我注意到，我的消费者不会永远运行，而是在一段时间后结束。在zookeeper一侧，我看到以下内容： INFO已处理的会话终止的setsionid： 0x144a4854325004d（org.apache.zookeeper.server.准备请

我的消费者配置有kafka批处理侦听器配置和@KafkaListener消耗消息列表。我有一个消费者拦截器，我想为每条记录设置唯一ID，并将其值存储在映射诊断上下文（MDC）中。如果我的kafka侦听器消耗单个消息，则唯一ID是正确的。但是我的kafka侦听器消耗消息列表，因此MDC. get（“id”）只获取最后一个值。如何处理它？我的拦截器；

我已经读过一些关于动画片和多个消费者的问题，但我仍然不明白它是如何工作的。 我的用例：我有一个只有一个分片的运动流。我想使用不同的lambda函数使用这个分片，每个函数都独立。就像每个lambda函数都有自己的分片迭代器一样。 有可能吗？是否设置多个lambda使用者（基于流）从同一流/碎片读取？

但是，consumer只从主题中第一个未提交的消息开始轮询。我希望总是从偏移量0开始，不管提交的消息是什么。使用Alpakka消费者，如何手动指定偏移量？

我目前正在开发Kafka模块，我正在使用Kafka通信的抽象。我能够集成生产者 Spring Boot测试类 监听器类 我的问题是：在测试类中，我断言分区、有效负载等是从BlockingQueue轮询的，然而，我的问题是如何验证用KafkaListener注释的类中的业务逻辑是否得到正确执行，并根据错误处理和其他业务场景将消息路由到不同的主题。在一些示例中，我看到了CountDownLatch的断

我在上添加了自定义费用： 但一旦用户点击结帐，费用就被删除，而不是计入总数。 我已通过执行以下检查来验证在调用

我有代码： 处理消费者异常的最佳方法是什么？现在，如果异常引发，它将被吞没。。。

我创建了一个简单的生产者-消费者应用程序，使用自定义序列化器和反序列化器。 在我生成的Message类中添加了一个新方法之后，使用者开始在反序列化时被堆栈。我的生产者使用的是新类（有新方法），消费者使用的是旧类（没有方法）。 JSONSerializer/Deserializer可以处理这些类型的修复吗？如果我要使用JSONSerialzier，它应该只关心类的模式，对吗？

Kafka新手。 Kafka版本：2.3.1 我正在尝试使用Spring cloud使用来自两个主题的Kafka消息。除了kafka活页夹和下面的一些简单配置之外，我没有做太多配置。每当（组协调器lbbb111a.uat.pncint.net:9092（id：2147483641机架：null）不可用或无效时，将尝试重新发现）发生时，已经处理的一堆消息会再次被处理。不确定发生了什么。

我用java编写我所有的微服务。我想在Amazon SQS中使用多个消费者，但每个消费者在负载均衡器后面的AWS上有多个实例。 我使用SNS作为输入流 我在SNS之后使用SQS标准队列。 我在stackoverflow上发现了同样的问题（使用多个消费者的Amazon SQS） 此示例为 https://aws.amazon.com/fr/blogs/aws/queues-and-notificat

我是pyflink的新手。我正在尝试编写一个python程序来从kafka主题读取数据并将数据打印到标准输出。我按照链接Flink Python Datastream API Kafka Producer Sink Serializaion进行了操作。但由于版本不匹配，我一直看到NoSuchMethod odError。我添加了https://repo.maven.apache.org/maven

apache kafka文档提到以下内容： 如果所有使用者实例具有相同的使用者组，那么记录将有效地在使用者实例上进行负载平衡。 如果所有的使用者实例都有不同的使用者组，那么每个记录都将广播给所有的使用者进程。

谢了。

我如何将电话限制在每5秒一次。注意：只能修改reallySlowApi。 编辑：我知道，但是如果Api变得更慢，它就不能解决问题。我需要使用的最佳方式。