thread safe queue --cruise

Posted 2021-01-14 apanda009

Implement a Queue with pop and push operations using concurrency coding.

　

 

package com.company;
import java.util.*;
import java.util.concurrent.locks.*;

class BQueue<T> {

     Condition isFullCondition;
     Condition isEmptyCondition;
     Lock lock;
int limit;
int cur = 0;
Queue<T> q = new LinkedList<>();

     public BQueue() {
         this(Integer.MAX_VALUE);
     }

     public BQueue(int limit) {
         this.limit = limit;
         lock = new ReentrantLock();
         isFullCondition = lock.newCondition();
         isEmptyCondition = lock.newCondition();
     }

     public void put (T t) {
         lock.lock();
         try {
             while (q.size() == limit) {
                 try {
                     isFullCondition.await();
                 } catch (InterruptedException ex) {}
             }
             q.add(t);
             isEmptyCondition.signalAll();
         } finally {
             lock.unlock();
         }
     }

     public T get() {
         T t = null;
         lock.lock();
         try {
             while (q.size() == 0) {
                 try {
                     isEmptyCondition.await();
                 } catch (InterruptedException ex) {}
             }
             t = q.poll();
             isFullCondition.signalAll();
         } finally {
             lock.unlock();
         }
         return t;
     }
}


class Producer extends Thread{

    private BQueue data;

    public Producer(BQueue data, String name) {
        this.data = data;
        this.setName(name);
    }

    @Override
    public void run() {
        for (int i=0; i<100 && data.cur < data.limit; i++) {
            data.put(data.cur);
            data.cur++;
            System.out.println(currentThread().getName() + " " + data.cur);
        }
    }
}


class Consumer extends Thread {

    private BQueue data;

    public Consumer(BQueue data, String name) {
        this.data = data;
        this.setName(name);
    }

    @Override
    public void run() {
        for (int i=0; i<data.limit; i++) {
            System.out.println(currentThread().getName()+ " " + i + "   " + data.get());
        }
    }
}




public class Test {

    public static void main (String args[]) {
        BQueue<Integer> data = new BQueue(10);
        new Producer(data, "put---1").start();
        new Producer(data, "put---2").start();
        new Consumer(data, "get---1").start();
        new Consumer(data, "get---2").start();

    }
}