Java 类java.util.concurrent.LinkedBlockingDeque 实例源码

public AnomalyDetector(KafkaCruiseControlConfig config,
                       LoadMonitor loadMonitor,
                       KafkaCruiseControl kafkaCruiseControl,
                       Time time,
                       MetricRegistry dropwizardMetricRegistry) {
  _anomalies = new LinkedBlockingDeque<>();
  _anomalyDetectionIntervalMs = config.getLong(KafkaCruiseControlConfig.ANOMALY_DETECTION_INTERVAL_MS_CONFIG);
  _anomalyNotifier = config.getConfiguredInstance(KafkaCruiseControlConfig.ANOMALY_NOTIFIER_CLASS_CONFIG,
                                                  AnomalyNotifier.class);
  _goalViolationDetector = new GoalViolationDetector(config, loadMonitor, _anomalies, time);
  _brokerFailureDetector = new BrokerFailureDetector(config, loadMonitor, _anomalies, time);
  _kafkaCruiseControl = kafkaCruiseControl;
  _detectorScheduler =
      Executors.newScheduledThreadPool(3, new KafkaCruiseControlThreadFactory("AnomalyDetector", false, LOG));
  _shutdown = false;
  _brokerFailureRate = dropwizardMetricRegistry.meter(MetricRegistry.name("AnomalyDetector", "broker-failure-rate"));
  _goalViolationRate = dropwizardMetricRegistry.meter(MetricRegistry.name("AnomalyDetector", "goal-violation-rate"));

}

/**
 * Package private constructor for unit test.
 */
AnomalyDetector(LinkedBlockingDeque<Anomaly> anomalies,
                long anomalyDetectionIntervalMs,
                KafkaCruiseControl kafkaCruiseControl,
                AnomalyNotifier anomalyNotifier,
                GoalViolationDetector goalViolationDetector,
                BrokerFailureDetector brokerFailureDetector,
                ScheduledExecutorService detectorScheduler) {
  _anomalies = anomalies;
  _anomalyDetectionIntervalMs = anomalyDetectionIntervalMs;
  _anomalyNotifier = anomalyNotifier;
  _goalViolationDetector = goalViolationDetector;
  _brokerFailureDetector = brokerFailureDetector;
  _kafkaCruiseControl = kafkaCruiseControl;
  _detectorScheduler = detectorScheduler;
  _shutdown = false;
  _brokerFailureRate = new Meter();
  _goalViolationRate = new Meter();
}

/**
 * Descending iterator ordering is reverse FIFO
 */
public void testDescendingIteratorOrdering() {
    final LinkedBlockingDeque q = new LinkedBlockingDeque();
    for (int iters = 0; iters < 100; ++iters) {
        q.add(new Integer(3));
        q.add(new Integer(2));
        q.add(new Integer(1));
        int k = 0;
        for (Iterator it = q.descendingIterator(); it.hasNext();) {
            assertEquals(++k, it.next());
        }

        assertEquals(3, k);
        q.remove();
        q.remove();
        q.remove();
    }
}

public static void main(String[] args) throws Exception {
     final int maxPairs = (args.length > 0)
         ? Integer.parseInt(args[0])
         : 5;
     int iters = 10000;

     pool = Executors.newCachedThreadPool();
     for (int i = 1; i <= maxPairs; i += (i+1) >>> 1) {
         // Adjust iterations to limit typical single runs to <= 10 ms;
         // Notably, fair queues get fewer iters.
         // Unbounded queues can legitimately OOME if iterations
         // high enough, but we have a sufficiently low limit here.
         run(new ArrayBlockingQueue<Integer>(100), i, 500);
         run(new LinkedBlockingQueue<Integer>(100), i, 1000);
         run(new LinkedBlockingDeque<Integer>(100), i, 1000);
         run(new LinkedTransferQueue<Integer>(), i, 1000);
         run(new PriorityBlockingQueue<Integer>(), i, 1000);
         run(new SynchronousQueue<Integer>(), i, 400);
         run(new SynchronousQueue<Integer>(true), i, 300);
         run(new ArrayBlockingQueue<Integer>(100, true), i, 100);
     }
     pool.shutdown();
     if (! pool.awaitTermination(LONG_DELAY_MS, MILLISECONDS))
         throw new Error();
     pool = null;
}

/**
 * descendingIterator.remove removes current element
 */
public void testDescendingIteratorRemove() {
    final LinkedBlockingDeque q = new LinkedBlockingDeque();
    for (int iters = 0; iters < 100; ++iters) {
        q.add(new Integer(3));
        q.add(new Integer(2));
        q.add(new Integer(1));
        Iterator it = q.descendingIterator();
        assertEquals(it.next(), new Integer(1));
        it.remove();
        assertEquals(it.next(), new Integer(2));
        it = q.descendingIterator();
        assertEquals(it.next(), new Integer(2));
        assertEquals(it.next(), new Integer(3));
        it.remove();
        assertFalse(it.hasNext());
        q.remove();
    }
}

static void putAllCollections(Map<Class<?>, IntFunction<?>> map, Map<Class<?>, Function<?, ?>> unmodMap)
{
    safePut(map, ArrayList.class, ArrayList::new);
    safePut(map, HashSet.class, LinkedHashSet::new);
    safePut(map, Properties.class, x -> new Properties());
    safePut(map, Hashtable.class, Hashtable::new);

    safePut(map, Collection.class, ArrayList::new);
    safePut(map, Set.class, LinkedHashSet::new);
    safePut(map, List.class, ArrayList::new);
    safePut(map, SortedSet.class, x -> new TreeSet<>());
    safePut(map, Queue.class, x -> new ConcurrentLinkedQueue<>());
    safePut(map, Deque.class, x -> new ConcurrentLinkedDeque<>());
    safePut(map, BlockingQueue.class, x -> new LinkedBlockingQueue<>());
    safePut(map, BlockingDeque.class, x -> new LinkedBlockingDeque<>());


    safePut(map, HashMap.class, LinkedHashMap::new);
    safePut(map, LinkedHashMap.class, LinkedHashMap::new);
    safePut(map, ConcurrentHashMap.class, ConcurrentHashMap::new);

    safePut(map, Map.class, LinkedHashMap::new);
    safePut(map, ConcurrentMap.class, x -> new ConcurrentSkipListMap<>());
    safePut(map, ConcurrentNavigableMap.class, x -> new ConcurrentSkipListMap<>());
    safePut(map, SortedMap.class, i -> new TreeMap<>());
}

public static void main(String[] args) throws Exception {
     final int maxConsumers = (args.length > 0)
         ? Integer.parseInt(args[0])
         : 5;

     pool = Executors.newCachedThreadPool();
     for (int i = 1; i <= maxConsumers; i += (i+1) >>> 1) {
         // Adjust iterations to limit typical single runs to <= 10 ms;
         // Notably, fair queues get fewer iters.
         // Unbounded queues can legitimately OOME if iterations
         // high enough, but we have a sufficiently low limit here.
         run(new ArrayBlockingQueue<Integer>(100), i, 1000);
         run(new LinkedBlockingQueue<Integer>(100), i, 1000);
         run(new LinkedBlockingDeque<Integer>(100), i, 1000);
         run(new LinkedTransferQueue<Integer>(), i, 700);
         run(new PriorityBlockingQueue<Integer>(), i, 1000);
         run(new SynchronousQueue<Integer>(), i, 300);
         run(new SynchronousQueue<Integer>(true), i, 200);
         run(new ArrayBlockingQueue<Integer>(100, true), i, 100);
     }
     pool.shutdown();
     if (! pool.awaitTermination(LONG_DELAY_MS, MILLISECONDS))
         throw new Error();
     pool = null;
}

/**
 * offer transfers elements across Executor tasks
 */
public void testOfferInExecutor() {
    final LinkedBlockingDeque q = new LinkedBlockingDeque(2);
    q.add(one);
    q.add(two);
    final CheckedBarrier threadsStarted = new CheckedBarrier(2);
    final ExecutorService executor = Executors.newFixedThreadPool(2);
    try (PoolCleaner cleaner = cleaner(executor)) {
        executor.execute(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                assertFalse(q.offer(three));
                threadsStarted.await();
                assertTrue(q.offer(three, LONG_DELAY_MS, MILLISECONDS));
                assertEquals(0, q.remainingCapacity());
            }});

        executor.execute(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                threadsStarted.await();
                assertSame(one, q.take());
            }});
    }
}

Collection<Queue<Boolean>> concurrentQueues() {
    List<Queue<Boolean>> queues = new ArrayList<>();
    queues.add(new ConcurrentLinkedDeque<Boolean>());
    queues.add(new ConcurrentLinkedQueue<Boolean>());
    queues.add(new ArrayBlockingQueue<Boolean>(count, false));
    queues.add(new ArrayBlockingQueue<Boolean>(count, true));
    queues.add(new LinkedBlockingQueue<Boolean>());
    queues.add(new LinkedBlockingDeque<Boolean>());
    queues.add(new LinkedTransferQueue<Boolean>());

    // Following additional implementations are available from:
    // http://gee.cs.oswego.edu/dl/concurrency-interest/index.html
    // queues.add(new SynchronizedLinkedListQueue<Boolean>());

    // Avoid "first fast, second slow" benchmark effect.
    Collections.shuffle(queues);
    return queues;
}

private <T> Collection<T> instantiateCollectionFromInterface(Class<? extends T> collectionType) {
    if (List.class.isAssignableFrom(collectionType)) {
        return new ArrayList<T>();
    } else if (SortedSet.class.isAssignableFrom(collectionType)) {
        return new TreeSet<T>();
    } else if (Set.class.isAssignableFrom(collectionType)) {
        return new LinkedHashSet<T>();
    } else if (BlockingDeque.class.isAssignableFrom(collectionType)) {
        return new LinkedBlockingDeque<T>();
    } else if (Deque.class.isAssignableFrom(collectionType)) {
        return new ArrayDeque<T>();
    } else if (BlockingQueue.class.isAssignableFrom(collectionType)) {
        return new LinkedBlockingDeque<T>();
    } else if (Queue.class.isAssignableFrom(collectionType)) {
        return new LinkedList<T>();
    }
    return new ArrayList<T>();
}

private <T> Collection<T> instantiateCollectionFromInterface(Class<? extends T> collectionType) {
    if (List.class.isAssignableFrom(collectionType)) {
        return new ArrayList<T>();
    } else if (SortedSet.class.isAssignableFrom(collectionType)) {
        return new TreeSet<T>();
    } else if (Set.class.isAssignableFrom(collectionType)) {
        return new LinkedHashSet<T>();
    } else if (BlockingDeque.class.isAssignableFrom(collectionType)) {
        return new LinkedBlockingDeque<T>();
    } else if (Deque.class.isAssignableFrom(collectionType)) {
        return new ArrayDeque<T>();
    } else if (BlockingQueue.class.isAssignableFrom(collectionType)) {
        return new LinkedBlockingDeque<T>();
    } else if (Queue.class.isAssignableFrom(collectionType)) {
        return new LinkedList<T>();
    }
    return new ArrayList<T>();
}

/**
 * Initializing from Collection of null elements throws NullPointerException
 */
public void testConstructor4() {
    Collection<Integer> elements = Arrays.asList(new Integer[SIZE]);
    try {
        new LinkedBlockingDeque(elements);
        shouldThrow();
    } catch (NullPointerException success) {}
}

/**
 * timed pollLast with zero timeout succeeds when non-empty, else times out
 */
public void testTimedPollLast0() throws InterruptedException {
    LinkedBlockingDeque q = populatedDeque(SIZE);
    for (int i = 0; i < SIZE; ++i) {
        assertEquals(SIZE - i - 1, q.pollLast(0, MILLISECONDS));
    }
    assertNull(q.pollLast(0, MILLISECONDS));
}

/**
 * Deque contains all elements, in traversal order, of successful addAll
 */
public void testAddAll5() {
    Integer[] empty = new Integer[0];
    Integer[] ints = new Integer[SIZE];
    for (int i = 0; i < SIZE; ++i)
        ints[i] = new Integer(i);
    LinkedBlockingDeque q = new LinkedBlockingDeque(SIZE);
    assertFalse(q.addAll(Arrays.asList(empty)));
    assertTrue(q.addAll(Arrays.asList(ints)));
    for (int i = 0; i < SIZE; ++i)
        assertEquals(ints[i], q.poll());
}

public TCPClient(String name, TCPResultListener mTCPResultListener){
    this.name = name;
    this.mTCPResultListener = mTCPResultListener;
    mEventBlockDeque = new LinkedBlockingDeque<>();
    mExit = new AtomicBoolean(false);
    mIOThread = new IOThread(name);
}

@Override
public Pack createPack(C config) {
    requireNonNull(config, "required non null config");
    final String id = config.getUniqueId();
    if (packs.containsKey(id))
        return packs.get(id);
    final Pack pack = new Pack(id, config.numberOfItems());
    packs.put(id, pack);
    final BlockingDeque<PackItem> deque = new LinkedBlockingDeque<>();
    executorService.submit(() -> repository.getPackItems(config, deque));
    pack.processing();
    executorService.submit(() -> packer.pack(deque, packsDir, pack));
    return pack;
}

/**
 * Constructor
 * @param partitionId
 * @param maxWaitTime
 * @param throttle_threshold TODO
 * @param throttle_release TODO
 * @param hstore_site
 */
public PartitionLockQueue(int partitionId, int maxWaitTime, int throttle_threshold, double throttle_release) {
    super(new LinkedBlockingDeque<AbstractTransaction>(), throttle_threshold, throttle_release);

    this.partitionId = partitionId;
    this.maxWaitTime = maxWaitTime;

    if (HStoreConf.singleton().site.queue_profiling) {
        this.profiler = new PartitionLockQueueProfiler();
    } else {
        this.profiler = null;
    }
}

/**
 * @param hstore_site
 */
public AbstractDispatcher(HStoreSite hstore_site, HStoreCoordinator hstore_coordinator) {
    super(hstore_site,
          "dispatcher",
          new LinkedBlockingDeque<E>(),
          hstore_site.getHStoreConf().site.exec_profiling);
    this.hstore_coordinator = hstore_coordinator;
}

public MapReduceHelperThread(HStoreSite hstore_site) {
    super(hstore_site,
          HStoreConstants.THREAD_NAME_MAPREDUCE,
          new LinkedBlockingDeque<MapReduceTransaction>(),
          false);
    this.p_estimator = hstore_site.getPartitionEstimator();
}

private void calculateTxnProfileTotals(Procedure catalog_proc, long totals[]) {
    long tuple[] = null;
    LinkedBlockingDeque<long[]> queue = this.txn_profile_queues.get(catalog_proc); 
    while ((tuple = queue.poll()) != null) {
        totals[0]++;
        for (int i = 0, cnt = tuple.length; i < cnt; i++) {
            totals[i+1] += tuple[i];
        } // FOR
    } // FOR
}

/**
 * pollFirst succeeds unless empty
 */
public void testPollFirst() {
    LinkedBlockingDeque q = populatedDeque(SIZE);
    for (int i = 0; i < SIZE; ++i) {
        assertEquals(i, q.pollFirst());
    }
    assertNull(q.pollFirst());
}

/**
 * Creates a {@code LinkedBlockingDeque} with a capacity of {@link Integer#MAX_VALUE},
 * containing the elements of the specified iterable,
 * in the order they are returned by the iterable's iterator.
 *
 * @since 12.0
 */
public static <E> LinkedBlockingDeque<E> newLinkedBlockingDeque(Iterable<? extends E> elements) {
  if (elements instanceof Collection) {
    return new LinkedBlockingDeque<E>(Collections2.cast(elements));
  }
  LinkedBlockingDeque<E> deque = new LinkedBlockingDeque<E>();
  Iterables.addAll(deque, elements);
  return deque;
}

/**
 * removeLastOccurrence(x) removes x and returns true if present
 */
public void testRemoveLastOccurrence() {
    LinkedBlockingDeque q = populatedDeque(SIZE);
    for (int i = 1; i < SIZE; i += 2) {
        assertTrue(q.removeLastOccurrence(new Integer(i)));
    }
    for (int i = 0; i < SIZE; i += 2) {
        assertTrue(q.removeLastOccurrence(new Integer(i)));
        assertFalse(q.removeLastOccurrence(new Integer(i + 1)));
    }
    assertTrue(q.isEmpty());
}

/**
 * putFirst(null) throws NPE
 */
public void testPutFirstNull() throws InterruptedException {
    LinkedBlockingDeque q = new LinkedBlockingDeque(SIZE);
    try {
        q.putFirst(null);
        shouldThrow();
    } catch (NullPointerException success) {}
}

/**
 * removeLast() removes last element, or throws NSEE if empty
 */
public void testRemoveLast() {
    LinkedBlockingDeque q = populatedDeque(SIZE);
    for (int i = SIZE - 1; i >= 0; --i) {
        assertEquals(i, q.removeLast());
    }
    try {
        q.removeLast();
        shouldThrow();
    } catch (NoSuchElementException success) {}
    assertNull(q.peekLast());
}

/**
 * 构建文件下载器,适用于下载单个大文件
 *
 * @param downloadUrl 下载路径
 * @param fileSaveDir 文件保存目录
 * @param threadNum   下载线程数
 */
public DownloadTask(Context context, String downloadUrl, File fileSaveDir, int threadNum) {
    try {
        System.out.println("DownloadTask>>>" + downloadUrl);
        this.context = context;
        this.downloadUrl = downloadUrl;
        fileService = FileService.getInstance();
        URL url = new URL(this.downloadUrl);
        if (!fileSaveDir.exists())
            fileSaveDir.mkdirs();
        this.threadnum = threadNum;
        threadPool = new ThreadPoolExecutor(threadnum + 1, threadnum + 1, 20, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>(), new ThreadPoolExecutor.CallerRunsPolicy());
        HttpURLConnection conn = getConnectionAndConnect(url, 3);
        this.fileSize = conn.getContentLength();//根据响应获取文件大小
        if (this.fileSize <= 0)
            throw new RuntimeException("Unkown file size ");

        String filename = getFileName(conn);
        this.saveFile = new File(fileSaveDir, filename);/* 保存文件 */
        Map<Integer, Integer> logdata = fileService.getData(downloadUrl);
        if (logdata.size() > 0) {
            for (Map.Entry<Integer, Integer> entry : logdata.entrySet())
                data.put(entry.getKey(), entry.getValue());
        }
        this.block = (this.fileSize % threadnum) == 0 ? this.fileSize / threadnum : this.fileSize / threadnum + 1;
        if (this.data.size() == threadnum) {
            for (int i = 0; i < threadnum; i++) {
                this.downloadSize += this.data.get(i);
            }
            Log.i(TAG, "已经下载的长度" + this.downloadSize);
        }
    } catch (Exception e) {
        e.printStackTrace();
        throw new RuntimeException("don't connection this url");
    }
}

/**
 * @param allocator An {@link Allocator} from which allocations for sample data can be obtained.
 */
public RollingSampleBuffer(Allocator allocator) {
  this.allocator = allocator;
  allocationLength = allocator.getIndividualAllocationLength();
  infoQueue = new InfoQueue();
  dataQueue = new LinkedBlockingDeque<>();
  extrasHolder = new SampleExtrasHolder();
  scratch = new ParsableByteArray(INITIAL_SCRATCH_SIZE);
  lastAllocationOffset = allocationLength;
}

/**
 * take retrieves elements in FIFO order
 */
public void testTakeFirst() throws InterruptedException {
    LinkedBlockingDeque q = populatedDeque(SIZE);
    for (int i = 0; i < SIZE; ++i) {
        assertEquals(i, q.takeFirst());
    }
}

public static void main ( final String[] args )
{
    final CountingThreadPoolExecutor exec = new CountingThreadPoolExecutor ( 1, 1, 1, TimeUnit.MINUTES, new LinkedBlockingDeque<Runnable> (), new NamedThreadFactory ( "Testing" ) );

    exec.addListener ( new Listener () {

        @Override
        public void countChanged ( final int count )
        {
            System.out.println ( "Count: " + count );
        }
    } );

    for ( int i = 0; i < 100; i++ )
    {
        exec.execute ( new Runnable () {

            @Override
            public void run ()
            {
                System.out.println ( "Test" );
            }
        } );
    }

    System.out.println ( "Before now" );
    exec.shutdownNow ();
    System.out.println ( "After now" );
}

/**
 * offerFirst(null) throws NullPointerException
 */
public void testOfferFirstNull() {
    LinkedBlockingDeque q = new LinkedBlockingDeque();
    try {
        q.offerFirst(null);
        shouldThrow();
    } catch (NullPointerException success) {}
}

private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException {
    stream.defaultReadObject();
    this.dom = javax.xml.parsers.DocumentBuilderFactory.newInstance();
    this.dom.setNamespaceAware(true);
    this.dom.setIgnoringComments(false);
    this.xform = javax.xml.transform.TransformerFactory.newInstance();
    int poolSize = Runtime.getRuntime().availableProcessors() * 2;
    this.documentBuilderPool = new LinkedBlockingDeque<DocumentBuilder>(poolSize);
}

/**
 * Spliterator characteristics are as advertised
 */
public void testSpliterator_characteristics() {
    LinkedBlockingDeque q = new LinkedBlockingDeque();
    Spliterator s = q.spliterator();
    int characteristics = s.characteristics();
    int required = Spliterator.CONCURRENT
        | Spliterator.NONNULL
        | Spliterator.ORDERED;
    assertEquals(required, characteristics & required);
    assertTrue(s.hasCharacteristics(required));
    assertEquals(0, characteristics
                 & (Spliterator.DISTINCT
                    | Spliterator.IMMUTABLE
                    | Spliterator.SORTED));
}

@Override
public Spliterator<E> trySplit() {
    Object h;
    LinkedBlockingDeque<E> q = queue;
    if (!exhausted &&
        ((h = current) != null || (h = getQueueFirst(q)) != null)
        && getNextNode(h) != null) {
        int n = batch = Math.min(batch + 1, MAX_BATCH);
        Object[] a = new Object[n];
        ReentrantLock lock = queueLock;
        int i = 0;
        Object p = current;
        lock.lock();
        try {
            if (p != null || (p = getQueueFirst(q)) != null)
                for (; p != null && i < n; p = succ(p))
                    if ((a[i] = getNodeItem(p)) != null)
                        i++;
        } finally {
            // checkInvariants();
            lock.unlock();
        }
        if ((current = p) == null) {
            est = 0L;
            exhausted = true;
        }
        else if ((est -= i) < 0L)
            est = 0L;
        if (i > 0)
            return Spliterators.spliterator
                (a, 0, i, (Spliterator.ORDERED |
                           Spliterator.NONNULL |
                           Spliterator.CONCURRENT));
    }
    return null;
}

/**
 * Creates a {@code LinkedBlockingDeque} with a capacity of {@link Integer#MAX_VALUE},
 * containing the elements of the specified iterable,
 * in the order they are returned by the iterable's iterator.
 *
 * @since 12.0
 */
@GwtIncompatible // LinkedBlockingDeque
public static <E> LinkedBlockingDeque<E> newLinkedBlockingDeque(Iterable<? extends E> elements) {
  if (elements instanceof Collection) {
    return new LinkedBlockingDeque<E>(Collections2.cast(elements));
  }
  LinkedBlockingDeque<E> deque = new LinkedBlockingDeque<E>();
  Iterables.addAll(deque, elements);
  return deque;
}

public Test testsForLinkedBlockingDeque() {
  return QueueTestSuiteBuilder.using(
          new TestStringQueueGenerator() {
            @Override
            public Queue<String> create(String[] elements) {
              return new LinkedBlockingDeque<String>(MinimalCollection.of(elements));
            }
          })
      .named("LinkedBlockingDeque")
      .withFeatures(
          CollectionFeature.GENERAL_PURPOSE, CollectionFeature.KNOWN_ORDER, CollectionSize.ANY)
      .suppressing(suppressForLinkedBlockingDeque())
      .createTestSuite();
}

/**
 * all elements successfully putLast are contained
 */
public void testPutLast() throws InterruptedException {
    LinkedBlockingDeque q = new LinkedBlockingDeque(SIZE);
    for (int i = 0; i < SIZE; ++i) {
        Integer x = new Integer(i);
        q.putLast(x);
        assertTrue(q.contains(x));
    }
    assertEquals(0, q.remainingCapacity());
}

/**
 * Initializing from null Collection throws NullPointerException
 */
public void testConstructor3() {
    try {
        new LinkedBlockingDeque(null);
        shouldThrow();
    } catch (NullPointerException success) {}
}

private static void realMain(String[] args) throws Throwable {
    Collection<Deque<Integer>> deqs = new ArrayDeque<>(3);
    deqs.add(new ArrayDeque<Integer>());
    deqs.add(new LinkedList<Integer>());
    deqs.add(new LinkedBlockingDeque<Integer>());
    deqs.add(new ConcurrentLinkedDeque<Integer>());

    equal(deqs);

    for (Tweaker tweaker : tweakers) {
        for (Deque<Integer> deq : deqs)
            tweaker.run(deq);
        equal(deqs);
    }
}

private static void realMain(final String[] args) throws Throwable {
    testQueue(new SynchronousQueue<Object>());
    testQueue(new ArrayBlockingQueue<Object>(1,false));
    testQueue(new ArrayBlockingQueue<Object>(1,true));
    testQueue(new LinkedBlockingQueue<Object>(1));
    testQueue(new LinkedBlockingDeque<Object>(1));
}