生产者消费者案例

女爷i 2021-09-25 08:06 391阅读 0赞

### Java生产者消费者 ###

*   *  synchronized模式
     *   *  消费者
         *  生产者
         *  main
     *  Lock 模式

## synchronized模式 ##

### 消费者 ###

@Data
    @NoArgsConstructor
    @AllArgsConstructor
    public class Consumer implements Runnable{ 
    	// 队列
        private ArrayBlockingQueue queue;
        // 线程名称
        private String threadName;
    
        @SneakyThrows
        @Override
        public void run() { 
            while(true) { 
                synchronized (queue) { 
                	// 如果队列为空，则让消费者等待
                    while (queue.size() == 0) { 
    // queue.notify();
                        System.out.println("队列为空");
                        queue.wait();
                    }
    				// 否则唤醒所有线程，让他们重新竞争资源。
                    System.out.println(threadName + " 消费了:" + queue.take());
                    queue.notifyAll();
                }
            }
        }
    }

### 生产者 ###

@Data
    @NoArgsConstructor
    @AllArgsConstructor
    public class Producer implements Runnable{ 
    	// 队列
        private ArrayBlockingQueue queue;
        // 线程名称
        private String threadName;
    
        @SneakyThrows
        @Override
        public void run() { 
            while(true) { 
                synchronized (queue) { 
                    // 如果队列达到最大值（假设为10）
                    while (queue.size() == 10) { 
    // queue.notify();
                        System.out.println("队列满了");
                        // 停止生产
                        queue.wait();
                    }
                    // 如果队列没满 就添加元素到队列
                    int num = (int) Math.round(Math.random() * 10);
                    System.out.println(threadName + " 生产了: " + num);
                    queue.add(num);
                    // 并通知消费者线程
                    queue.notifyAll();
    
                    Thread.sleep(100);
                }
            }
        }
    }

### main ###

public static void main(String[] args) { 
        // 手动创建线程池
        ExecutorService pool = new ThreadPoolExecutor(5, 5,
                10, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>()
        );
    	// 创建队列
        ArrayBlockingQueue queue = new ArrayBlockingQueue(10);
        // 生产者
        Producer producer1 = new Producer(queue,"producer01");
        Producer producer2 = new Producer(queue,"producer02");
        // 消费者
        Consumer c1 = new Consumer(queue,"consumer01");
        Consumer c2 = new Consumer(queue,"consumer02");
    	
    	// 启动线程
        pool.submit(producer1);
        pool.submit(c1);
        pool.submit(producer2);
        pool.submit(c2);
    
    }

## Lock 模式 ##

public class SyncPandC { 
    
        private final int MAX_LEN = 10;
        // 队列
        private Queue<Integer> queue = new LinkedList<Integer>();
        // 锁
        private Lock lock = new ReentrantLock();
        private Condition condition = lock.newCondition();
    
        class Producer extends Thread { 
            @Override
            public void run() { 
                while (true) { 
                    lock.lock();
                    try { 
                        while (queue.size() == MAX_LEN) { 
                            System.out.println("当前队列满");
                            try { 
                                condition.await();	// 注意是await
                            } catch (InterruptedException e) { 
                                e.printStackTrace();
                            }
                        }
                        queue.add(1);
                        // 唤醒
                        condition.signalAll();
                        System.out.println("生产者生产一条任务，当前队列长度为" + queue.size());
                        try { 
                            Thread.sleep(500);
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    } catch (Exception e) { 
                        e.printStackTrace();
                    } finally { 
                        lock.unlock();
                    }
                }
            }
        }
    
        class Consumer extends Thread { 
    
            @Override
            public void run() { 
                while (true) { 
                    lock.lock();
                    try { 
                        while (queue.size() == 0) { 
                            System.out.println("当前队列为空");
                            try { 
                                condition.await();
                            } catch (InterruptedException e) { 
                                e.printStackTrace();
                            }
                        }
                        queue.poll();
                        condition.signal();
                        System.out.println("消费者消费一条任务，当前队列长度为" + queue.size());
                        try { 
                            Thread.sleep(500);
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    } catch (Exception e) { 
                        e.printStackTrace();
                    } finally { 
                        lock.unlock();
                    }
    
                }
            }
        }
    
        public static void main(String[] args) { 
            SyncPandC spc = new SyncPandC();
            spc.new Producer().start();
            spc.new Consumer().start();
        }
    
    }