基于java callable及future接口解决生产者消费者问题

  1 package net.xicp.tarbitrary.producerandcustomer;
  2 
  3 import java.util.LinkedList;
  4 import java.util.Random;
  5 import java.util.concurrent.Callable;
  6 import java.util.concurrent.ExecutionException;
  7 import java.util.concurrent.ExecutorService;
  8 import java.util.concurrent.Executors;
  9 import java.util.concurrent.FutureTask;
 10 import java.util.concurrent.TimeUnit;
 11 import java.util.concurrent.locks.Condition;
 12 import java.util.concurrent.locks.Lock;
 13 import java.util.concurrent.locks.ReentrantLock;
 14 
 15 
 16 public class NewFeatureTest {
 17 
 18  public static void main(String[] args) throws InterruptedException,
 19    ExecutionException {
 20   // 创建一个具有10个线程的线程池
 21   ExecutorService se = Executors.newFixedThreadPool(30);
 22   // 产生一张桌子，桌子的最大容量为10
 23   DiningTable table = new DiningTable(10);
 24   // 产生生产者对象
 25   Chef chef = new Chef(table);
 26   // 产生消费者对象
 27   Eater eater = new Eater(table);
 28   
 29   
 30   //本例只是模拟10个生产者与10个消费者，并没有完全按照题目的意思来进行解答
 31   for (int i = 1; i < 11; i++) {
 32    final FutureTask cheafTask = new FutureTask(chef);
 33    final FutureTask eaterTask = new FutureTask(eater);
 34    // 启动一个生产者线程
 35    se.execute(cheafTask);
 36    // 启动一个消费者线程
 37    se.execute(eaterTask);
 38    // 启动一个线程监听当前生产者线程的结果
 39    se.execute(new Runnable() {
 40     @Override
 41     public void run() {
 42      try {
 43       String st = cheafTask.get();
 44       System.out.println(st);
 45      } catch (InterruptedException e) {
 46       e.printStackTrace();
 47      } catch (ExecutionException e) {
 48       e.printStackTrace();
 49      }
 50 
 51     }
 52    });
 53    // 启动一个线程监听当前消费者线程的结果
 54    se.execute(new Runnable() {
 55     @Override
 56     public void run() {
 57      try {
 58       String st = eaterTask.get();
 59       System.out.println(st);
 60      } catch (InterruptedException e) {
 61       e.printStackTrace();
 62      } catch (ExecutionException e) {
 63       e.printStackTrace();
 64      }
 65     }
 66    });
 67    //有序地关闭前面已经启动的一些线程
 68   }
 69   se.shutdown();
 70  }
 71 }
 72 
 73 
 74 class Chef implements Callable {
 75  private DiningTable t;
 76  private Random random = new Random();
 77 
 78  public Chef(DiningTable table) {
 79   this.t = table;
 80  }
 81 
 82  public DiningTable getT() {
 83   return t;
 84  }
 85 
 86  public void setT(DiningTable t) {
 87   this.t = t;
 88  }
 89 
 90  public Dish produce() {
 91   try {
 92    // 模拟做菜用的时间,200-400毫秒一道菜
 93    // 跟下面一句的效果一致
 94    TimeUnit.MILLISECONDS.sleep(200 + random.nextInt(201));
 95    // Thread.sleep(200 + random.nextInt(201));
 96   } catch (InterruptedException e) {
 97    e.printStackTrace();
 98   }
 99   Dish food = new Dish();
100   return food;
101  }
102 
103  @Override
104  public String call() throws Exception {
105   for (int i = 0; i < 10; i++) {
106    Dish food = produce();
107    t.putFood(food);
108   }
109   return Thread.currentThread().getName() + "结束";
110  }
111 
112 }
113 
114 
115 class Eater implements Callable {
116  private DiningTable t;
117  private Random random = new Random();
118 
119  public Eater(DiningTable table) {
120   this.t = table;
121  }
122 
123  public DiningTable getT() {
124   return t;
125  }
126 
127  public void setT(DiningTable t) {
128   this.t = t;
129  }
130 
131  
132  public void consume(Dish food) {
133   try {
134    // 模拟吃菜用的时间
135    // 跟下面一句的效果一致
136    TimeUnit.MILLISECONDS.sleep(400 + random.nextInt(201));
137    // Thread.sleep(400 + random.nextInt(201));
138   } catch (InterruptedException e) {
139    e.printStackTrace();
140   }
141  }
142 
143  @Override
144  public String call() throws Exception {
145   for (int i = 0; i < 10; i++) {
146    Dish food = t.getFood();
147    consume(food);
148   }
149 
150   return Thread.currentThread().getName() + "结束";
151  }
152 
153 }
154 
155 
156 class Dish {
157  Random random = new Random();
158 
159  private String foodName;
160 
161  public Dish() {
162   super();
163   int posi = random.nextInt(name.length);
164   this.foodName = name[posi];
165  }
166 
167  String name[] = { "巫山烤鱼", "清炒小白菜 ", "糖醋里脊", "手撕包菜", "拔丝香蕉", "水煮肉片", "酸菜鱼",
168    "土豆肉片" };
169 
170  public String getName() {
171   return foodName;
172  }
173 
174 }
175 
176 
177 class DiningTable {
178  // 装食物用的容器
179  private LinkedList list = new LinkedList();
180  // 定义最大放菜量
181  private int maxSize;
182  // 创建一把锁
183  private Lock lock = new ReentrantLock();
184  // 获得当前lock对象的 Condition 实例
185  private Condition condition;
186 
187  public Lock getLock() {
188   return lock;
189  }
190 
191  public void setLock(Lock lock) {
192   this.lock = lock;
193  }
194 
195  public Condition getCondition() {
196   return condition;
197  }
198 
199  public void setCondition(Condition condition) {
200   this.condition = condition;
201  }
202 
203  public LinkedList getList() {
204   return list;
205  }
206 
207  public void setList(LinkedList list) {
208   this.list = list;
209  }
210 
211  public int getMaxSize() {
212   return maxSize;
213  }
214 
215  
216  public void setMaxSize(int maxSize) {
217   this.maxSize = maxSize;
218  }
219 
220  public DiningTable(int maxSize) {
221   super();
222   this.maxSize = maxSize;
223   this.condition = lock.newCondition();
224  }
225 
226  
227  public void putFood(Dish food) {
228   // 先进行锁定, 与synchronize起的作用类似
229   lock.lock();
230   try {
231    // 防止假唤醒
232    while (list.size() >= maxSize) {
233 
234     System.out.println( "菜已经上满了" + Thread.currentThread().getName() + ",让他们先吃,等消灭几盘再做菜");
235     // 如果桌子的菜已经放满了，则让当前线程等待
236     this.condition.await();
237    }
238    list.addLast(food);
239    System.out.println(Thread.currentThread().getName() + "烹饪了"
240      + food.getName());
241    // 通知所有休眠的食客线程继续吃菜
242    this.condition.signalAll();
243   } catch (Exception e) {
244    e.printStackTrace();
245   } finally {
246    // 解除锁定
247    lock.unlock();
248   }
249  }
250 
251  
252  public Dish getFood() {
253   // 先进行锁定, 与synchronize起的作用类似
254   lock.lock();
255   Dish dish = null;
256   try {
257    // 防止假唤醒
258    while (list.size() <= 0) {
259     System.out.println( "菜已经被吃光了" + Thread.currentThread().getName() + "歇会再吃");
260 
261      // 如果桌子上已经没有菜了，则让当前线程等待
262      this.condition.await();
263     }
264    dish = list.removeFirst();
265    System.out.println(Thread.currentThread().getName() + "吃掉了"
266      + dish.getName());
267    // 通知所有休眠的做菜线程继续做菜
268    this.condition.signalAll();
269   } catch (Exception e) {
270    e.printStackTrace();
271   } finally {
272    // 解除锁定
273    lock.unlock();
274   }
275 
276   return dish;
277  }
278 
279 }