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

/**
 * Returns a new cached thread pool with the given thread count and
 * {@link UncaughtThrowableStrategy} to use when loading frames of animations.
 */
// Public API.
@SuppressWarnings("WeakerAccess")
public static GlideExecutor newAnimationExecutor(
    int threadCount, UncaughtThrowableStrategy uncaughtThrowableStrategy) {
   return new GlideExecutor(
      new ThreadPoolExecutor(
          0 /* corePoolSize */,
          threadCount,
          KEEP_ALIVE_TIME_MS,
          TimeUnit.MILLISECONDS,
          new PriorityBlockingQueue<Runnable>(),
          new DefaultThreadFactory(
              ANIMATION_EXECUTOR_NAME,
              uncaughtThrowableStrategy,
              true)));
}

public void testPriorityQueue() throws Exception {
    PriorityBlockingQueue<Priority> queue = new PriorityBlockingQueue<>();
    List<Priority> priorities = Arrays.asList(Priority.values());
    Collections.shuffle(priorities, random());

    for (Priority priority : priorities) {
        queue.add(priority);
    }

    Priority prevPriority = null;
    while (!queue.isEmpty()) {
        if (prevPriority == null) {
            prevPriority = queue.poll();
        } else {
            assertThat(queue.poll().after(prevPriority), is(true));
        }
    }
}

public void testGet_concurrent() {
  assertTrue(ArbitraryInstances.get(BlockingDeque.class).isEmpty());
  assertTrue(ArbitraryInstances.get(BlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(DelayQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(SynchronousQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(PriorityBlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentMap.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentNavigableMap.class).isEmpty());
  ArbitraryInstances.get(Executor.class).execute(ArbitraryInstances.get(Runnable.class));
  assertNotNull(ArbitraryInstances.get(ThreadFactory.class));
  assertFreshInstanceReturned(
      BlockingQueue.class, BlockingDeque.class, PriorityBlockingQueue.class,
      DelayQueue.class, SynchronousQueue.class,
      ConcurrentMap.class, ConcurrentNavigableMap.class,
      AtomicReference.class, AtomicBoolean.class,
      AtomicInteger.class, AtomicLong.class, AtomicDouble.class);
}

public XExecutor getExecutor() {
    if (executor == null) {
        synchronized (DownloadThreadPool.class) {
            if (executor == null) {
                executor = new XExecutor(corePoolSize,
                        MAX_POOL_SIZE,
                        KEEP_ALIVE_TIME,
                        UNIT,
                        new PriorityBlockingQueue<Runnable>()/*无限容量的缓冲队列*/,
                        Executors.defaultThreadFactory()/*线程创建工厂*/,
                        new ThreadPoolExecutor.AbortPolicy()/*继续超出上限的策略，阻止*/);
            }
        }
    }

    return executor;
}

private ThreadTask()
{
    final long keepAliveTime = 60L;
    taskCompare = new TaskCompare();
    dbThreadQueue = new PriorityBlockingQueue<PrioriTask>(dbThreadCount,
            taskCompare);
    netThreadQueue = new PriorityBlockingQueue<PrioriTask>(netThreadCount,
            taskCompare);
    otherThreadQueue = new PriorityBlockingQueue<PrioriTask>(dbThreadCount,
            taskCompare);
    dbThreadPool = new ThreadPoolExecutor(dbThreadCount, dbThreadCount, 0L,
            TimeUnit.MILLISECONDS, dbThreadQueue);
    netThreadPool = new ThreadPoolExecutor(netThreadCount, netThreadCount,
            0L, TimeUnit.MILLISECONDS, netThreadQueue);
    otherThreadPool = new ThreadPoolExecutor(otherThreadCount,
            Integer.MAX_VALUE, keepAliveTime, TimeUnit.SECONDS,
            otherThreadQueue);
}

public BlockchainManager(CrypDist crypDist, byte[] session_key)
{
    this.crypDist = crypDist;
    dbManager = new PostgresDB("blockchain", "postgres", "", false);
    serverAccessor = new ServerAccessor();
    transactionPendingBucket = new ConcurrentHashMap<>();
    transactionBucket = new PriorityBlockingQueue<>();
    transactionBucket_solid = new ArrayList<>(BLOCK_SIZE);
    buildBlockchain();
    hashes = new ConcurrentHashMap<>();
    numOfPairs = 0;
    serverTime = getServerTime();
    systemTime = System.currentTimeMillis();
    updating = false;
    Timer timer = new Timer();
    timer.schedule(new BlockchainBatch(),0, Config.BLOCKCHAIN_BATCH_PERIOD);
}

public double calculaCosteAtencionVictimasFinalesAsignadas(double factorMultiplicativo, VictimsToRescue victims2R, MisObjetivos misObjs){

        double tiempo = 0;     //Variable para calcular el tiempo

        PriorityBlockingQueue <Objetivo> colaobjetivos = misObjs.getMisObjetivosPriorizados();
        int tamaniocola = colaobjetivos.size();

        Iterator<Objetivo> it = colaobjetivos.iterator();

        if (tamaniocola==0){
            return 0;
        }

        while (it.hasNext()){
          //Hay al menos un objetivo
          Objetivo ob = it.next();
          String referenciaIdObjetivo = ob.getobjectReferenceId();

          //Obtener la victima de la cola
          Victim victimaActualCola = victims2R.getVictimToRescue(referenciaIdObjetivo);                       
          int prioridadVictimaActualCola = victimaActualCola.getPriority();

          tiempo = tiempo + (factorMultiplicativo*prioridadVictimaActualCola);
        }                       
        return tiempo;
    }

/**
 * Initial MiniDownloader.
 *
 * @param context
 */
public void init(Context context) {
    this.appContext = context.getApplicationContext();
    /** Create work executor. */
    this.workExecutor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors(), 0L, TimeUnit.MILLISECONDS, new PriorityBlockingQueue<Runnable>()) {
        @Override
        protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
            if (callable instanceof CustomFutureCallable) {
                return ((CustomFutureCallable) callable).newTaskFor();
            }
            return super.newTaskFor(callable);
        }
    };
    /** Create command executor. */
    this.commandExecutor = Executors.newSingleThreadExecutor();
    /** Create and initial task manager. */
    taskManager = new TaskManager();
    taskManager.init(context);
    /** Create and start ProgressUpdater. */
    progressUpdater = new ProgressUpdater();
    progressUpdater.start();
}

/**
 * 构建队列
 *
 * @param size
 *         队列大小
 * @param isPriority
 *         是否优先级队列
 * @return 队列
 */
public static BlockingQueue<Runnable> buildQueue(int size, boolean isPriority) {
    BlockingQueue<Runnable> queue;
    if (size == 0) { // 默认无队列
        queue = new SynchronousQueue<Runnable>();
    } else { // 有限队列或无限队列
        if (isPriority) {
            queue = size < 0 ? new PriorityBlockingQueue<Runnable>()
                    : new PriorityBlockingQueue<Runnable>(size);
        } else {
            queue = size < 0 ? new LinkedBlockingQueue<Runnable>()
                    : new LinkedBlockingQueue<Runnable>(size);
        }
    }
    return queue;
}

/**
 * Store sensor data so that it can be published in the next publishing cycle. Unlike
 * the other log methods, this method saves the {@link #BUFFER_SIZE_FOR_ONCHANGE_SENSORS} most
 * recent sensor readings per sensor type.
 * @param data
 */
public void logSensorDataOnChange(SensorData data) {
    PriorityBlockingQueue<SensorData> newQueue =
            new PriorityBlockingQueue<SensorData>(BUFFER_SIZE_FOR_ONCHANGE_SENSORS,
                    new Comparator<SensorData>() {
        @Override
        public int compare(SensorData o1, SensorData o2) {
            return Long.compare(o1.getTimestamp(), o2.getTimestamp());
        }
    });
    PriorityBlockingQueue<SensorData> lastData = mOnChangeData.putIfAbsent(
            data.getSensorName(), newQueue);

    if (lastData == null) {
        lastData = newQueue;
    }

    // remove old entries if necessary
    while (lastData.size() >= BUFFER_SIZE_FOR_ONCHANGE_SENSORS) {
       lastData.poll();
    }

    lastData.offer(data);
}

/**
 * retainAll(c) retains only those elements of c and reports true if changed
 */
public void testRetainAll() {
    PriorityBlockingQueue q = populatedQueue(SIZE);
    PriorityBlockingQueue p = populatedQueue(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();
    }
}

public static void main(String[] args) {
    final Comparator<String> firstChar = new Comparator<>() {
        public int compare(String x, String y) {
            return x.charAt(0) - y.charAt(0); }};

    test(new PriorityQueue<String>(firstChar));
    test(new PriorityQueue<String>(10, firstChar));
    test(new PriorityBlockingQueue<String>(10, firstChar));
    test(new ArrayBlockingQueue<String>(10));
    test(new LinkedBlockingQueue<String>(10));
    test(new LinkedBlockingDeque<String>(10));
    test(new LinkedTransferQueue<String>());
    test(new ArrayDeque<String>(10));

    System.out.printf("%nPassed = %d, failed = %d%n%n", passed, failed);
    if (failed > 0) throw new Error("Some tests failed");
}

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

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

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

    pool = Executors.newCachedThreadPool();
    for (int i = 1; i <= maxProducers; 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, 300);
        run(new LinkedBlockingQueue<Integer>(100), i, 700);
        run(new LinkedBlockingDeque<Integer>(100), i , 500);
        run(new LinkedTransferQueue<Integer>(), i, 1000);
        run(new PriorityBlockingQueue<Integer>(), i, 1000);
        run(new SynchronousQueue<Integer>(), i, 500);
        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;
}

/**
 * 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);
}

public void testGet_concurrent() {
  assertTrue(ArbitraryInstances.get(BlockingDeque.class).isEmpty());
  assertTrue(ArbitraryInstances.get(BlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(DelayQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(SynchronousQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(PriorityBlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentMap.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentNavigableMap.class).isEmpty());
  ArbitraryInstances.get(Executor.class).execute(ArbitraryInstances.get(Runnable.class));
  assertNotNull(ArbitraryInstances.get(ThreadFactory.class));
  assertFreshInstanceReturned(
      BlockingQueue.class, BlockingDeque.class, PriorityBlockingQueue.class,
      DelayQueue.class, SynchronousQueue.class,
      ConcurrentMap.class, ConcurrentNavigableMap.class,
      AtomicReference.class, AtomicBoolean.class,
      AtomicInteger.class, AtomicLong.class, AtomicDouble.class);
}

GlideExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTimeInMs, String name,
    UncaughtThrowableStrategy uncaughtThrowableStrategy, boolean preventNetworkOperations,
    boolean executeSynchronously) {
  this(
      corePoolSize,
      maximumPoolSize,
      keepAliveTimeInMs,
      name,
      uncaughtThrowableStrategy,
      preventNetworkOperations,
      executeSynchronously,
      new PriorityBlockingQueue<Runnable>());
}

/**
 * @param poolSize 工作线程数
 * @param fifo     优先级相同时, 等待队列的是否优先执行先加入的任务.
 */
public PriorityExecutor(int poolSize, boolean fifo) {
    BlockingQueue<Runnable> mPoolWorkQueue =
            new PriorityBlockingQueue<Runnable>(MAXIMUM_POOL_SIZE, fifo ? FIFO_CMP : FILO_CMP);
    mThreadPoolExecutor = new ThreadPoolExecutor(
            poolSize,
            MAXIMUM_POOL_SIZE,
            KEEP_ALIVE,
            TimeUnit.SECONDS,
            mPoolWorkQueue,
            sThreadFactory);
}

/**
 * Returns a new fixed thread pool with the given thread count, thread name prefix,
 * and {@link com.bumptech.glide.load.engine.executor.GlideExecutor.UncaughtThrowableStrategy}.
 *
 * <p>Disk cache executors do not allow network operations on their threads.
 *
 * @param threadCount The number of threads.
 * @param name The prefix for each thread name.
 * @param uncaughtThrowableStrategy The {@link
 * com.bumptech.glide.load.engine.executor.GlideExecutor.UncaughtThrowableStrategy} to use to
 *                                  handle uncaught exceptions.
 */
// Public API.
@SuppressWarnings("WeakerAccess")
public static GlideExecutor newDiskCacheExecutor(
    int threadCount, String name, UncaughtThrowableStrategy uncaughtThrowableStrategy) {
  return new GlideExecutor(
      new ThreadPoolExecutor(
          threadCount /* corePoolSize */,
          threadCount /* maximumPoolSize */,
          0 /* keepAliveTime */,
          TimeUnit.MILLISECONDS,
          new PriorityBlockingQueue<Runnable>(),
          new DefaultThreadFactory(name, uncaughtThrowableStrategy, true)));
}

private ExecutorService createExecutor(Config config) {
    ThreadPoolExecutor service = new ThreadPoolExecutor(config.core, config.max, config.timeout,
            TimeUnit.MILLISECONDS, new PriorityBlockingQueue<Runnable>(QUEUE_INIT_CAPACITY, mQueueComparator),
            new TaskThreadFactory(name), new ThreadPoolExecutor.DiscardPolicy());

    allowCoreThreadTimeOut(service, config.allowCoreTimeOut);

    return service;
}

public AiParameters(L2NpcInstance actor)
{
    _eventQueue = new PriorityBlockingQueue<AiEvent>();
    _hated = new ArrayList<>();
    _liked = new ArrayList<>();
    _actor = actor;
}

/**
 * Creates a {@code PriorityBlockingQueue} containing the given elements.
 *
 * <b>Note:</b> If the specified iterable is a {@code SortedSet} or a {@code PriorityQueue},
 * this priority queue will be ordered according to the same ordering.
 *
 * @since 11.0 (requires that {@code E} be {@code Comparable} since 15.0).
 */
public static <E extends Comparable> PriorityBlockingQueue<E> newPriorityBlockingQueue(
    Iterable<? extends E> elements) {
  if (elements instanceof Collection) {
    return new PriorityBlockingQueue<E>(Collections2.cast(elements));
  }
  PriorityBlockingQueue<E> queue = new PriorityBlockingQueue<E>();
  Iterables.addAll(queue, elements);
  return queue;
}

public SyncMessageEventPipeline(MessageStoreConfig storeConfig, MessageQueueHolder messageQueueHolder, int capacity) {
    this.storeConfig = storeConfig;
    this.messageQueueHolder = messageQueueHolder;
    pipeline = new PriorityBlockingQueue<>(capacity);
    handlers.add(new MessageQueueHandler(this.storeConfig.getPutMQRetryTime(), this.messageQueueHolder));
    thread = new Thread(() -> dispatch(), "message-event-pipeline");
    thread.setDaemon(true);
}

/**
 * Creates a {@code PriorityBlockingQueue} containing the given elements.
 *
 * <b>Note:</b> If the specified iterable is a {@code SortedSet} or a {@code PriorityQueue},
 * this priority queue will be ordered according to the same ordering.
 *
 * @since 11.0 (requires that {@code E} be {@code Comparable} since 15.0).
 */
@GwtIncompatible // PriorityBlockingQueue
public static <E extends Comparable> PriorityBlockingQueue<E> newPriorityBlockingQueue(
    Iterable<? extends E> elements) {
  if (elements instanceof Collection) {
    return new PriorityBlockingQueue<E>(Collections2.cast(elements));
  }
  PriorityBlockingQueue<E> queue = new PriorityBlockingQueue<E>();
  Iterables.addAll(queue, elements);
  return queue;
}

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

public static List<BlockingQueue<Object>> blockingQueues() {
  return ImmutableList.<BlockingQueue<Object>>of(
      new LinkedBlockingQueue<Object>(),
      new LinkedBlockingQueue<Object>(10),
      new SynchronousQueue<Object>(),
      new ArrayBlockingQueue<Object>(10),
      new LinkedBlockingDeque<Object>(),
      new LinkedBlockingDeque<Object>(10),
      new PriorityBlockingQueue<Object>(10, Ordering.arbitrary()));
}

private ThumbWorkManger(Context context) {
    this.mThumbPool = new ThreadPoolExecutor(5, Integer.MAX_VALUE,
            60L, TimeUnit.SECONDS,
            new PriorityBlockingQueue<Runnable>());
    this.context = context;
    packageManager = context.getPackageManager();
}

/**
 * iterator iterates through all elements
 */
public void testIterator() {
    PriorityBlockingQueue q = populatedQueue(SIZE);
    Iterator it = q.iterator();
    int i;
    for (i = 0; it.hasNext(); i++)
        assertTrue(q.contains(it.next()));
    assertEquals(i, SIZE);
    assertIteratorExhausted(it);
}

public XExecutor getExecutor() {
    if (executor == null) {
        synchronized (UploadThreadPool.class) {
            if (executor == null) {
                executor = new XExecutor(corePoolSize, MAX_IMUM_POOL_SIZE, KEEP_ALIVE_TIME, UNIT, //
                                         new PriorityBlockingQueue<Runnable>(),   //无限容量的缓冲队列
                                         Executors.defaultThreadFactory(),        //线程创建工厂
                                         new ThreadPoolExecutor.AbortPolicy());   //继续超出上限的策略，阻止
            }
        }
    }
    return executor;
}

public VolleyRequestQueue(Network network, Network fileNetwork, int threadPoolSize, int cacheThreadPoolSize, ResponseDelivery delivery) {
    this.mSequenceGenerator = new AtomicInteger();
    this.mCurrentRequests = new HashSet();
    this.mCacheQueue = new PriorityBlockingQueue();
    this.mNetworkQueue = new PriorityBlockingQueue();
    this.mFileNetworkQueue = new PriorityBlockingQueue();
    this.mDispatchers = new NetworkDispatcher[threadPoolSize];
    this.mCacheDispatchers = new CacheDispatcher[cacheThreadPoolSize];
    this.mDelivery = delivery;
}

/**
 * toString contains toStrings of elements
 */
public void testToString() {
    PriorityBlockingQueue q = populatedQueue(SIZE);
    String s = q.toString();
    for (int i = 0; i < SIZE; ++i) {
        assertTrue(s.contains(String.valueOf(i)));
    }
}

@Override
public void enqueueLog(Path log) {
  String logPrefix = DefaultWALProvider.getWALPrefixFromWALName(log.getName());
  PriorityBlockingQueue<Path> queue = queues.get(logPrefix);
  if (queue == null) {
    queue = new PriorityBlockingQueue<Path>(queueSizePerGroup, new LogsComparator());
    queues.put(logPrefix, queue);
    if (this.sourceRunning) {
      // new wal group observed after source startup, start a new worker thread to track it
      // notice: it's possible that log enqueued when this.running is set but worker thread
      // still not launched, so it's necessary to check workerThreads before start the worker
      final ReplicationSourceWorkerThread worker =
          new ReplicationSourceWorkerThread(logPrefix, queue, replicationQueueInfo, this);
      ReplicationSourceWorkerThread extant = workerThreads.putIfAbsent(logPrefix, worker);
      if (extant != null) {
        LOG.debug("Someone has beat us to start a worker thread for wal group " + logPrefix);
      } else {
        LOG.debug("Starting up worker for wal group " + logPrefix);
        worker.startup();
      }
    }
  }
  queue.put(log);
  int queueSize = logQueueSize.incrementAndGet();
  this.metrics.setSizeOfLogQueue(queueSize);
  // This will log a warning for each new log that gets created above the warn threshold
  if (queue.size() > this.logQueueWarnThreshold) {
    LOG.warn("WAL group " + logPrefix + " queue size: " + queueSize
        + " exceeds value of replication.source.log.queue.warn: " + logQueueWarnThreshold);
  }
}

/**
 * containsAll(c) is true when c contains a subset of elements
 */
public void testContainsAll() {
    PriorityBlockingQueue q = populatedQueue(SIZE);
    PriorityBlockingQueue p = new PriorityBlockingQueue(SIZE);
    for (int i = 0; i < SIZE; ++i) {
        assertTrue(q.containsAll(p));
        assertFalse(p.containsAll(q));
        p.add(new Integer(i));
    }
    assertTrue(p.containsAll(q));
}

@Test
public void testPriorityBlockingQueue() throws InterruptedException {
    PriorityBlockingQueue<Event> queue = new PriorityBlockingQueue<>(20, new PriorityEventQueueComparator());

    Event e1p1 = addEvent(queue, 1, 1);
    Event e2p1 = addEvent(queue, 2, 1);
    Event e3p1 = addEvent(queue, 3, 1);
    Event e4p1 = addEvent(queue, 4, 1);

    Event e5p2 = addEvent(queue, 5, 2);
    Event e6p2 = addEvent(queue, 6, 2);
    Event e7p2 = addEvent(queue, 7, 2);
    Event e8p2 = addEvent(queue, 8, 2);

    Event e21p0 = addEvent(queue, 21, 0);

    Event e31p5 = addEvent(queue, 31, 5);

    // First event should have the greatest priority and the lowest id
    Assert.assertEquals(e31p5, queue.take());

    Assert.assertEquals(e5p2, queue.take());
    Assert.assertEquals(e6p2, queue.take());
    Assert.assertEquals(e7p2, queue.take());
    Assert.assertEquals(e8p2, queue.take());

    Assert.assertEquals(e1p1, queue.take());
    Assert.assertEquals(e2p1, queue.take());
    Assert.assertEquals(e3p1, queue.take());
    Assert.assertEquals(e4p1, queue.take());

    Assert.assertEquals(e21p0, queue.take());
}

/**
 * Registers listener for a callback when specified event is executed.
 * @param listener
 * @return
 */
public AbstractEventListener addListener(AbstractEventListener listener)
{
    if ((listener == null))
    {
        throw new NullPointerException("Listener cannot be null!");
    }
    getListeners().computeIfAbsent(listener.getType(), k -> new PriorityBlockingQueue<>()).add(listener);
    return listener;
}

private ArrayList getIdsVictimasFinalesAsignadas(MisObjetivos misObjs, VictimsToRescue victims2R){
    ArrayList aux = new ArrayList();

    PriorityBlockingQueue <Objetivo> colaobjetivos = misObjs.getMisObjetivosPriorizados();
    int tamaniocola = colaobjetivos.size();

    Iterator<Objetivo> it = colaobjetivos.iterator();

    if (tamaniocola==0){
        return aux;
    }

    int index = 0;
    while (it.hasNext()){           
          //Hay al menos un objetivo
          Objetivo ob = it.next();
          String referenciaIdObjetivo = ob.getobjectReferenceId();

         //Obtener la victima de la cola
         Victim victimaActualCola = victims2R.getVictimToRescue(referenciaIdObjetivo);                
         String nameVictim = victimaActualCola.getName();

         aux.add(index, nameVictim);
         index ++;              
    }

    return aux;
}

public double CalculaTiempoAtencion(double factorMultiplicativo, Victim nuevaVictima, VictimsToRescue victims2R, MisObjetivos misObjs){

        double tiempo = 0;     //Variable para calcular el tiempo

        //Obtener la prioridad de la victima
        int prioridadNuevaVictima = nuevaVictima.getPriority();
        // si la victima no esta entre las vicitimas a rescatar o en los objetivos

        PriorityBlockingQueue <Objetivo> colaobjetivos = misObjs.getMisObjetivosPriorizados();

        Iterator<Objetivo> it = colaobjetivos.iterator();
         boolean hayVictimasArescatar = victims2R.getvictims2Rescue().isEmpty();

        while (it.hasNext()&&hayVictimasArescatar){
          //Hay al menos un objetivo
          Objetivo ob = it.next();
          String referenciaIdObjetivo = ob.getobjectReferenceId();
          //Obtener la victima de la cola
              if (referenciaIdObjetivo !=null){
                    Victim  victimaActualCola = victims2R.getVictimToRescue(referenciaIdObjetivo);
                    if(victimaActualCola !=null){
                        int prioridadVictimaActualCola = victimaActualCola.getPriority();
                        tiempo = tiempo + (factorMultiplicativo*prioridadVictimaActualCola);
                        }
              }
        }
        tiempo = tiempo + (factorMultiplicativo*prioridadNuevaVictima);
        return tiempo;
    }

public double CalculaTiempoAtencion(double factorMultiplicativo, Victim nuevaVictima, VictimsToRescue victims2R, MisObjetivos misObjs){

        double tiempo = 0;     //Variable para calcular el tiempo

        //Obtener la prioridad de la victima
        int prioridadNuevaVictima = nuevaVictima.getPriority();
        // si la victima no esta entre las vicitimas a rescatar o en los objetivos

        PriorityBlockingQueue <Objetivo> colaobjetivos = misObjs.getMisObjetivosPriorizados();

        Iterator<Objetivo> it = colaobjetivos.iterator();
         boolean hayVictimasArescatar = victims2R.getvictims2Rescue().isEmpty();

        while (it.hasNext()&&hayVictimasArescatar){
          //Hay al menos un objetivo
          Objetivo ob = it.next();
          String referenciaIdObjetivo = ob.getobjectReferenceId();
          //Obtener la victima de la cola
              if (referenciaIdObjetivo !=null){
                    Victim  victimaActualCola = victims2R.getVictimToRescue(referenciaIdObjetivo);
                    if(victimaActualCola !=null){
                        int prioridadVictimaActualCola = victimaActualCola.getPriority();
                        tiempo = tiempo + (factorMultiplicativo*prioridadVictimaActualCola);
                        }
              }
        }
        tiempo = tiempo + (factorMultiplicativo*prioridadNuevaVictima);
        return tiempo;
    }

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