代码之家 › 专栏 › 技术社区 › Duncan Krebs

在线程池中重用Runnables有意义吗?

runnable threadpool garbage-collection multithreading java

Duncan Krebs · 技术社区 · 11 年前

我正在实现一个用于处理大量市场数据馈送的线程池,并对重用我的工作实例的策略有疑问,这些工作实例实现了提交给线程池执行的可运行实例。在我的案例中,我只有一种类型的工作程序,它接受一个String并解析它以创建一个Quote对象,然后在正确的Security上设置该对象。考虑到来自提要的数据量,每秒可能有1000多个引号要处理,我认为有两种方法可以创建提交到线程池的工作线程。

第一个选项是,每当从底层套接字中检索到一行时,只需创建一个Worker的新实例,然后将其添加到线程池中,线程池最终将在其运行方法执行后被垃圾收集。但这让我想到了性能,每秒实例化10000个Worker类的新实例真的有意义吗。本着与线程池相同的精神,人们知道拥有一个可运行的池或队列是否是一种常见的模式,这样我就可以回收我的工作人员,以避免创建对象和垃圾收集。我看到的实现方式是,在返回run()方法之前,Worker将自己添加回可用Worker的队列,然后在处理新的馈线而不是创建Worker的新实例时从中提取。

从性能的角度来看,采用第二种方法我会有所收获吗?还是采用第一种方法更有意义?以前有人实现过这种模式吗?

谢谢-邓肯

5 回复 | 直到 11 年前

Peter Lawrey 11 年前

为此,我使用了一个名为Java Chronicle的库。它被设计为每秒持久化和排队一百万次报价,而不会产生任何重大垃圾。

我有一个演示 here 它以每秒一百万条消息的速度发送具有纳秒定时信息的类似报价的对象,并且它可以在具有32MB堆的JVM中发送数千万条消息,甚至不触发一个小的收集。往返延迟小于0.6微秒,在我的ultra book上90%的时间都是这样。)

从性能的角度来看,采用第二种方法我会有所收获吗?还是采用第一种方法更有意义?

我强烈建议不要用垃圾填充CPU缓存。事实上,我避免任何会产生任何重要垃圾的构造。您可以构建一个系统,使每个事件创建的对象少于一个 端到端 。我有一个伊甸园大小,比我一天产生的垃圾量还大,所以不用担心轻微或严重的GC。

以前有人实现过这种模式吗?

五年前,我用Java编写了一个有利可图的低延迟交易系统。当时,用Java交易的速度已经快到60微秒,但现在你可以做得更好。

如果你想要低延迟的市场数据处理系统,我就是这样做的。你可能会发现我在JavaOne上做的这个演示也很有趣。

http://www.slideshare.net/PeterLawrey/writing-and-testing-high-frequency-trading-engines-in-java

编辑我已经添加了这个 parsing example

ByteBuffer wrap = ByteBuffer.allocate(1024);
ByteBufferBytes bufferBytes = new ByteBufferBytes(wrap);
byte[] bytes = "BAC,12.32,12.54,12.56,232443".getBytes();

int runs = 10000000;
long start = System.nanoTime();
for (int i = 0; i < runs; i++) {
    bufferBytes.reset();
    // read the next message.
    bufferBytes.write(bytes);
    bufferBytes.position(0);
    // decode message
    String word = bufferBytes.parseUTF(StopCharTesters.COMMA_STOP);
    double low = bufferBytes.parseDouble();
    double curr = bufferBytes.parseDouble();
    double high = bufferBytes.parseDouble();
    long sequence = bufferBytes.parseLong();
    if (i == 0) {
        assertEquals("BAC", word);
        assertEquals(12.32, low, 0.0);
        assertEquals(12.54, curr, 0.0);
        assertEquals(12.56, high, 0.0);
        assertEquals(232443, sequence);
    }
}
long time = System.nanoTime() - start;
System.out.println("Average time was " + time / runs + " nano-seconds");

当设置为-verbose:gc-Xmx32m时,它将打印

Average time was 226 nano-seconds

注意:没有触发GC。

duffymo 11 年前

我会用 Executor 来自并发程序包。我相信它能为你处理这一切。

Gray droiddeveloper 11 年前

每秒实例化10000个Worker类的新实例真的有意义吗。

不一定,但是你必须把 Runnable s变成了某种 BlockingQueue 以便能够被重用,并且队列并发的成本可能超过GC开销。使用探查器或通过Jconsole查看GC编号将告诉您它是否在GC中花费了大量时间,并且需要解决这一问题。

如果这确实是一个问题,那么另一种方法是 String 变成你自己的 阻止队列 并提交 Worker 对象只添加到线程池一次。每个 工人 实例将从的队列中退出 一串 永远不会放弃。类似于:

public void run() {
    while (!shutdown) {
        String value = myQueue.take();
        ...
    }
}

所以你不需要创建你的1000 工人 s每秒。

user2511414 user2511414 11 年前

是的,当然,有点像 this ,因为操作系统和JVM不关心线程上发生了什么,所以通常这是重用可回收对象的好做法。

codethulhu 11 年前

我看到你的问题中有两个问题。一个是线程池,另一个是对象池。对于线程池问题,Java提供了 ExecutorService 。以下是使用ExecutorService的示例。

Runnable r = new Runnable() {
    public void run() {
        //Do some work
    }
};

// Thread pool of size 2
ExecutorService executor = Executors.newFixedThreadPool(2);
// Add the runnables to the executor service
executor.execute(r);

ExecutorService为许多不同类型的线程池提供了不同的行为。

就对象池化而言,(每秒创建1000个对象,然后将它们留给垃圾收集有意义吗?这一切都取决于对象的状态和开销。如果你担心工作线程的状态会受到影响,你可以考虑使用轻量级模式将你的状态封装在工作线程之外。此外,如果你要遵循 flyweight pattern ,你也可以看看有多有用 Future 和 Callable 对象将在您的应用程序体系结构中。