问题标签 [disruptor-pattern]

我正在尝试读取文本文件并使用Disruptor.

但是我发现CPU使用率太高了（200％，根据top命令）。

我是性能调优和线程转储分析的新手。我不知道出了什么问题。

所以我执行top -H并找到两个最高的线程（都是 99%），并找到线程转储：

基本上这两个线程会将数据发布到 Disruptor。我Disruptor以这种方式创建：

请帮助我并分析线程转储以找到高 CPU 使用率的根本原因。

我有一个问题，如果在 x86 Linux 上运行多个生产者和单个消费者的 lmax 破坏者（如环形缓冲区）中的消费者缓慢，该怎么办。使用类似环形缓冲区的 lmax 模式，您会不断地覆盖数据，但如果消费者速度很慢怎么办。因此，您如何处理在 10 大小的环形缓冲区 0-9 环形槽中，您的消费者位于槽 5 并且现在您的编写者已准备好开始写入槽 15，这也是缓冲区中的槽 5（即：槽5 = 15 % 10)? 处理此问题的典型方法是什么，使得编写者仍然按传入的顺序生成数据，而客户端将以相同的顺序接收数据？这真的是我的问题。下面是关于我的设计的一些细节，它工作正常，只是我目前没有一个好的方法来处理这个问题。

设计细节

我有一个环形缓冲区，设计目前有多个生产者线程和一个消费者线程。这部分设计是存在的，目前无法更改。我正在尝试使用无锁环形缓冲区删除现有的排队系统。我所拥有的如下。

代码在 x86 Linux 上运行，编写器运行多个线程，读取器运行单个线程。读取器和写入器从一个插槽开始，并且是，因此读取器从插槽 0 开始，写入器从插槽 1 开始，然后每个写入器首先通过调用如下所示的方法对写入器序列执行原子操作，然后使用 compare_and_swap 循环std::atomic<uint64_t>来声明一个插槽要更新阅读器序列以让客户端知道此插槽可用，请参阅.fetch_add(1, std::memory_order::memory_order_acq_rel)incrementSequenceupdateSequence

我正在尝试在 java 应用程序中使用 log4j2 和中断器。我的类路径中有以下 jar 文件：

1. log4j-api-2.0-rc2.jar
2. log4j-core-2.0-rc2.jar
3. 破坏者-3.2.0.jar

在我的 Java 类中，我做了以下测试：

我的 log4j2.xml 文件如下：

当我运行应用程序时，我收到以下错误（没有日志输出）：

我正在查看 Google Disruptor 的来源，刚刚发现了一个观察结果：

以下两种方法，它们都在做同样的操作。其中之一是私有的，但它并没有真正回答重复方法的原因。

仅仅是为了程序员的可读性吗？

我开发了一个队列，它允许单个消费者和生产者同时提供/轮询队列中的元素，而无需对每个提供/轮询进行同步或 CAS 操作。相反，当队列的尾部为空时，只需要一个原子操作。该队列旨在减少队列被缓冲并且消费者没有赶上生产者的情况下的延迟。

在这个问题中，我想审查实现（代码还没有被其他人审查过，所以获得第二意见会很好）并讨论一种我认为应该显着减少延迟的使用模式，以及这种架构是否可以可能比 LMAX 干扰器运行得更快。

代码在 github 上： https ://github.com/aranhakki/experimental-performance/blob/master/java/src/concurrency/messaging/ConcurrentPollOfferArrayQueue.java

通过使用这些队列中的许多来代替都引用同一个队列的多个生产者和消费者，我认为延迟可以显着减少。

考虑生产者 A、B、C 都引用单个队列 Q，而消费者 E、E 和 F 都引用同一个队列。这导致了以下一组关系，因此产生了很多争用：

A 写给 Q

B 写给 Q

C 写到 Q

E 写给 Q

D 写给 Q

F 写入 Q

使用我开发的队列，可以在每个生产者和单个消费者聚合线程之间建立一个队列，该线程将获取每个生产者队列尾部的元素并将它们放在消费者队列的头部。这将显着减少争用，因为我们只有一个写入器来写入一段内存。关系船现在看起来如下：

A writeTo headOf(AQ)

B writeTo headOf(BQ)

C writesTo headOf(CQ)

ConsumerAggregationThread writesTo tailOf(AQ)

ConsumerAggregationThread writesTo tailOf(BQ)

ConsumerAggregationThread writesTo tailOf(CQ)

ConsumerAggregationThread writesTo headOf(EQ)

ConsumerAggregationThread writesTo headOf(FQ)

ConsumerAggregationThread writesTo headOf(GQ)

E writeTo tailOf(EQ)

F writeTo tailOf(FQ)

G writeTo tailOf(GQ)

上述关系确保了单写原则。

我很想听听你的想法。

有没有人写过关于如何使用 LMAX Disruptor 3.x 版的完整（不是片段）基本示例？我在官方 wiki 和其他地方看到了 2.x 版本的代码示例，但没有找到 3.x 版本的任何基本示例。有一个使用Scala和 Java替代多生产者实现。我很惊讶我能找到一个替代的例子，却找不到一个基本的例子。到目前为止，我正在使用测试用例来查看各个组件是如何工作的。如果有一个您可以参考的示例，那就太好了，这样您就可以了解所有新组件如何协同工作。提前感谢那些会回复的人。

我在我的 Web 应用程序中使用 Spring 的反应器模式。在内部，它使用LMAX 的 RingBuffer实现作为它的消息队列之一。我想知道是否有任何方法可以动态找出当前的 RingBuffer 占用情况。这将帮助我确定所需的生产者和消费者的数量（以及它们的相对比率），以及作为消息队列的 RingBuffer 是否得到最佳使用。我尝试了reactor.event.dispatch.Abs ​​tractSingleThreadDispatcher类的getBacklog()，但它似乎总是给出相同的值：我在实例化反应器时使用的 RingBuffer 的大小。 任何有关该问题的信息将不胜感激。

客户端（僵尸网络服务器）试图通过 TCP 套接字向服务器（破坏者）发送连续消息，但破坏者只收到一条消息。Disruptor 是由僵尸网络服务器创建的线程。

代码：僵尸网络服务器

破坏者运行（）

我正在为 Disruptor 环形缓冲区中的事件使用 Javolution Struct 类——这似乎是使我的消息对缓存更友好的最简单方法。不过，我以前没有使用 Javolution 的经验，并且几乎没有找到有关调整 Javolution 的教程的资料。

我的问题是：

1. 使用这种方法（即使用 Javolution Structs 作为事件类）我应该注意哪些缺点？

2. 是否有任何我应该注意的特定于 Javolution 的优化？到目前为止，我发现并使用的唯一调整参数是：

   @Override public boolean isPacked() { return true; }

提前致谢

- 编辑 -

给予或接受，课程如下所示：

我试图弄清楚 Disruptor 是否适合我的应用程序：一个相当复杂的负载生成器，用于长时间对高性能数据库进行基准测试。Disruptor 模式之所以吸引人，是因为：

• 高吞吐量
• 低延迟
• 没有垃圾/可预测的长时间运行行为

应用拓扑的简化版本如下：

=== P1 ===

馈送的系统组件P1有 1 个生产者馈送许多并行的消费者（当前是一个Executor）。每个事件都应该由这些消费者之一处理一次。排序约束是：如果 A 在 B 之前产生，那么 A 将在 B 之前开始执行，但是 B 可能先完成。

处理完一个事件后，它会被转发到最后阶段，Cx. 将Cx其视为 reducer/logger：它必须按顺序处理事件，因为它所处理的数据结构不是线程安全的。

问题 1：使用 Disruptor 进行处理的最佳方式是什么C1a-C1e？

使用 aWorkerPool是我的倾向。

=== P2 ===

由 馈送的系统组件P2是一个简单的工作流链：一个事件由 产生P2，C2a对其执行一些计算，对其C2b执行进一步的计算，最后将其转发给Cx。

问题 2：是否可以有 1 个 Disruptor 和 2 个生产者，每个生产者提供不同的依赖子图？

问题 3：如果问题 2 的答案是否定的，是否可以Cx在两个不同的破坏者之间共享一个消费者，如果可以，如何？

问题 4：如果问题 2 和问题 3 的答案都是否定的，有没有办法让这个拓扑与 Disruptor 一起工作？

提前致谢！

- 以供参考 -

这不是解决方案，而是相关的。对于任何感兴趣的人，这个博客解释了如何创建一个半复杂的工作流，例如，像这样：