java.util.concurrent.ArrayBlockingQueue#addAll ( )源码实例Demo

/**
 * addAll(this) throws IAE
 */
public void testAddAllSelf() {
    ArrayBlockingQueue q = populatedQueue(SIZE);
    try {
        q.addAll(q);
        shouldThrow();
    } catch (IllegalArgumentException success) {}
}
 

/**
 * addAll of a collection with any null elements throws NPE after
 * possibly adding some elements
 */
public void testAddAll3() {
    ArrayBlockingQueue q = new ArrayBlockingQueue(SIZE);
    Integer[] ints = new Integer[SIZE];
    for (int i = 0; i < SIZE - 1; ++i)
        ints[i] = new Integer(i);
    try {
        q.addAll(Arrays.asList(ints));
        shouldThrow();
    } catch (NullPointerException success) {}
}
 

/**
 * addAll throws ISE if not enough room
 */
public void testAddAll_insufficientSpace() {
    int size = ThreadLocalRandom.current().nextInt(1, SIZE);
    ArrayBlockingQueue q = populatedQueue(0, size, size, false);
    // Just fits:
    q.addAll(populatedQueue(size, size, 2 * size, false));
    assertEquals(0, q.remainingCapacity());
    assertEquals(size, q.size());
    assertEquals(0, q.peek());
    try {
        q = populatedQueue(0, size, size, false);
        q.addAll(Collections.nCopies(size + 1, 42));
        shouldThrow();
    } catch (IllegalStateException success) {}
}
 

/**
 * addAll(this) throws IAE
 */
public void testAddAllSelf() {
    ArrayBlockingQueue q = populatedQueue(SIZE);
    try {
        q.addAll(q);
        shouldThrow();
    } catch (IllegalArgumentException success) {}
}
 

/**
 * addAll of a collection with any null elements throws NPE after
 * possibly adding some elements
 */
public void testAddAll3() {
    ArrayBlockingQueue q = new ArrayBlockingQueue(SIZE);
    Integer[] ints = new Integer[SIZE];
    for (int i = 0; i < SIZE - 1; ++i)
        ints[i] = new Integer(i);
    try {
        q.addAll(Arrays.asList(ints));
        shouldThrow();
    } catch (NullPointerException success) {}
}
 

/**
 * addAll throws ISE if not enough room
 */
public void testAddAll4() {
    ArrayBlockingQueue q = new ArrayBlockingQueue(1);
    Integer[] ints = new Integer[SIZE];
    for (int i = 0; i < SIZE; ++i)
        ints[i] = new Integer(i);
    try {
        q.addAll(Arrays.asList(ints));
        shouldThrow();
    } catch (IllegalStateException success) {}
}
 

void runExperiment(Experiment experiment, List<Scorer> scorers,
                   int maxRows, ExperimentDB experimentDB, JudgmentList judgmentList,
                   String judgmentListId, boolean logResults)
        throws SQLException, IOException, SearchClientException {
    if (experimentDB.hasScores(experiment.getName())) {
        LOG.info("Already has scores for " + experiment.getName() + "; skipping.  " +
                "Use the -freshStart commandline option to clear all scores");
        return;
    }
    experimentDB.initScoreTable(scorers);
    SearchClient searchClient = SearchClientFactory.getClient(experiment.getSearchServerUrl());

    if (StringUtils.isBlank(experimentConfig.getIdField())) {
        LOG.info("default document 'idField' not set in experiment config. " +
                "Will use default: '"
                + searchClient.getDefaultIdField() + "'");
        experimentConfig.setIdField(searchClient.getDefaultIdField());
    }

    JudgmentList validated = searchServerValidatedMap.get(
            experiment.getSearchServerUrl() +
                    "_" + judgmentListId);
    if (validated == null) {
        validated = validate(searchClient, judgmentList);
        searchServerValidatedMap.put(experiment.getSearchServerUrl()
                + "_" + judgmentListId, validated);
    }
    ExecutorService executorService = Executors.newFixedThreadPool(
            experimentConfig.getNumThreads());
    ExecutorCompletionService<Integer> executorCompletionService =
            new ExecutorCompletionService<>(executorService);
    ArrayBlockingQueue<Judgments> queue = new ArrayBlockingQueue<>(
            validated.getJudgmentsList().size() +
                    experimentConfig.getNumThreads());

    queue.addAll(validated.getJudgmentsList());
    for (int i = 0; i < experimentConfig.getNumThreads(); i++) {
        queue.add(POISON);
    }

    for (int i = 0; i < experimentConfig.getNumThreads(); i++) {
        executorCompletionService.submit(
                new QueryRunner(experimentConfig.getIdField(), maxRows,
                        queue, experiment, experimentDB, scorers));
    }

    int completed = 0;
    while (completed < experimentConfig.getNumThreads()) {
        try {
            Future<Integer> future = executorCompletionService.take();
            future.get();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            completed++;
        }
    }
    executorService.shutdown();
    executorService.shutdownNow();
    //insertScores(experimentDB, experimentName, scoreAggregators);
    experimentDB.insertScoresAggregated(experiment.getName(), scorers);
    if (logResults) {
        logResults(experiment.getName(), scorers);
    }
}
 

void run() throws Throwable {
//         System.out.printf(
//             "iterations=%d size=%d, warmup=%1g, filter=\"%s\"%n",
//             iterations, size, warmupSeconds, filter);

        final ArrayList<Integer> al = new ArrayList<>(size);

        // Populate collections with random data
        final ThreadLocalRandom rnd = ThreadLocalRandom.current();
        for (int i = 0; i < size; i++)
            al.add(rnd.nextInt(size));

        final ArrayDeque<Integer> ad = new ArrayDeque<>(al);
        final ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(al.size());
        abq.addAll(al);

        // shuffle circular array elements so they wrap
        for (int i = 0, n = rnd.nextInt(size); i < n; i++) {
            ad.addLast(ad.removeFirst());
            abq.add(abq.remove());
        }

        ArrayList<Job> jobs = new ArrayList<>(Arrays.asList());

        List.of(al, ad, abq,
                new LinkedList<>(al),
                new PriorityQueue<>(al),
                new Vector<>(al),
                new ConcurrentLinkedQueue<>(al),
                new ConcurrentLinkedDeque<>(al),
                new LinkedBlockingQueue<>(al),
                new LinkedBlockingDeque<>(al),
                new LinkedTransferQueue<>(al),
                new PriorityBlockingQueue<>(al))
            .stream()
            .forEach(x -> {
                         jobs.addAll(collectionJobs(x));
                         if (x instanceof Deque)
                             jobs.addAll(dequeJobs((Deque<Integer>)x));
                     });

        if (reverse) Collections.reverse(jobs);
        if (shuffle) Collections.shuffle(jobs);

        time(filter(filter, jobs));
    }