cdmlib-commons/src/main/java/eu/etaxonomy/cdm/common/concurrent/ConcurrentQueue.java

   1 /**
   2 * Copyright (C) 2019 EDIT
   3 * European Distributed Institute of Taxonomy
   4 * http://www.e-taxonomy.eu
   5 *
   6 * The contents of this file are subject to the Mozilla Public License Version 1.1
   7 * See LICENSE.TXT at the top of this package for the full license terms.
   8 */
   9 package eu.etaxonomy.cdm.common.concurrent;
  10
  11 import java.util.LinkedList;
  12 import java.util.List;
  13
  14 import org.apache.logging.log4j.LogManager;
  15 import org.apache.logging.log4j.Logger;
  16
  17 /**
  18  * Concurrent queue. Example taken from
  19  * {@link
  20  * https://stackoverflow.com/questions/33195290/producer-consumer-model-using-javasynchronized-but-always-run-the-same-thread
  21  * } for Producer-Consumer pattern.
  22  *
  23  * @author a.mueller
  24  * @since 26.08.2019
  25  */
  26 public class ConcurrentQueue<T> {
  27
  28     private static final Logger logger = LogManager.getLogger();
  29
  30     private int capacity;
  31     private List<T> queue = new LinkedList<>();
  32
  33     public ConcurrentQueue(int capacity) {
  34         this.capacity = capacity;
  35     }
  36
  37     public synchronized void enqueue(T item) throws InterruptedException {
  38         while (queue.size() == this.capacity) {
  39             wait();
  40         }
  41
  42 //            System.out.println("Thread " + Thread.currentThread().getName() +
  43 //                               " producing " + item);
  44         queue.add(item);
  45         printSize();
  46
  47         if (queue.size() >= 1) {
  48             notifyAll();
  49         }
  50     }
  51
  52     public synchronized T dequeue() throws InterruptedException {
  53         T item;
  54
  55         while (queue.size() == 0) {
  56             wait();
  57         }
  58
  59         item = queue.remove(0);
  60         printSize();
  61 //            System.out.println("Thread " + Thread.currentThread().getName() +
  62 //                               " consuming " + item);
  63
  64         if (queue.size() == (capacity - 1)) {
  65             notifyAll();
  66         }
  67
  68         return item;
  69     }
  70
  71     /**
  72      *
  73      */
  74     private void printSize() {
  75         logger.warn("size:" + queue.size());
  76
  77     }
  78
  79 }