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

Collection<Queue<Boolean>> concurrentQueues() {
    List<Queue<Boolean>> queues = new ArrayList<Queue<Boolean>>();
    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;
}

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

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

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

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;
}

/**
 * Main method of the example. Creates three WriterTasks and a CleanerTask
 * 
 * @param args
 */
public static void main(String[] args) {

    // Creates the Event data structure
    Deque<Event> deque = new ConcurrentLinkedDeque<>();

    // Creates the three WriterTask and starts them
    WriterTask writer = new WriterTask(deque);
    for (int i = 0; i < Runtime.getRuntime().availableProcessors(); i++) {
        Thread thread = new Thread(writer);
        thread.start();
    }

    // Creates a cleaner task and starts them
    CleanerTask cleaner = new CleanerTask(deque);
    cleaner.start();

}

/**
 * remove(x) removes x and returns true if present
 */
public void testRemoveElement() {
    ConcurrentLinkedDeque q = populatedDeque(SIZE);
    for (int i = 1; i < SIZE; i += 2) {
        assertTrue(q.contains(i));
        assertTrue(q.remove(i));
        assertFalse(q.contains(i));
        assertTrue(q.contains(i - 1));
    }
    for (int i = 0; i < SIZE; i += 2) {
        assertTrue(q.contains(i));
        assertTrue(q.remove(i));
        assertFalse(q.contains(i));
        assertFalse(q.remove(i + 1));
        assertFalse(q.contains(i + 1));
    }
    assertTrue(q.isEmpty());
}

/**
 * retainAll(c) retains only those elements of c and reports true if change
 */
public void testRetainAll() {
    ConcurrentLinkedDeque q = populatedDeque(SIZE);
    ConcurrentLinkedDeque p = populatedDeque(SIZE);
    for (int i = 0; i < SIZE; ++i) {
        boolean changed = q.retainAll(p);
        if (i == 0)
            assertFalse(changed);
        else
            assertTrue(changed);

        assertTrue(q.containsAll(p));
        assertEquals(SIZE - i, q.size());
        p.remove();
    }
}

/**
 * iterator.remove() removes current element
 */
public void testIteratorRemove() {
    final ConcurrentLinkedDeque q = new ConcurrentLinkedDeque();
    final Random rng = new Random();
    for (int iters = 0; iters < 100; ++iters) {
        int max = rng.nextInt(5) + 2;
        int split = rng.nextInt(max - 1) + 1;
        for (int j = 1; j <= max; ++j)
            q.add(new Integer(j));
        Iterator it = q.iterator();
        for (int j = 1; j <= split; ++j)
            assertEquals(it.next(), new Integer(j));
        it.remove();
        assertEquals(it.next(), new Integer(split + 1));
        for (int j = 1; j <= split; ++j)
            q.remove(new Integer(j));
        it = q.iterator();
        for (int j = split + 1; j <= max; ++j) {
            assertEquals(it.next(), new Integer(j));
            it.remove();
        }
        assertFalse(it.hasNext());
        assertTrue(q.isEmpty());
    }
}

/**
 * Descending iterator iterates through all elements
 */
public void testDescendingIterator() {
    ConcurrentLinkedDeque q = populatedDeque(SIZE);
    int i = 0;
    Iterator it = q.descendingIterator();
    while (it.hasNext()) {
        assertTrue(q.contains(it.next()));
        ++i;
    }
    assertEquals(i, SIZE);
    assertFalse(it.hasNext());
    try {
        it.next();
        shouldThrow();
    } catch (NoSuchElementException success) {}
}

/**
 * Descending iterator ordering is reverse FIFO
 */
public void testDescendingIteratorOrdering() {
    final ConcurrentLinkedDeque q = new ConcurrentLinkedDeque();
    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();
    }
}

/**
 * descendingIterator.remove() removes current element
 */
public void testDescendingIteratorRemove() {
    final ConcurrentLinkedDeque q = new ConcurrentLinkedDeque();
    final Random rng = new Random();
    for (int iters = 0; iters < 100; ++iters) {
        int max = rng.nextInt(5) + 2;
        int split = rng.nextInt(max - 1) + 1;
        for (int j = max; j >= 1; --j)
            q.add(new Integer(j));
        Iterator it = q.descendingIterator();
        for (int j = 1; j <= split; ++j)
            assertEquals(it.next(), new Integer(j));
        it.remove();
        assertEquals(it.next(), new Integer(split + 1));
        for (int j = 1; j <= split; ++j)
            q.remove(new Integer(j));
        it = q.descendingIterator();
        for (int j = split + 1; j <= max; ++j) {
            assertEquals(it.next(), new Integer(j));
            it.remove();
        }
        assertFalse(it.hasNext());
        assertTrue(q.isEmpty());
    }
}

public boolean initialize(Class<?> cls, ConcurrentLinkedDeque<String> alerts) {
    mThreat_alerts = alerts;
    if (cls == ConnectorMThreatModuleRest.class) {
        module = new ConnectorMThreatModuleRest();

        for (int i = 0; i < THREADS_NUMBER; i++) {
            executor.execute(this);
        }
        isRunning = true;
        return true;
    }
    return false;
}

/**
 * Instantiates a new Metrics.
 *
 * @param period           the period
 * @param maxSegmentNumber the max segment number
 */
public Metrics(Long period, Integer maxSegmentNumber) {
    this.period = period;
    this.maxSegmentNumber = maxSegmentNumber;
    this.metricsDeque = new ConcurrentLinkedDeque<>();
    SegmentMetrics newMetrics = new SegmentMetrics();
    this.metricsDeque.addFirst(newMetrics);
}

@Override
@SuppressWarnings("unchecked")
public synchronized void init() {
    // Byte: [-128, 127]
    final int length = 256;
    this.guiElements = new ConcurrentLinkedDeque<>();

    this.gameElements = (Queue<GameElement>[]) new Queue<?>[length];
    for (int i = 0; i < this.gameElements.length; i++) {
        this.gameElements[i] = new ArrayDeque<>();
    }

    this.gameElementAddQueue = new ArrayDeque<>();
    this.gameElementRemoveQueue = new ArrayDeque<>();

    // never start the game in paused state
    this.setPause(false);
}

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

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

public DefaultAccessLogReceiver(final Executor logWriteExecutor, final File outputDirectory, final String logBaseName, final String logNameSuffix) {
    this.logWriteExecutor = logWriteExecutor;
    this.outputDirectory = outputDirectory;
    this.logBaseName = logBaseName;
    this.logNameSuffix = (logNameSuffix != null) ? logNameSuffix : DEFAULT_LOG_SUFFIX;
    this.pendingMessages = new ConcurrentLinkedDeque<>();
    this.defaultLogFile = new File(outputDirectory, logBaseName + this.logNameSuffix);
    calculateChangeOverPoint();
}

public JDBCLogHandler(final HttpHandler next, final Executor logWriteExecutor, final String formatString, DataSource dataSource) {
    this.next = next;
    this.formatString = formatString;
    this.dataSource = dataSource;

    tableName = "access";
    remoteHostField = "remoteHost";
    userField = "userName";
    timestampField = "timestamp";
    virtualHostField = "virtualHost";
    methodField = "method";
    queryField = "query";
    statusField = "status";
    bytesField = "bytes";
    refererField = "referer";
    userAgentField = "userAgent";
    this.logWriteExecutor = logWriteExecutor;
    this.pendingMessages = new ConcurrentLinkedDeque<>();
}

public ProcessManager(final ScheduledExecutorService scheduler, SignalTransceiver com) {

        this.scheduler = scheduler;
        this.com = com;
        this.processEndListeners = new ConcurrentHashMap<>();
        this.processQueue = new ConcurrentLinkedDeque<>();
        this.shutdownHook = new Thread(this::destroyProcesses);
        this.observers = new ConcurrentHashMap<>();
        this.monitorEventHandlers = new ConcurrentHashMap<>();

    }

public static void main(String[] args) {

        // 用来存储事件
        Deque<Event> deque = new ConcurrentLinkedDeque<>();

        // 创建可用 cpu 个 WriterTask 线程
        WriterTask writer = new WriterTask(deque);
        for (int i = 0; i < Runtime.getRuntime().availableProcessors(); i++) {
            Thread thread = new Thread(writer);
            thread.start();
        }

        // 创建一个 CleanerTask 守护进程
        CleanerTask cleaner = new CleanerTask(deque);
        cleaner.start();

    }

/**
 * Constructs a new instance of the {@link AsyncBarrier} class.
 *
 * @param participants The number of participants.
 */
public AsyncBarrier(int participants) {
    Requires.range(participants > 0, "participants");

    this.participantCount = participants;
    this.remainingParticipants.set(participants);
    this.waiters = new ConcurrentLinkedDeque<>();
}

/**
 * Signals that a participant has completed work, and returns a future that completes when all other participants
 * have also completed work.
 *
 * @return A future.
 */
public final CompletableFuture<Void> signalAndWait() {
    CompletableFuture<Void> future = new CompletableFuture<>();
    this.waiters.push(future);
    if (remainingParticipants.decrementAndGet() == 0) {
        remainingParticipants.set(participantCount);
        ConcurrentLinkedDeque<CompletableFuture<Void>> localWaiters = this.waiters;
        this.waiters = new ConcurrentLinkedDeque<>();

        for (CompletableFuture<Void> waiter : localWaiters) {
            ForkJoinPool.commonPool().execute(ExecutionContext.wrap(() -> {
                waiter.complete(null);
            }));
        }
    }

    return future;
}

/**
 * @param layerUrl Does not include "/query" appended to end of url to layer.
 */
public ScrapeJob(String layerUrl) {
  executor = Executors.newWorkStealingPool();
  current = new AtomicInteger();
  total = new AtomicInteger();
  failed = new AtomicBoolean( false);
  this.layerUrl = layerUrl ;
  this.queryUrlStr = layerUrl + "/query";
  this.layerName = getLayerName();
  this.outputFileBase =  OUTPUT_FOLDER + "/" + layerName;
  this.outputZip =  OUTPUT_FOLDER + "/" + layerName + ".zip";
  this.deleteQueue = new ConcurrentLinkedDeque<>();
}

/**
 * Main method of the example. Creates three WriterTasks and a CleanerTask
 * 
 * @param args
 */
public static void main(String[] args) {

    // Creates the Event data structure
    Deque<Event> deque = new ConcurrentLinkedDeque<>();

    // Creates the three WriterTask and starts them
    WriterTask writer = new WriterTask(deque);
    for (int i = 0; i < Runtime.getRuntime().availableProcessors(); i++) {
        Thread thread = new Thread(writer);
        thread.start();
    }

    // Creates a cleaner task and starts them
    CleanerTask cleaner = new CleanerTask(deque);
    cleaner.start();

}

public static void main(String[] args) {

        ConcurrentHashMap<String, ConcurrentLinkedDeque<Operation>> userHash = new ConcurrentHashMap<>();
        HashFiller hashFiller = new HashFiller(userHash);

        Thread[] threads = new Thread[10];
        for (int i = 0; i < 10; i++) {
            threads[i] = new Thread(hashFiller);
            threads[i].start();
        }

        for (int i = 0; i < 10; i++) {
            try {
                threads[i].join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        System.out.printf("Size: %d\n", userHash.size());

        userHash.forEach(10, (user, list) -> {
            System.out.printf("%s: %s: %d\n", Thread.currentThread().getName(), user, list.size());
        });

        userHash.forEachEntry(10, entry -> {
            System.out.printf("%s: %s: %d\n", Thread.currentThread().getName(), entry.getKey(),
                    entry.getValue().size());
        });

        Operation op = userHash.search(10, (user, list) -> {
            for (Operation operation : list) {
                if (operation.getOperation().endsWith("1")) {
                    return operation;
                }
            }
            return null;
        });

        System.out.printf("The operation we have found is: %s, %s, %s,\n", op.getUser(), op.getOperation(),
                op.getTime());

        ConcurrentLinkedDeque<Operation> operations = userHash.search(10, (user, list) -> {
            if (list.size() > 10) {
                return list;
            }
            return null;
        });

        System.out.printf("The user we have found is: %s: %d operations\n", operations.getFirst().getUser(),
                operations.size());

        int totalSize = userHash.reduce(10, (user, list) -> {
            return list.size();
        }, (n1, n2) -> {
            return n1 + n2;
        });

        System.out.printf("The total size is: %d\n", totalSize);

    }

@Override
protected void compute() {

    if (persons.size() < 1000) {

        for (Person person: persons) {
            ConcurrentLinkedDeque<Person> personList=personMap.computeIfAbsent(person.getFirstName(), name -> {
                return new ConcurrentLinkedDeque<>();
            });
            personList.add(person);
        }
        return;
    }

    PersonMapTask child1, child2;

    child1 = new PersonMapTask(persons.subList(0, persons.size() / 2), personMap);
    child2 = new PersonMapTask(persons.subList(persons.size() / 2, persons.size()), personMap);

    invokeAll(child1, child2);

}

private void transitiveDeps(int depth) throws IOException {
    Stream<Location> deps = archives.stream()
                                    .flatMap(Archive::getDependencies);

    Deque<Location> unresolved = deps.collect(Collectors.toCollection(LinkedList::new));
    ConcurrentLinkedDeque<Location> deque = new ConcurrentLinkedDeque<>();
    do {
        Location target;
        while ((target = unresolved.poll()) != null) {
            if (finder.isParsed(target))
                continue;

            Archive archive = configuration.findClass(target).orElse(null);
            if (archive != null) {
                archives.add(archive);

                String name = target.getName();
                Set<Location> targets = apiOnly
                        ? finder.parseExportedAPIs(archive, name)
                        : finder.parse(archive, name);

                // build unresolved dependencies
                targets.stream()
                       .filter(t -> !finder.isParsed(t))
                       .forEach(deque::add);
            }
        }
        unresolved = deque;
        deque = new ConcurrentLinkedDeque<>();
    } while (!unresolved.isEmpty() && depth-- > 0);
}

DependencyFinder(JdepsConfiguration configuration,
                 JdepsFilter filter) {
    this.configuration = configuration;
    this.filter = filter;
    this.parsedArchives.put(API_FINDER, new ConcurrentLinkedDeque<>());
    this.parsedArchives.put(CLASS_FINDER, new ConcurrentLinkedDeque<>());
}

void test(String[] args) throws Throwable {
    test(new LinkedBlockingQueue());
    test(new LinkedBlockingQueue(20));
    test(new LinkedBlockingDeque());
    test(new LinkedBlockingDeque(20));
    test(new ConcurrentLinkedDeque());
    test(new ConcurrentLinkedQueue());
    test(new LinkedTransferQueue());
    test(new ArrayBlockingQueue(20));
}

ScalableBloomFilter(double ratio, double fpp, double pratio, long hint, boolean useOffHeapMemory) throws IOException {
  this.ratio = ratio;
  this.fpp = fpp;
  this.pratio = pratio;
  this.hint = hint;
  this.useOffHeapMemory = useOffHeapMemory;
  this.filters = new ConcurrentLinkedDeque<>(); // must be concurrent to safe publishing inside synchronized
  this.filters.addFirst(newFilter());
}

public static Test suite() {
    class Implementation implements CollectionImplementation {
        public Class<?> klazz() { return ConcurrentLinkedDeque.class; }
        public Collection emptyCollection() { return new ConcurrentLinkedDeque(); }
        public Object makeElement(int i) { return i; }
        public boolean isConcurrent() { return true; }
        public boolean permitsNulls() { return false; }
    }
    return newTestSuite(ConcurrentLinkedDequeTest.class,
                        CollectionTest.testSuite(new Implementation()));
}

/**
 * Returns a new deque of given size containing consecutive
 * Integers 0 ... n - 1.
 */
private static ConcurrentLinkedDeque<Integer> populatedDeque(int n) {
    ConcurrentLinkedDeque<Integer> q = new ConcurrentLinkedDeque<>();
    assertTrue(q.isEmpty());
    for (int i = 0; i < n; ++i)
        assertTrue(q.offer(new Integer(i)));
    assertFalse(q.isEmpty());
    assertEquals(n, q.size());
    assertEquals((Integer) 0, q.peekFirst());
    assertEquals((Integer) (n - 1), q.peekLast());
    return q;
}

/**
 * Initializing from null Collection throws NPE
 */
public void testConstructor3() {
    try {
        new ConcurrentLinkedDeque((Collection)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);
    }
}

/**
 * Initializing from Collection with some null elements throws NPE
 */
public void testConstructor5() {
    Integer[] ints = new Integer[SIZE];
    for (int i = 0; i < SIZE - 1; ++i)
        ints[i] = new Integer(i);
    try {
        new ConcurrentLinkedDeque(Arrays.asList(ints));
        shouldThrow();
    } catch (NullPointerException success) {}
}

/**
 * Deque contains all elements of collection used to initialize
 */
public void testConstructor6() {
    Integer[] ints = new Integer[SIZE];
    for (int i = 0; i < SIZE; ++i)
        ints[i] = new Integer(i);
    ConcurrentLinkedDeque q = new ConcurrentLinkedDeque(Arrays.asList(ints));
    for (int i = 0; i < SIZE; ++i)
        assertEquals(ints[i], q.poll());
}

/**
 * isEmpty is true before add, false after
 */
public void testEmpty() {
    ConcurrentLinkedDeque q = new ConcurrentLinkedDeque();
    assertTrue(q.isEmpty());
    q.add(one);
    assertFalse(q.isEmpty());
    q.add(two);
    q.remove();
    q.remove();
    assertTrue(q.isEmpty());
}

/**
 * size() changes when elements added and removed
 */
public void testSize() {
    ConcurrentLinkedDeque q = populatedDeque(SIZE);
    for (int i = 0; i < SIZE; ++i) {
        assertEquals(SIZE - i, q.size());
        q.remove();
    }
    for (int i = 0; i < SIZE; ++i) {
        assertEquals(i, q.size());
        q.add(new Integer(i));
    }
}

/**
 * peekFirst() returns element inserted with push
 */
public void testPush() {
    ConcurrentLinkedDeque q = populatedDeque(3);
    q.pollLast();
    q.push(four);
    assertSame(four, q.peekFirst());
}

/**
 * pop() removes first element, or throws NSEE if empty
 */
public void testPop() {
    ConcurrentLinkedDeque q = populatedDeque(SIZE);
    for (int i = 0; i < SIZE; ++i) {
        assertEquals(i, q.pop());
    }
    try {
        q.pop();
        shouldThrow();
    } catch (NoSuchElementException success) {}
}

/**
 * offer(null) throws NPE
 */
public void testOfferNull() {
    ConcurrentLinkedDeque q = new ConcurrentLinkedDeque();
    try {
        q.offer(null);
        shouldThrow();
    } catch (NullPointerException success) {}
}