线程在 notify() 后未恢复执行

Question

我有两个类(Customer和Till(。客户线程等待直到收到till线程的通知。在我的程序中，客户线程在收到till线程的通知后不会执行它的代码。till线程继续执行。

Customer.java(客户线程扩展线程(

import java.util.concurrent.*;
import java.util.*;
public class Customer extends Thread
{
Random random_generator = new Random();
public int minimumQueueLength;
public Set set;
public Iterator iterator;
public boolean placed_in_queue;
public List<Integer> queue_length_list;
public CopyOnWriteArrayList till_set = new CopyOnWriteArrayList();
public Till till, till_to_join;
public final Object lock;
public Customer(CopyOnWriteArrayList till_set)
{
this.till_set = till_set;
this.placed_in_queue = false;
queue_length_list = new ArrayList<Integer>();
lock = new Object();
}
public void run()
{   
try 
{
place_in_queue();
} 
catch (InterruptedException e1)
{
// TODO Auto-generated catch block
e1.printStackTrace();
}
if(placed_in_queue)
{   
synchronized(this.lock)
{
System.out.println(this.getName()+" waiting");
try {
this.lock.wait();
System.out.println(this.getName()+" has been woken");
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
}
}
else
{
}
}
public void place_in_queue() throws InterruptedException
{
placed_in_queue = false;
iterator = till_set.iterator();
while(iterator.hasNext())
{
till = (Till)iterator.next();
queue_length_list.add(till.customer_queue.size());
} 
minimumQueueLength = 
queue_length_list.indexOf(Collections.min(queue_length_list));
if(minimumQueueLength < 5)
{
try 
{
till_to_join = (Till)till_set.get(minimumQueueLength);
till_to_join.customer_queue.put(this);
placed_in_queue = true;
} 
catch (InterruptedException e) 
{
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}

Till.java(直到线程扩展线程(

import java.util.HashMap;
import java.util.Iterator;
import java.util.Random;
import java.util.concurrent.*;
public class Till extends Thread
{
BlockingQueue<String> item_queue = new ArrayBlockingQueue<String>(200);
BlockingQueue<Customer> customer_queue = new ArrayBlockingQueue<Customer>(10);
public Random random;
public Customer c;
public Till(BlockingQueue<String> item_queue) throws InterruptedException
{
this.item_queue = item_queue;
random = new Random();
}
public void run()
{                   
while(true)
{   
try 
{
c = customer_queue.take();
synchronized(c.lock)
{
System.out.println(this.getName()+" Waking up : "+c.getName());
c.lock.notify();
System.out.println(c.getName()+" has been notified!");
}           
} 
catch (InterruptedException e) 
{
e.printStackTrace();
}
}
}
}

CustomerGenerator.java

import java.util.*;
import java.util.concurrent.*;
public class CustomerGenerator extends Thread
{
public int customer_generation_rate;
//0 - slow
//1 - fast
public Random random_generator;
public static BlockingQueue<String> item_queue = new ArrayBlockingQueue<String>(200);
public static CopyOnWriteArrayList till_set = new CopyOnWriteArrayList();
public int i;
public CustomerGenerator(int customer_generation_rate, CopyOnWriteArrayList till_set)
{
this.customer_generation_rate = customer_generation_rate;
this.till_set = till_set;
this.i = 0;
random_generator = new Random();    
}
public void run()
{
while(i<1)
{
switch(customer_generation_rate)
{         
case 0 : try 
{
Thread.sleep(random_generator.nextInt(1000));
} 
catch (InterruptedException e) 
{
// TODO Auto-generated catch block
e.printStackTrace();
}
break;
case 1 : try
{
Thread.sleep(random_generator.nextInt(500));
}
catch(InterruptedException e)
{
// TODO Auto-generated catch block
e.printStackTrace();
}
break;
default : customer_generation_rate = 0;
break;
}
Customer customer = new Customer(till_set);
customer.start();
total_customer_count++;
i++;
}
}
} 

Driver.java

import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Driver
{
public static BlockingQueue<String> item_queue = new ArrayBlockingQueue<>(200);
public static CopyOnWriteArrayList<Till> till_set = new CopyOnWriteArrayList<Till>();
public static Set set;
public static Iterator iterator;
public static int i;
public static final int till_count = 5;
public static Thread till_thread;
public static Till till_object;
public static ExecutorService till_service = Executors.newFixedThreadPool(5);
public static void main(final String[] args) throws InterruptedException
{
for(i=0; i<till_count; i++)
{
till_object = new Till(item_queue);
till_set.add(till_object);
}
final CustomerGenerator customer_generator = new CustomerGenerator(0, till_set);
customer_generator.start();
Thread.sleep(5000);
for(final Till t : till_set)
{
till_service.submit(t);
}
}
}

获得的输出：

线程7等待
线程1唤醒：线程7
已通知线程7

预期输出：

线程7等待
线程1唤醒：线程7
已通知线程7
Thread-7已被唤醒

请帮忙。谢谢。：(

Answer 1

CustomerGenerator仅在调用时生成一个客户。制作一个mcve版本会非常清楚：

//i was initialized: i=0;
public void run()
{
while(i<1)
{
final Customer customer = new Customer(till_set);
customer.start();
i++;
}
}

我不认为那是你的意思
我发现mcve是一种非常有用的技术。它不仅让帮助变得更容易，而且是一个强大的调试工具。在很多情况下，在准备一个时，你很可能会发现问题。mcve应该演示问题，而不是您的应用程序。

代码中可能存在其他问题。如需更多帮助，请发布Mcve
其他一些评论：

在CustomerGenerator中，通过以下方式将所有收银台的引用传递给Customer：final Customer customer = new Customer(till_set);，后来用于选择收银台。我认为收银台选择计算最好在另一个类中完成，比如TillsManager，它可以让所有客户都在等待收银台的堆栈。

在定义的Driver中

public static Till till_object; 
for(i=0; i<5 ; i++)
{
till_object = new Till(item_queue);
till_set.add(till_object);
}

意味着您将在till_set中得到5倍相同的对象。我想你想要：

for(i=0; i<till_count; i++)
{
Till till_object = new Till(item_queue);
till_set.add(till_object);
}

Answer 2

只需从Till等待，直到队列获得超过零个元素。将自己添加到队列后，从客户线程通知Till.