3.5生产者与消费者

多生产者与多消费者-操作值：假死

假死 是所有线程都进入了waiting状态

下面是一个假死的例子

package com.myObject;

public class ProducerObject {
private String lock;

public ProducerObject(String lock) {
this.lock = lock;
}

public void p() {
synchronized (lock) {
try {
while (!"".equals(ValueObject.value)) {
System.out.println("P wait begin "+ System.currentTimeMillis());
lock.wait();
}
System.out.println("P wait end " + System.currentTimeMillis());
String value = System.currentTimeMillis() + "_"+ System.nanoTime();
ValueObject.value = value;
lock.notify();
} catch (InterruptedException e) {
e.printStackTrace();
}

}
}
}

package com.myObject;

public class CustomerObject {
private String lock;

public CustomerObject(String lock) {
this.lock = lock;
}

public void c() {
synchronized (lock) {
try {
while ("".equals(ValueObject.value)) {
System.out.println("C wait begin "+ System.currentTimeMillis());
lock.wait();
}
System.out.println("C wait end " + System.currentTimeMillis());
String value = "";
ValueObject.value = value;
lock.notify();
} catch (InterruptedException e) {
e.printStackTrace();
}

}
}
}

package com.myObject;

public class ValueObject {

public static String value = "";

}

package com.myThread;

import com.myObject.ProducerObject;

public class Thread1a extends Thread {
ProducerObject producerObject;

public Thread1a(ProducerObject producerObject) {
this.producerObject = producerObject;
}

@Override
public void run() {
while(true){
producerObject.p();
}
}
}

package com.myThread;

import com.myObject.CustomerObject;

public class Thread1b extends Thread {
CustomerObject customerObject;

public Thread1b(CustomerObject customerObject) {
this.customerObject = customerObject;
}

@Override
public void run() {
while(true){
customerObject.c();
}
}
}

package com.test;

import com.myObject.CustomerObject;
import com.myObject.ProducerObject;
import com.myThread.Thread1a;
import com.myThread.Thread1b;

public class Test1 {
public static void main(String[] args) throws InterruptedException{
String lock = new String("");
ProducerObject p= new ProducerObject(lock);//生产者
CustomerObject c = new CustomerObject(lock);//消费者
Thread1a[] arrThread1a = new Thread1a[3];
Thread1b[] arrThread1b = new Thread1b[3];
for(int i = 0;i < 3;i++){
arrThread1a[i] = new Thread1a(p);
arrThread1a[i].setName("p"+(i+1));
arrThread1b[i] = new Thread1b(c);
arrThread1b[i].setName("c"+(i+1));
arrThread1a[i].start();
arrThread1b[i].start();

}
Thread.sleep(5000);
Thread[] threads = new Thread[Thread.currentThread().getThreadGroup().activeCount()];
Thread.currentThread().getThreadGroup().enumerate(threads);
for(int  i= 0;i< threads.length;i++ ){
System.out.println(threads[i].getName()+" "+threads[i].getState());
}

}
}

打印结果

C wait begin 1453112093337

C wait begin 1453112093338

C wait begin 1453112093338

P wait end 1453112093338

P wait begin 1453112093338

P wait begin 1453112093338

P wait begin 1453112093338

main RUNNABLE

p1 WAITING

c1 WAITING

p2 WAITING

c2 WAITING

p3 WAITING

c3 WAITING

结果分析：问题出在lock.notify（）。比如p唤醒p，c唤醒c都会进入waiting，这样运行久了，会导致所有线程都进入waiting，程序最后进入假死，不能继续运行

解决方法是将notify改为notifyAll,这样无论通知“同类”还是“异类”，保证至少有一个线程还在运行

多生产者与多消费者-操作栈：解决wait条件改变与假死

多个生产者向List中放入数据，多个消费者从List中取出数据，List最大容量为1即控制List.size最大为1。

1)解决wait条件改变需将if改为while

wait条件改变错误结果

pop end 0

pop=anyStr0.04638034362500809

push end 1

Exception in thread "c3" java.lang.IndexOutOfBoundsException: Index: 0, Size: 0

at java.util.ArrayList.RangeCheck(ArrayList.java:547)

at java.util.ArrayList.get(ArrayList.java:322)

at com.myObject.StackObject.pop(StackObject.java:30)

at com.myObject.CustomerObject2.pop(CustomerObject2.java:11)

at com.myThread.Thread2b.run(Thread2b.java:15)

pop end 0

pop=anyStr0.98486043968257

pop begin and c2is waiting

2)解决wait条件改变之后还需要解决假死，则需将notify改为notifyAll

假死错误结果

pop=anyStr0.08915212675142581

push end 1

pop end 0

pop=anyStr0.24479882846755296

pop begin and c2is waiting

pop begin and c4is waiting

pop begin and c3is waiting

push end 1

先来看正确例子

package com.myObject;

public class CustomerObject2 {
private StackObject stackObject;

public CustomerObject2(StackObject stackObject) {
this.stackObject = stackObject;
}

public void pop(){
System.out.println("pop="+stackObject.pop());
}
}

package com.myObject;

public class ProducerObject2 {
private StackObject stackObject;

public ProducerObject2(StackObject stackObject) {
this.stackObject = stackObject;
}

public void push(){
stackObject.push();
}
}

package com.myThread;

import com.myObject.ProducerObject2;

public class Thread2a extends Thread {
ProducerObject2 producerObject2;

public Thread2a(ProducerObject2 producerObject2) {
this.producerObject2 = producerObject2;
}

@Override
public void run() {
while(true){
producerObject2.push();
}
}
}

package com.myThread;

import com.myObject.CustomerObject2;

public class Thread2b extends Thread {
CustomerObject2 customerObject2;

public Thread2b(CustomerObject2 customerObject2) {
this.customerObject2 = customerObject2;
}

@Override
public void run() {
while(true){
customerObject2.pop();
}
}
}

package com.test;

import com.myObject.CustomerObject2;
import com.myObject.ProducerObject2;
import com.myObject.StackObject;
import com.myThread.Thread2a;
import com.myThread.Thread2b;

public class Test2 {
public static void main(String[] args) throws InterruptedException{
StackObject stackObject = new StackObject();
ProducerObject2 p= new ProducerObject2(stackObject);
CustomerObject2 c = new CustomerObject2(stackObject);
Thread2a[] arrThread2a = new Thread2a[3];
Thread2b[] arrThread2b = new Thread2b[3];
for(int i = 0;i < 3;i++){
arrThread2a[i] = new Thread2a(p);
arrThread2a[i].setName("p"+(i+2));
arrThread2b[i] = new Thread2b(c);
arrThread2b[i].setName("c"+(i+2));
arrThread2a[i].start();
arrThread2b[i].start();

}

}
}

package com.myObject;

import java.util.ArrayList;
import java.util.List;

public class StackObject {
private List list = new ArrayList();
//生产者压栈
synchronized public void push() {
try {
while(list.size()==1){
this.wait();
}
list.add("anyStr"+Math.random());
this.notifyAll();
System.out.println("push end "+list.size());
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//消费者出栈
synchronized public String pop() {
String pop="";
try {
while(list.size()==0){
System.out.println("pop begin and "+Thread.currentThread().getName()+"is waiting");
this.wait();
}
pop=""+list.get(0);
list.remove(0);
this.notifyAll();
System.out.println("pop end "+list.size());
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return pop;
}

}

下面是关键代码错误的地方 if 和 notify

package com.myObject;

import java.util.ArrayList;
import java.util.List;

public class StackObject {
private List list = new ArrayList();
//生产者压栈
synchronized public void push() {
try {
if(list.size()==1){
this.wait();
}
list.add("anyStr"+Math.random());
this.notify();
System.out.println("push end "+list.size());
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//消费者出栈
synchronized public String pop() {
String pop="";
try {
if(list.size()==0){
System.out.println("pop begin and "+Thread.currentThread().getName()+"is waiting");
this.wait();
}
pop=""+list.get(0);
list.remove(0);
this.notify();
System.out.println("pop end "+list.size());
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return pop;
}

}