ArrayBlockingQueue クラスと LinkedBlockingQueue クラスの処理性能を比較するためのデモプログラムです。

BlockingQueueDemo.java

import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * ArrayBlockingQueue クラスと LinkedBlockingQueue クラスの性能を比較するデモプログラムです。
 * 
 * @author KOMIYA Atsushi
 */
public class BlockingQueueDemo {
  public static void main(String[] args) throws InterruptedException {
    final int COUNT = 1024 * 1024 * 16;

    Properties sysProps = System.getProperties();
    System.out.println("java.runtime.version : "
      + sysProps.getProperty("java.runtime.version"));
    System.out.println("sun.arch.data.model : "
      + sysProps.getProperty("sun.arch.data.model"));

    DemoRunner runner = new DemoRunner();

    System.out.println("\nWARMIMG UP");
    runner.run(new DemoAdd(), COUNT);

    System.out.println("\nBENCHMARK -- method invokation");
    runner.run(new DemoAdd(), COUNT);
    runner.run(new DemoOffer(), COUNT);
    runner.run(new DemoPut(), COUNT);
    runner.run(new DemoRemove(), COUNT);
    runner.run(new DemoPoll(), COUNT);
    runner.run(new DemoTake(), COUNT);

    System.out.println("\nBENCHMARK -- throughput");
    runner.run(new DemoProducerConsumer(1, 1), COUNT);
    runner.run(new DemoProducerConsumer(2, 1), COUNT);
    runner.run(new DemoProducerConsumer(1, 2), COUNT);
    runner.run(new DemoProducerConsumer(2, 2), COUNT);
  }

  static class DemoRunner {
    void run(DemoRunnable runnable, int count) throws InterruptedException {
      BlockingQueue<Object> queue = null;

      queue = new ArrayBlockingQueue<Object>(count + 1);
      runnable.prepare(queue, count);

      System.gc();
      Thread.sleep(1000);

      long begin = System.currentTimeMillis();
      runnable.run(queue, count);
      long elapsed1 = System.currentTimeMillis() - begin;

      queue = new LinkedBlockingDeque<Object>(count + 1);
      runnable.prepare(queue, count);

      System.gc();
      Thread.sleep(1000);

      begin = System.currentTimeMillis();
      runnable.run(queue, count);
      long elapsed2 = System.currentTimeMillis() - begin;

      System.out.println("-----");
      System.out.println(runnable.demoName());
      System.out.printf("  ArrayBlockingQueue : %d\n", elapsed1);
      System.out.printf("  LinkedBlockingQueue : %d\n", elapsed2);
    }
  }

  static abstract class DemoRunnable {
    public void prepare(BlockingQueue<Object> queue, int count) {
      // do nothing
    }

    public abstract void run(BlockingQueue<Object> queue, int count)
        throws InterruptedException;

    public abstract String demoName();
  }

  static class DemoAdd extends DemoRunnable {
    @Override
    public void run(BlockingQueue<Object> queue, int count)
        throws InterruptedException {
      Object obj = new Object();
      for (int i = 0; i < count; i++) {
        queue.add(obj);
      }
    }

    @Override
    public String demoName() {
      return "BlockingQueue#add()";
    }
  }

  static class DemoOffer extends DemoRunnable {
    @Override
    public void run(BlockingQueue<Object> queue, int count)
        throws InterruptedException {
      Object obj = new Object();
      for (int i = 0; i < count; i++) {
        queue.offer(obj);
      }
    }

    @Override
    public String demoName() {
      return "BlockingQueue#offer()";
    }
  }

  static class DemoPut extends DemoRunnable {
    @Override
    public void run(BlockingQueue<Object> queue, int count)
        throws InterruptedException {
      Object obj = new Object();
      for (int i = 0; i < count; i++) {
        queue.put(obj);
      }
    }

    @Override
    public String demoName() {
      return "BlockingQueue#put()";
    }
  }

  static class DemoRemove extends DemoRunnable {
    @Override
    public void prepare(BlockingQueue<Object> queue, int count) {
      Object obj = new Object();
      for (int i = 0; i < count; i++) {
        queue.add(obj);
      }
    }

    @Override
    public void run(BlockingQueue<Object> queue, int count)
        throws InterruptedException {
      for (int i = 0; i < count; i++) {
        queue.remove();
      }
    }

    @Override
    public String demoName() {
      return "BlockingQueue#remove()";
    }
  }

  static class DemoPoll extends DemoRunnable {
    @Override
    public void prepare(BlockingQueue<Object> queue, int count) {
      Object obj = new Object();
      for (int i = 0; i < count; i++) {
        queue.add(obj);
      }
    }

    @Override
    public void run(BlockingQueue<Object> queue, int count)
        throws InterruptedException {
      for (int i = 0; i < count; i++) {
        queue.poll();
      }
    }

    @Override
    public String demoName() {
      return "BlockingQueue#poll()";
    }
  }

  static class DemoTake extends DemoRunnable {
    @Override
    public void prepare(BlockingQueue<Object> queue, int count) {
      Object obj = new Object();
      for (int i = 0; i < count; i++) {
        queue.add(obj);
      }
    }

    @Override
    public void run(BlockingQueue<Object> queue, int count)
        throws InterruptedException {
      for (int i = 0; i < count; i++) {
        queue.take();
      }
    }

    @Override
    public String demoName() {
      return "BlockingQueue#take()";
    }
  }

  static class DemoProducerConsumer extends DemoRunnable {
    private static final Object END_MARKER = new Object();
    private Thread[] producers;

    private Thread[] consumers;

    DemoProducerConsumer(int numProducers, int numConsumers) {
      producers = new Thread[numProducers];
      consumers = new Thread[numConsumers];
    }

    @Override
    public void prepare(BlockingQueue<Object> queue, int count) {
      final BlockingQueue<Object> q = queue;
      final int cnt = count;

      for (int i = 0; i < producers.length; i++) {
        producers[i] = new Thread() {
          @Override
          public void run() {
            try {
              Object obj = new Object();
              for (int i = 0; i < cnt; i++) {
                q.put(obj);
              }
            } catch (Exception e) {
              throw new RuntimeException(e);
            }
          }
        };
      }

      for (int i = 0; i < consumers.length; i++) {
        consumers[i] = new Thread() {
          @Override
          public void run() {
            try {
              while (true) {
                if (q.take() == END_MARKER) {
                  break;
                }
              }
            } catch (Exception e) {
              throw new RuntimeException(e);
            }
          }
        };
      }
    }

    @Override
    public void run(BlockingQueue<Object> queue, int count)
        throws InterruptedException {
      for (int i = 0; i < producers.length; i++) {
        producers[i].start();
      }
      for (int i = 0; i < consumers.length; i++) {
        consumers[i].start();
      }

      for (int i = 0; i < producers.length; i++) {
        producers[i].join();
      }
      for (int i = 0; i < consumers.length; i++) {
        queue.put(END_MARKER);
      }
      for (int i = 0; i < consumers.length; i++) {
        consumers[i].join();
      }
    }

    @Override
    public String demoName() {
      return String.format(
        "Producer(%d) / Consumer(%d)",
        producers.length,
        consumers.length);
    }
  }
}