Black House

Java 实现生产者消费者问题

1/9/2016 JavaSE Concurrent

生产者消费者问题是研究多线程程序时绕不开的经典问题之一, 它描述是有一块缓冲区作为仓库, 生产者可以将产品放入仓库, 消费者则可以从仓库中取走产品

生产者消费者问题

生产者消费者问题是研究多线程程序时绕不开的经典问题之一, 它描述是有一块缓冲区作为仓库, 生产者可以将产品放入仓库, 消费者则可以从仓库中取走产品. 解决生产者/消费者问题的方法可分为两类:

  1. 采用某种机制保护生产者和消费者之间的同步；
  2. 在生产者和消费者之间建立一个管道.

第一种方式有较高的效率, 并且易于实现, 代码的可控制性较好, 属于常用的模式. 第二种管道缓冲区不易控制, 被传输数据对象不易于封装等, 实用性不强.

同步问题核心在于: 如何保证同一资源被多个线程并发访问时的完整性. 常用的同步方法是采用信号或加锁机制, 保证资源在任意时刻至多被一个线程访问. Java语言在多线程编程上实现了完全对象化, 提供了对同步机制的良好支持. 在Java中一共有五种方法支持同步, 其中前四个是同步方法, 一个是管道方法.

  • wait() / notify()方法
  • await() / signal()方法
  • BlockingQueue阻塞队列方法
  • Semaphore方法
  • PipedInputStream / PipedOutputStream

wait() / notify() 方法

wait() / nofity()方法是基类Object的两个方法, 也就意味着所有Java类都会拥有这两个方法, 这样, 我们就可以为任何对象实现同步机制.

wait()方法: 当缓冲区已满/空时, 生产者/消费者线程停止自己的执行, 放弃锁, 使自己处于等等状态, 让其他线程执行.

notify()方法: 当生产者/消费者向缓冲区放入/取出一个产品时, 向其他等待的线程发出可执行的通知, 同时放弃锁, 使自己处于等待状态.

public class Hosee {
    private static       Integer count = 0;
    private static final Integer FULL  = 10;
    private static final String  LOCK  = "LOCK";

    // 生产者
    class Producer implements Runnable {
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                synchronized (LOCK) {
                    while (count.equals(FULL)) {
                        try {
                            // 释放锁,进去等待池等待唤醒,生产者停止生产商品
                            LOCK.wait();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    count++;
                    System.out.println(Thread.currentThread().getName() + "生产者生产, 目前总共有" + count);
                    // 唤醒等待池中的所有线程,这里唤醒消费者消费商品
                    LOCK.notifyAll();
                }
            }
        }
    }

    // 消费者
    class Consumer implements Runnable {
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    // 不放弃同步锁
                    Thread.sleep(3000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                synchronized (LOCK) {
                    while (count == 0) {
                        try {
                            // 放弃锁,进入等待池等待被唤醒, 消费者停止消费商品
                            LOCK.wait();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    count--;
                    System.out.println(Thread.currentThread().getName() + "消费者消费, 目前总共有" + count);
                    // 唤醒生产者生产商品
                    LOCK.notifyAll();
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Hosee hosee = new Hosee();
        new Thread(hosee.new Producer()).start();
        new Thread(hosee.new Consumer()).start();
        new Thread(hosee.new Producer()).start();
        new Thread(hosee.new Consumer()).start();
        new Thread(hosee.new Producer()).start();
        new Thread(hosee.new Consumer()).start();
        new Thread(hosee.new Producer()).start();
        new Thread(hosee.new Consumer()).start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

await() / signal() 方法

wait()和notify()必须在synchronized的代码块中使用 因为只有在获取当前对象的锁时才能进行这两个操作 否则会报异常 而await()和signal()一般与Lock()配合使用. wait是Object的方法, 而await只有部分类有, 如Condition. await()/signal()和新引入的锁定机制Lock直接挂钩, 具有更大的灵活性.

public class Test2 {
    private static Integer   count    = 0;
    private final  Integer   FULL     = 10;
    final          Lock      lock     = new ReentrantLock();
    final          Condition NotFull  = lock.newCondition();
    final          Condition NotEmpty = lock.newCondition();

    class Producer implements Runnable {
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                lock.lock();
                try {
                    while (count == FULL) {
                        try {
                            NotFull.await();
                        } catch (InterruptedException e) {
                            // TODO Auto-generated catch block
                            e.printStackTrace();
                        }
                    }
                    count++;
                    System.out.println(Thread.currentThread().getName()
                            + "生产者生产, 目前总共有" + count);
                    NotEmpty.signal();
                } finally {
                    lock.unlock();
                }

            }
        }
    }

    class Consumer implements Runnable {

        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                lock.lock();
                try {
                    while (count == 0) {
                        try {
                            NotEmpty.await();
                        } catch (Exception e) {
                            // TODO: handle exception
                            e.printStackTrace();
                        }
                    }
                    count--;
                    System.out.println(Thread.currentThread().getName()
                            + "消费者消费, 目前总共有" + count);
                    NotFull.signal();
                } finally {
                    lock.unlock();
                }

            }

        }

    }

    public static void main(String[] args) throws Exception {
        Hosee hosee = new Hosee();
        new Thread(hosee.new Producer()).start();
        new Thread(hosee.new Consumer()).start();
        new Thread(hosee.new Producer()).start();
        new Thread(hosee.new Consumer()).start();

        new Thread(hosee.new Producer()).start();
        new Thread(hosee.new Consumer()).start();
        new Thread(hosee.new Producer()).start();
        new Thread(hosee.new Consumer()).start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

BlockingQueue 阻塞队列方法

put()方法: 类似于我们上面的生产者线程, 容量达到最大时, 自动阻塞. take()方法: 类似于我们上面的消费者线程, 容量为0时, 自动阻塞.

public class Hosee {
	private static Integer count = 0;
	final BlockingQueue<Integer> bq = new ArrayBlockingQueue<Integer>(10);
	class Producer implements Runnable {
		public void run() {
			for (int i = 0; i < 10; i++) {
				try {
					Thread.sleep(3000);
				} catch (Exception e) {
					e.printStackTrace();
				}
				try {
					bq.put(1);
					count++;
					System.out.println(Thread.currentThread().getName()
							+ "生产者生产, 目前总共有" + count);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	}

	class Consumer implements Runnable {
		public void run() {
			for (int i = 0; i < 10; i++) {
				try {
					Thread.sleep(3000);
				} catch (InterruptedException e1) {
					e1.printStackTrace();
				}
				try {
					bq.take();
					count--;
					System.out.println(Thread.currentThread().getName()
							+ "消费者消费, 目前总共有" + count);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}

	}

	public static void main(String[] args) throws Exception {
		Hosee hosee = new Hosee();
		new Thread(hosee.new Producer()).start();
		new Thread(hosee.new Consumer()).start();
		new Thread(hosee.new Producer()).start();
		new Thread(hosee.new Consumer()).start();
		new Thread(hosee.new Producer()).start();
		new Thread(hosee.new Consumer()).start();
		new Thread(hosee.new Producer()).start();
		new Thread(hosee.new Consumer()).start();
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

Semaphore 方法

Semaphore 信号量, 就是一个允许实现设置好的令牌. 也许有1个, 也许有10个或更多. 谁拿到令牌(acquire)就可以去执行了, 如果没有令牌则需要等待. 执行完毕, 一定要归还(release)令牌, 否则令牌会被很快用光, 别的线程就无法获得令牌而执行下去了

public class Hosee{
	int count = 0;
	final Semaphore notFull = new Semaphore(10);
	final Semaphore notEmpty = new Semaphore(0);
	final Semaphore mutex = new Semaphore(1);
	class Producer implements Runnable{
		public void run(){
			for (int i = 0; i < 10; i++){
				try{
					Thread.sleep(3000);
				}
				catch (Exception e){
					e.printStackTrace();
				}
				try{
					notFull.acquire();//顺序不能颠倒, 否则会造成死锁. 
					mutex.acquire();
					count++;
					System.out.println(Thread.currentThread().getName()
							+ "生产者生产, 目前总共有" + count);
				}
				catch (Exception e){
					e.printStackTrace();
				}
				finally{
					mutex.release();
					notEmpty.release();
				}
			}
		}
	}

	class Consumer implements Runnable{
		public void run(){
			for (int i = 0; i < 10; i++){
				try{
					Thread.sleep(3000);
				}
				catch (InterruptedException e1){
					e1.printStackTrace();
				}
				try{
					notEmpty.acquire();//顺序不能颠倒, 否则会造成死锁. 
					mutex.acquire();
					count--;
					System.out.println(Thread.currentThread().getName()
							+ "消费者消费, 目前总共有" + count);
				}
				catch (Exception e){
					e.printStackTrace();
				}
				finally{
					mutex.release();
					notFull.release();
				}
			}
		}
	}
	public static void main(String[] args) throws Exception{
		Hosee hosee = new Hosee();
		new Thread(hosee.new Producer()).start();
		new Thread(hosee.new Consumer()).start();
		new Thread(hosee.new Producer()).start();
		new Thread(hosee.new Consumer()).start();
		new Thread(hosee.new Producer()).start();
		new Thread(hosee.new Consumer()).start();
		new Thread(hosee.new Producer()).start();
		new Thread(hosee.new Consumer()).start();
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

PipedInputStream / PipedOutputStream

这个类位于java.io包中, 是解决同步问题的最简单的办法, 一个线程将数据写入管道, 另一个线程从管道读取数据, 这样便构成了一种生产者/消费者的缓冲区编程模式. PipedInputStream/PipedOutputStream只能用于多线程模式, 用于单线程下可能会引发死锁.

public class Hosee {
	final PipedInputStream pis = new PipedInputStream();
	final PipedOutputStream pos = new PipedOutputStream();
	{
		try {
			pis.connect(pos);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	class Producer implements Runnable {
		@Override
		public void run() {
			try{
                while(true){
                    int b = (int) (Math.random() * 255);
                    System.out.println("Producer: a byte, the value is " + b);
                    pos.write(b);
                    pos.flush();
                }
            }catch(Exception e){
                e.printStackTrace();
            }finally{
                try{
                    pos.close();
                    pis.close();
                }catch(IOException e){
                    System.out.println(e);
                }
            }
		}
	}

	class Consumer implements Runnable {

		@Override
		public void run() {
			try{
                while(true){
                    int b = pis.read();
                    System.out.println("Consumer: a byte, the value is " + String.valueOf(b));
                }
            }catch(Exception e){
                e.printStackTrace();
            }finally{
                try{
                    pos.close();
                    pis.close();
                }catch(IOException e){
                    System.out.println(e);
                }
            }
		}

	}

	public static void main(String[] args) throws Exception {
		Hosee hosee = new Hosee();
		new Thread(hosee.new Producer()).start();
		new Thread(hosee.new Consumer()).start();
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
Last Updated: 7/3/2019, 6:17:56 PM