Linux上的Java BlockingQueue延迟很高

4 回答

• 3

  如果延迟是关键的并且您不需要严格的FIFO语义，那么您可能需要考虑JSR-166的LinkedTransferQueue . 它可以消除，以便相反的操作可以交换值而不是在队列数据结构上进行同步 . 这种方法有助于减少争用，实现并行交换，并避免线程休眠/唤醒惩罚 .

  回复于 2024-04-25T09:59:10+08:00
• 6

  您的测试不是衡量队列切换延迟的一个好方法，因为您有一个线程读取队列，该队列在再次执行之前同步写入 System.out （执行字符串和长连接） . 要正确测量这一点，您需要将此活动移出此线程，并在获取线程中尽可能少地工作 .

  你最好只在接受者中进行计算（然后现在）并将结果添加到其他集合中，该集合由另一个输出结果的线程定期排干 . 我倾向于通过添加到通过AtomicReference访问的适当规定的数组支持结构来实现这一点（因此报告线程只需要在该引用上使用该存储结构的另一个实例的getAndSet来获取最新批次的结果;例如make 2列表，将一个设置为活动，每个xsa线程唤醒并交换主动和被动的线程） . 然后，您可以报告某些分布而不是每个结果（例如，十分位数范围），这意味着您不会在每次运行时生成大量日志文件并获取为您打印的有用信息 .

  FWIW我同意Peter Lawrey所说的时间，如果延迟真的很关键，那么你需要考虑忙于等待适当的cpu亲和力（即将核心专用于该线程）

  EDIT after Jan 6

  如果我删除对Thread.sleep（）的调用，而是让 生产环境 者和消费者在每次迭代中调用barrier.await（）（消费者在将经过的时间打印到控制台后调用它），则测量的延迟会减少从60微秒到10微秒以下 . 如果在同一核心上运行线程，则延迟低于1微秒 . 任何人都可以解释为什么这会显着减少延迟？

  您正在查看 java.util.concurrent.locks.LockSupport#park （和相应的 unpark ）和 Thread#sleep 之间的区别 . 大多数j.u.c. stuff Build 在 LockSupport 上（通常通过_235960_提供或直接提供），这个（在Hotspot中）解析为 sun.misc.Unsafe#park （和 unpark ），这往往最终落在pthread（posix线程）lib的手中 . 通常 pthread_cond_broadcast 唤醒， pthread_cond_wait 或 pthread_cond_timedwait 为 BlockingQueue#take 之类的东西 .

  我可以't say I'曾经看过 Thread#sleep 实际上是如何实现的（但是我会想到它会导致它被调度程序以比pthread信令机制更激进的方式降级，这就是什么帐户对于延迟差异 .

  回复于 2024-04-25T09:59:10+08:00
• 0

  如果可以，我会使用一个ArrayBlockingQueue . 当我使用它时，Linux上的延迟在8-18微秒之间 . 有些注意事项 .

  • 成本主要是唤醒线程所需的时间 . 当你唤醒一个线程时，它的数据/代码不会在缓存中，所以你会发现，如果你计算线程唤醒后发生的事情的时间比你重复运行相同的事情要长2-5倍 .

  • 某些操作使用OS调用（例如锁定/循环障碍），这些操作在低延迟情况下通常比繁忙等待更昂贵 . 我建议你努力等待你的制作人而不是使用CyclicBarrier . 您也可以忙着等待您的消费者，但这在真实系统上可能会非常昂贵 .

  回复于 2024-04-25T09:59:10+08:00
• 1

  @Peter Lawrey

  某些操作使用OS调用（例如锁定/循环障碍）

  那些不是OS（内核）调用 . 通过简单的CAS实现（在x86上也带有免费的内存栅栏）

  还有一个：除非你知道为什么（你使用它），否则不要使用ArrayBlockingQueue .

  @OP：看看ThreadPoolExecutor，它提供了出色的 生产环境 者/消费者框架 .

  编辑如下：

  要减少延迟（限制繁忙等待），将队列更改为SynchronousQueue，在启动使用者之前添加以下内容

  ...
  consumerThread.setPriority(Thread.MAX_PRIORITY);
  consumerThread.start();
  

  这是你能得到的最好的 .

  ---

  Edit2：这里有同步 . 队列 . 而不是打印结果 .

  package t1;
  
  import java.math.BigDecimal;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.SynchronousQueue;
  
  public class QueueTest {
  
      static final int RUNS = 250000;
  
      final SynchronousQueue<Long> queue = new SynchronousQueue<Long>();
  
      int sleep = 1000;
  
      long[] results  = new long[0];
      public void start(final int runs) throws Exception {
          results = new long[runs];
          final CountDownLatch barrier = new CountDownLatch(1);
          Thread consumerThread = new Thread(new Runnable() {
              @Override
              public void run() {
                  barrier.countDown();
                  try {
  
                      for(int i = 0; i < runs; i++) {                        
                          results[i] = consume(); 
  
                      }
                  } catch (Exception e) {
                      return;
                  } 
              }
          });
          consumerThread.setPriority(Thread.MAX_PRIORITY);
          consumerThread.start();
  
  
          barrier.await();
          final long sleep = this.sleep;
          for(int i = 0; i < runs; i++) {
              try {                
                  doProduce(sleep);
  
              } catch (Exception e) {
                  return;
              }
          }
      }
  
      private void doProduce(final long sleep) throws InterruptedException {
          produce();
      }
  
      public void produce() throws InterruptedException {
          queue.put(new Long(System.nanoTime()));//new Long() is faster than value of
      }
  
      public long consume() throws InterruptedException {
          long t = queue.take();
          long now = System.nanoTime();
          return now-t;
      }
  
      public static void main(String[] args) throws Throwable {           
          QueueTest test = new QueueTest();
          System.out.println("Starting + warming up...");
          // Run first once, ignoring results
          test.sleep = 0;
          test.start(15000);//10k is the normal warm-up for -server hotspot
          // Run again, printing the results
          System.gc();
          System.out.println("Starting again...");
          test.sleep = 1000;//ignored now
          Thread.yield();
          test.start(RUNS);
          long sum = 0;
          for (long elapsed: test.results){
              sum+=elapsed;
          }
          BigDecimal elapsed = BigDecimal.valueOf(sum, 3).divide(BigDecimal.valueOf(test.results.length), BigDecimal.ROUND_HALF_UP);        
          System.out.printf("Avg: %1.3f micros%n", elapsed); 
      }
  }
  

  回复于 2024-04-25T09:59:10+08:00