java.util.concurrent.PriorityBlockingQueue#drainTo ( )源码实例Demo

/**
 * drainTo(c) empties queue into another collection c
 */
public void testDrainTo() {
    PriorityBlockingQueue q = populatedQueue(SIZE);
    ArrayList l = new ArrayList();
    q.drainTo(l);
    assertEquals(0, q.size());
    assertEquals(SIZE, l.size());
    for (int i = 0; i < SIZE; ++i)
        assertEquals(l.get(i), new Integer(i));
    q.add(zero);
    q.add(one);
    assertFalse(q.isEmpty());
    assertTrue(q.contains(zero));
    assertTrue(q.contains(one));
    l.clear();
    q.drainTo(l);
    assertEquals(0, q.size());
    assertEquals(2, l.size());
    for (int i = 0; i < 2; ++i)
        assertEquals(l.get(i), new Integer(i));
}
 

/**
 * drainTo empties queue
 */
public void testDrainToWithActivePut() throws InterruptedException {
    final PriorityBlockingQueue q = populatedQueue(SIZE);
    Thread t = new Thread(new CheckedRunnable() {
        public void realRun() {
            q.put(new Integer(SIZE + 1));
        }});

    t.start();
    ArrayList l = new ArrayList();
    q.drainTo(l);
    assertTrue(l.size() >= SIZE);
    for (int i = 0; i < SIZE; ++i)
        assertEquals(l.get(i), new Integer(i));
    t.join();
    assertTrue(q.size() + l.size() >= SIZE);
}
 

/**
 * drainTo(c, n) empties first min(n, size) elements of queue into c
 */
public void testDrainToN() {
    PriorityBlockingQueue q = new PriorityBlockingQueue(SIZE * 2);
    for (int i = 0; i < SIZE + 2; ++i) {
        for (int j = 0; j < SIZE; j++)
            assertTrue(q.offer(new Integer(j)));
        ArrayList l = new ArrayList();
        q.drainTo(l, i);
        int k = (i < SIZE) ? i : SIZE;
        assertEquals(k, l.size());
        assertEquals(SIZE - k, q.size());
        for (int j = 0; j < k; ++j)
            assertEquals(l.get(j), new Integer(j));
        do {} while (q.poll() != null);
    }
}
 

/**
 * drainTo(c) empties queue into another collection c
 */
public void testDrainTo() {
    PriorityBlockingQueue q = populatedQueue(SIZE);
    ArrayList l = new ArrayList();
    q.drainTo(l);
    assertEquals(0, q.size());
    assertEquals(SIZE, l.size());
    for (int i = 0; i < SIZE; ++i)
        assertEquals(l.get(i), new Integer(i));
    q.add(zero);
    q.add(one);
    assertFalse(q.isEmpty());
    assertTrue(q.contains(zero));
    assertTrue(q.contains(one));
    l.clear();
    q.drainTo(l);
    assertEquals(0, q.size());
    assertEquals(2, l.size());
    for (int i = 0; i < 2; ++i)
        assertEquals(l.get(i), new Integer(i));
}
 

/**
 * drainTo empties queue
 */
public void testDrainToWithActivePut() throws InterruptedException {
    final PriorityBlockingQueue q = populatedQueue(SIZE);
    Thread t = new Thread(new CheckedRunnable() {
        public void realRun() {
            q.put(new Integer(SIZE + 1));
        }});

    t.start();
    ArrayList l = new ArrayList();
    q.drainTo(l);
    assertTrue(l.size() >= SIZE);
    for (int i = 0; i < SIZE; ++i)
        assertEquals(l.get(i), new Integer(i));
    t.join();
    assertTrue(q.size() + l.size() >= SIZE);
}
 

/**
 * drainTo(c, n) empties first min(n, size) elements of queue into c
 */
public void testDrainToN() {
    PriorityBlockingQueue q = new PriorityBlockingQueue(SIZE * 2);
    for (int i = 0; i < SIZE + 2; ++i) {
        for (int j = 0; j < SIZE; j++)
            assertTrue(q.offer(new Integer(j)));
        ArrayList l = new ArrayList();
        q.drainTo(l, i);
        int k = (i < SIZE) ? i : SIZE;
        assertEquals(k, l.size());
        assertEquals(SIZE - k, q.size());
        for (int j = 0; j < k; ++j)
            assertEquals(l.get(j), new Integer(j));
        do {} while (q.poll() != null);
    }
}
 

@Test
public void givenUnorderedValues_whenPolling_thenShouldOrderQueue() throws InterruptedException {
    PriorityBlockingQueue<Integer> queue = new PriorityBlockingQueue<>();
    ArrayList<Integer> polledElements = new ArrayList<>();

    queue.add(1);
    queue.add(5);
    queue.add(2);
    queue.add(3);
    queue.add(4);

    queue.drainTo(polledElements);

    assertThat(polledElements).containsExactly(1, 2, 3, 4, 5);
}
 

/**
 * Queries with a deadline that lies significantly in the future can lie
 * around in the priority queue until that deadline is reached if there are
 * other queries in front of them that are not terminated and whose deadline
 * has not be reached. Therefore, periodically, we need to scan the queue
 * and clear out entries for terminated queries.
 */
static private void scanDeadlineQueue(final long nowNanos,
        final PriorityBlockingQueue<QueryDeadline> deadlineQueue) {

    final List<QueryDeadline> c = new ArrayList<QueryDeadline>(
            DEADLINE_QUEUE_SCAN_SIZE);

    // drain up to that many elements.
    deadlineQueue.drainTo(c, DEADLINE_QUEUE_SCAN_SIZE);

    int ndropped = 0, nrunning = 0;
    
    for (QueryDeadline x : c) {

        if (x.checkDeadline(nowNanos) != null) {

            // return this query to the deadline queue.
            deadlineQueue.add(x);

            nrunning++;

        } else {
            
            ndropped++;
            
        }

    }
    
    if (log.isInfoEnabled())
        log.info("Scan: threadhold=" + DEADLINE_QUEUE_SCAN_SIZE
                + ", ndropped=" + ndropped + ", nrunning=" + nrunning
                + ", deadlineQueueSize=" + deadlineQueue.size());

}