问题标签 [disruptor-pattern]

我正在寻找使用 dirsuptor 环形缓冲区来解析文件。但我看不到如何为环形缓冲区设置一个值范围。

在下面的示例中似乎。它循环到每个项目以将其分配给缓冲区。但是我，我想直接分配 x 项。

当我执行 FileInputStream.read( byte[] bytes ) 时，我想将这些字节放入环形缓冲区。

通常我的缓冲区比我读取的字节大两倍。像这样我可以在计算一个页面时读取另一页（例如字节数组长度==缓冲区/2）：

谢谢

使用 LMAX Disruptor，我们观察到，如果我们在一个应用程序中一起使用 5-10 个干扰器（有点像一个干扰器链，每个干扰器都有一个消费者执行指定的任务，然后将消息移交给下一个干扰器/ ringbuffer), 发生的情况是 CPU 利用率达到 90% 及以上，系统变得无响应，直到我们关闭应用程序，我们觉得这是因为有这么多活跃的中断线程。即使破坏者没有真正处理任何事情，也会发生这种情况。任何人都可以评论在应用程序中使用的最佳干扰器数量是多少？

应用程序不断接收命名Report的对象并将对象放入Disruptor三个不同的消费者。

在 Eclipse Memory Analysis 的帮助下，每个Report对象的 Retained Heap Size 平均为 20KB。应用以 开头-Xmx2048，表示应用的堆大小为 2GB。

但是，一次对象的数量约为 100,000 个，这意味着所有对象的总大小约为 2GB。

要求是所有 100,000 个对象都应加载到Disruptor其中，以便消费者异步使用数据。但是如果每个对象的大小都大到 20KB 是不可能的。

所以我想将对象序列化String并压缩它：

之后compress(toBytes(Report))，对象尺寸变小：

压缩前

压缩后

现在对象的字符串大约是 6KB。现在好多了。

这是我的问题：

1. 是否有任何其他数据格式的大小小于字符串？

2. 每次调用序列化和压缩都会创建对象ByteArrayOutputStream，ObjectOutputStream诸如此类。我不想创建很多类似的对象ByteArrayOutputStream，ObjectOutputStream因为我需要迭代 100,000 次。如何设计代码以便类似的对象ByteArrayOutputStream只ObjectOutputStream创建一次并在每次迭代中使用它？

3. 消费者需要反序列化和解压缩来自Disruptor. 如果我有三个消费者，那么我需要反序列化和解压缩三次。有什么办法吗？

更新：

正如@BoristheSpider 所建议的，序列化和压缩应该在一个动作中执行：

当我运行这段代码

我得到这个输出

但我希望每个事件只由一个单独的线程处理一次。我怎样才能让它发生？

其中一个EventHandler（DatabaseConsumer）Disruptor调用数据库中的存储过程，它太慢以至于阻塞了Disruptor一段时间。

因为我需要Disruptor继续运行而不会阻塞。我正在考虑添加一个额外的队列，以便EventHandler可以用作Producer另一个新创建的线程可以用作Consumer处理数据库的工作，这可以是异步的而不影响Disruptor

这是一些约束：

1. Disruptor传递给 的对象EventHandler大约是 30KB，这个对象的数量大约是 400k。理论上，需要处理的对象的总大小在 30KBX400K =12GB 左右。所以额外的队列对他们来说应该足够了。
2. 由于性能很重要，应该避免 GC 暂停。
3. Java 程序的堆大小只有 2GB。

我正在考虑将文本文件作为一种选择。EventHandler（生产者）将对象写入文件并Consumer从中读取并调用存储过程。问题是如何处理它到达文件末尾的情况以及如何知道新的行。

有谁以前解决过这种情况？有什么建议吗？

我正在构建一个简单的反应式服务器，它应该使用来自多个客户端的传入 protobuf/protostuff 消息，在它们上执行一些业务逻辑，并可能向其他消费者发送即发即弃的消息。我想在 Netty 中实现传输和解码部分。我的问题是：将解码消息发布到 Disruptor 的环形缓冲区性能方面是否有任何意义，或者 Disruptor 提供的额外性能会被内部 Netty 调度所否定？我应该为 Netty 提供两个线程（一个用于“接受”，另一个用于“连接”组）还是只有一个更好？也许，我应该在 Netty 的处理程序中按长度字段拆分消息，并在 Disruptor 的处理程序中执行解码？

无论如何，消费者是否会通知生产者任何特定事件已成功处理？

这个数字是否会影响 IO 操作？或者它只是架构可以匹配的事务数量，只对内存进行读写？

他们在测试期间使用了哪些硬件？可能无关紧要，但他们使用的是哪个操作系统？

LMAX Disruptor 通常使用以下方法实现：

如本例所示，Replicator 负责将输入事件\命令复制到从节点。跨一组节点进行复制需要我们应用共识算法，以防我们希望系统在出现网络故障、主故障和从故障时可用。

我正在考虑将 RAFT 共识算法应用于这个问题。一项观察是：“RAFT 要求在复制期间将输入事件\命令存储到磁盘（持久存储）”（参考此链接）

这种观察本质上意味着我们无法执行内存复制。因此，我们可能必须结合复制器和日志器的功能才能成功地将 RAFT 算法应用于 LMAX。

有两种方法可以做到这一点：

选项 1：使用复制的日志作为输入事件队列

• 接收者将从网络读取并将事件推送到复制的日志而不是环形缓冲区
• 一个单独的“阅读器”可以从日志中读取并将事件发布到环形缓冲区。
• 可以使用 RAFT 跨节点复制日志。我们不需要复制器和日志器，因为功能已经由 RAFT 的复制日志完成

我认为这个选项的一个缺点与我们做了一个额外的数据复制步骤（接收器到事件队列而不是环形缓冲区）有关。

选项 2：使用 Replicator 将输入事件\命令推送到从属的输入日志文件

我想知道 Replicator 的设计是否还有其他解决方案？人们为复制器采用了哪些不同的设计选项？特别是任何可以支持内存复制的设计？

我想有一个环形缓冲区队列，它将接收对象并将它们分布在线程池中的多个线程中，以单个生产者到多个消费者的方式。我如何使用破坏者模式来实现这一点？任何HelloDemux代码示例？谢谢！！！