All about Greg Pata

Producer Consumer with delayed consumption

In this example, we have two producers (type 0 and 1) and one consumer. Again, the consumer waits for 50 ms and processes requests from the same type.
I thought it would be easier. The code does not end properly...

public class ProducerConsumerExample2 {
static ProducerConsumerExample2 INSTANCE;
public ProducerConsumerExample2(){
}
public static void main(String [] args){
start( new ProducerConsumerExample2());
}
public static void start( ProducerConsumerExample2 producerConsumerExample2 ) {
       INSTANCE = producerConsumerExample2;
       INSTANCE.run();  
   }
private void run() {
final BlockingQueue<Long> dropMs = new SynchronousQueue<Long> ();
final Request queue = new Request(dropMs);
(new Thread(new Consumer(queue,dropMs))).start();
       (new Thread(new Producer(queue,0))).start();
       try {
           Thread.sleep(9L);
       } catch (InterruptedException e) {}
       (new Thread(new Producer(queue,1))).start();
       
}
public class Producer implements Runnable {
   private Request queue;
   private int type;
   public Producer(Request queue, int type) {
       this.queue = queue;
       this.type = type;
   }
   public void run() {
       String importantInfo[] = {
           "Mares eat oats",
           "Does eat oats",
           "Little lambs eat ivy",
           "A kid will eat ivy too"
       };
       SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss.SSS");
       for (int i = 0;i < importantInfo.length;i++) {
           try {
            queue.add(importantInfo[i], type);
            System.out.format("MESSAGE OFFERED at %s: %d %s%n", sdf.format(Calendar.getInstance().getTime()), type, importantInfo[i]);
            Thread.sleep(1000 * (i == 1 ? 0 : 1) + 10 * (type == 1 ? 0 : 1));
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
       }
       queue.add("DONE",type);
   }    
}
public class Consumer implements Runnable {
       private Request queue;
       private BlockingQueue<Long> dropMs;
       public Consumer(Request queue, BlockingQueue<Long> dropMs) {
           this.queue = queue;
           this.dropMs = dropMs;
       }
       public void run() {
        SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss.SSS");
        try {
            for (Long ms = dropMs.take();true;ms = dropMs.take()) {
            String message = queue.read(ms);
            System.out.format("MESSAGE RECEIVED at %s: %s%n", sdf.format(Calendar.getInstance().getTime()), message);
            if (message.contains("DONE")) break;
           }
           } catch (InterruptedException e) {
  e.printStackTrace();
           }
       }
   }
public class Request {
private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
private ArrayList<String> requests;
private ArrayList<Integer> types;
private ArrayList<Boolean> oks;
private BlockingQueue<Long> dropMs;
private int start;
private boolean waiting;
private boolean nextWaiting;
private int oldType;
private int nextOldType;
public Request(BlockingQueue<Long> dropMs) {
this.dropMs = dropMs;
requests = new ArrayList<String>();
types = new ArrayList<Integer>();
oks = new ArrayList<Boolean>();
start = 0;
waiting = false;
nextWaiting = false;
oldType = -1;
nextOldType = -1;
}
public synchronized void add(String str, int type) {
requests.add(str);
types.add(type);
oks.add(false);
try {
dropMs.put(0L);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
public synchronized String read(Long ms) {
boolean once = true;
int startType = types.get(start);
String str = Integer.toString(startType);
int currentType = types.get(types.size()-1);
if (ms.equals(0L)) { // signal from producer, wait for 50ms
//System.out.format("%b %b oldType %d type %d%n", waiting, nextWaiting, oldType, currentType);
if (!waiting) {
waiting = true;
oldType = currentType;
// new future task to process
ScheduledFuture<?> countdown = scheduler.schedule(new Runnable() {
           @Override
           public void run() {
            try {
dropMs.put(1L);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
           }}, 50, TimeUnit.MILLISECONDS);
return "waiting";
} else {
if (currentType != oldType && !nextWaiting) {
nextWaiting = true;
nextOldType = currentType;
// next future task to process
ScheduledFuture<?> countdown = scheduler.schedule(new Runnable() {
           @Override
           public void run() {            
            try {
dropMs.put(1L);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
           }}, 50, TimeUnit.MILLISECONDS);
return "still waiting dif type";
} else {
return "still waiting";
}
}
} else { // signal from itself: time to process
waiting = false;
if (nextWaiting) {
nextWaiting = false;
waiting = true;
oldType = nextOldType;
}
// process request
for (int i = start;i < requests.size();i++) {
if (startType == types.get(i) && !oks.get(i)) {
str += requests.get(i);
oks.set(i, true);
//update next request start index
} else if(once && startType != types.get(i) && !oks.get(i)) {
once = false;
start = i;
}
}
if (once) start = requests.size();
return str;
}
}
}
}

Greg Pata

Recent posts by Greg Pata