生产者消费者问题是研究多线程程序时绕不开的经典问题之一, 它描述是有一块缓冲区作为仓库, 生产者可以将产品放入仓库, 消费者则可以从仓库中取走产品
生产者消费者问题
生产者消费者问题是研究多线程程序时绕不开的经典问题之一, 它描述是有一块缓冲区作为仓库, 生产者可以将产品放入仓库, 消费者则可以从仓库中取走产品. 解决生产者 / 消费者问题的方法可分为两类:
- 采用某种机制保护生产者和消费者之间的同步；
- 在生产者和消费者之间建立一个管道.
第一种方式有较高的效率, 并且易于实现, 代码的可控制性较好, 属于常用的模式. 第二种管道缓冲区不易控制, 被传输数据对象不易于封装等, 实用性不强.
同步问题核心在于: 如何保证同一资源被多个线程并发访问时的完整性. 常用的同步方法是采用信号或加锁机制, 保证资源在任意时刻至多被一个线程访问. 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();
}
}
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();
}
}
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();
}
}
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();
}
}
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();
}
}