Java多线程间通信

作者: 学编程的小屁孩 | 来源:发表于2020-02-21 19:46 被阅读0次

1.多线程之等待与通知机制
1.1什么是等待通知机制？
在生活中，如我们去饭店，服务员拿菜单给我们点菜，然后记录完告诉厨房去做，然后服务员就处于等待状态了。然后厨师把菜做好，就通知服务员把菜端上去。然后服务员被通知之后，就处于被唤醒，然后把菜端过去，这就是一个等待通知的过程。

1.2等待通知机制的实现
1.2.1.等待:
<a>.在调用wait之前，线程必须获得该对象的对象级别锁，即只能在同步方法或者同步代码块中才能调用wait()方法。

 <b>.在执行wait()方法后，当前线程释放锁。此时在wait()方法返回前，其他的线程可以竞争获得此锁。

 <c>.如果wait()方法被调用时，没有持有适当的锁，那么就会报一个IllegalMonitorStateException异常。

 <d>.wait()所在代码行处停止执行的条件是接到notify()通知或者被中断。

 1.2.2通知：
  <a>.同样notify();方法也需要，线程获得对象的对象级别锁，即只能在同步方法或者同步代码块中才能调用notify();方法。

  <b>.notify();方法被调用时，如果当前线程没有持有适当的锁，那么就会报一个IllegalMonitorStateException异常。

  <c>.当前如果有多个线程处于等待的状态，那么使用notify();方法，线程规划器只会随机的挑选出其中一个线程。

最重要的一点：等待通知的执行顺序是如果我们执行了notify()；方法并不会马上就释放线程了，而那个wait状态的线程也不能马 

上获取该对象锁。得等到notify();方法的线程执行完（也就是退出了synchronized代码块后），线程才会释放锁，而呈wait()；

状态的线程菜可以获取该对象锁。

1.2.3实现：

public class ThreadA extends Thread{
private Object lock;

public ThreadA(Object lock) {
    super();
    this.lock = lock;
}

@Override
public void run() {
    try{
        synchronized(lock){
            System.out.println("start  wait time="+System.currentTimeMillis());
            lock.wait();
            System.out.println("end  wait time="+System.currentTimeMillis());
        }
    }catch(InterruptedException e){
        e.printStackTrace();
    }
}

}

public class ThreadB extends Thread{
private Object lock;

public ThreadB(Object lock) {
    super();
    this.lock = lock;
}

@Override
public void run() {
    synchronized(lock){
        System.out.println("start  notify time="+System.currentTimeMillis());
        lock.notify();
        System.out.println("end  notify time="+System.currentTimeMillis());
    }
}

}

public class Test {

public static void main(String[] args) {
    Object lock=new Object();
    ThreadA a=new ThreadA(lock);
    a.start();
    ThreadB b=new ThreadB(lock);
    b.start();
}

}
结果：
start wait time=1474785491542
start notify time=1474785491542
end notify time=1474785491542
end wait time=1474785491542

1.3线程的状态
一般我们新创建一个新的线程对象后，再去调用start();方法，线程就会为此分配CPU资源，此时线程对象处于 "可运行态"。如果竞争到了资源，那么线程对象就处于运行态了。

 1.3.1可运行态出现的情况
  <a>.调用sleep();方法后经过的时间超过了指定的睡眠时间。

  <b>.线程调用的阻塞的IO已经返回，阻塞方法执行完毕了。

  <c>.线程成功地获得了试图同步的监视器。

  <d>.线程正在等待某个通知，其他线程发出了通知。

  <e>.处于挂起状态的线程调用了resume();恢复方法。

  1.3.2阻塞状态出现的情况
  <a>.线程调用sleep();方法，主动放弃占用的处理器资源。

  <b>.线程调用了阻塞式IO方法，在该方法返回前，该线程被阻塞。

  <c>.线程试图获得一个同步监视器，但该同步监视器正被其他线程所持有。

  <d>.线程等待某个通知。

  <e>.程序调用了suspend();方法将该线程挂起。

清楚一点：锁对象，都有两个队列，一个是就绪队列，一个是阻塞队列。就绪队列存储了将要获得锁的线程而阻塞队列存储了被阻塞的线程。过程就是一个线程被唤醒之后，就进入了就绪队列，等待CPU的调度。如果一个线程被wait后，就会进入到阻塞队

列，等待下一次被唤醒。

1.4唤醒所有线程
唤醒单个线程使用notify();唤醒所有线程采用notifyAll();方法。特点就是让所有正在等待队列中等待同一共享资源的全部线程从等待状态退出，进入可运行状态。优先级最高的那个线程最先执行，但是也可能会随机执行，这取决于JVM是如何实现的了。

1.5方法wait(long time);的使用
功能就是等待某一时间内是否有线程对锁进行唤醒。如果超过这个时间，则自动唤醒。

public class MyRunnable {
static private Object lock=new Object();
static private Runnable runnable=new Runnable(){

    @Override
    public void run() {
        // TODO Auto-generated method stub
        try{
            synchronized (lock) {
                System.out.println("start wait time="+System.currentTimeMillis());
                lock.wait(5000);
                System.out.println("end   wait time="+System.currentTimeMillis());
            }
        }catch(InterruptedException e){
            e.printStackTrace();
        }
    }
};
public static void main(String[] args) {
    Thread t=new Thread(runnable);
    t.start();
}

}
结果：
start wait time=1474788690635
end wait time=1474788695635

1.6等待通知机制的特殊情况
1.6.1当interrupt();方法遇到wait();方法
当线程呈wait();状态时，调用线程对象的interrupt();方法会报一个InterruptedException异常。

public class Service {
public void service(Object lock){
try {
synchronized (lock) {
System.out.println("begin wait");
lock.wait();
System.out.println("end wait");
}
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
System.out.println("产生异常，处于等待态的线程被中断了！");
}
}
}

public class MyThread extends Thread{

private Object lock;
public MyThread(Object lock) {
    super();
    this.lock = lock;
}

@Override
public void run() {
    Service service=new Service();
    service.service(lock);
}

}

public class Test {

public static void main(String[] args) {
    try{
        Object lock=new Object();
        MyThread t=new MyThread(lock);
        t.start();
        Thread.sleep(5000);
        t.interrupt();
    }catch(InterruptedException e){
        e.printStackTrace();
    }
}

}
结果：

1.6.2通知过早
 一般都是线程对象已经处于等待状态，然后再去通知。如果我们通知过早，那么就会打乱程序的正常运行逻辑。那么此时就不用

 等待了。

public class Test {
private Object lock=new Object();
private Runnable runnable1=new Runnable(){

    @Override
    public void run() {
        // TODO Auto-generated method stub
        try{
            synchronized (lock) {
                System.out.println("start wait");
                lock.wait();
                System.out.println("end   wait");
            }
        }catch(InterruptedException e){
            e.printStackTrace();
        }
    }
};
private Runnable runnable2=new Runnable(){

    @Override
    public void run() {
        // TODO Auto-generated method stub
        synchronized (lock) {
            System.out.println("start notify");
            lock.notify();
            System.out.println("end   notify");
        }
    }
};
public static void main(String[] args) {
    Test test=new Test();
    Thread t2=new Thread(test.runnable2);
    t2.start();
    Thread t1=new Thread(test.runnable1);
    t1.start();
}

}
结果：

 解释：由于先执行了notify();方法，则后续的wait只能一直等待了。解决办法，就是让wait线程先执行，然后避免notify先执行。

1.7生产者与消费者
1.7.1一生产一消费：操作值

public class Producer {
private String lock;
public Producer(String lock) {
super();
this.lock = lock;
}
public void setValue(){
try{
synchronized (lock) {
if(!ValueObject.value.equals("")){
lock.wait();
}
String value=System.currentTimeMillis()+"_"+System.nanoTime();
System.out.println("set的值是"+value);
ValueObject.value=value;
lock.notify();
}
}catch(InterruptedException e){
e.printStackTrace();
}
}
}

public class Consumer {
private String lock;
public Consumer(String lock) {
super();
this.lock = lock;
}
public void getValue(){
try{
synchronized (lock) {
if(ValueObject.value.equals("")){
lock.wait();
}
System.out.println("get的值是"+ValueObject.value);
ValueObject.value="";
lock.notify();
}
}catch(InterruptedException e){
e.printStackTrace();
}
}
}

public class ThreadA extends Thread{
private Producer producer;
public ThreadA(Producer producer) {
super();
this.producer = producer;
}
@Override
public void run() {
while(true){
producer.setValue();
}
}
}

public class ThreadB extends Thread{
private Consumer consumer;
public ThreadB(Consumer consumer) {
super();
this.consumer = consumer;
}
@Override
public void run() {
while(true){
consumer.getValue();
}
}
}

public class Test {
public static void main(String[] args) {
String lock=new String("");
Producer producer=new Producer(lock);
Consumer consumer=new Consumer(lock);
ThreadA a=new ThreadA(producer);
ThreadB b=new ThreadB(consumer);
a.start();
b.start();
}
}
结果：
set的值是1474810313983_38415328499642
get的值是1474810313983_38415328499642
set的值是1474810313983_38415328514832
get的值是1474810313983_38415328514832
set的值是1474810313983_38415328526738
get的值是1474810313983_38415328526738
。。。。。

 1.7.2多生产与多消费---操作值

假死情况：就是由于当前是多个生产者、多个消费者，而可能通知是同类之间的通知，如生产者通知生产者，消费者没有被通

知，最后消费者都在等待，生产者看产品未消费，也等待，最后都等待了。结果就是所有的线程都进入了等待态，最后整个项

目都停止了。

例子：

生产者和消费者及其对应的线程都不变。就是多造几个线程。

public class Test {
public static void main(String[] args) throws InterruptedException {
String lock=new String("");
Producer producer=new Producer(lock);
Consumer consumer=new Consumer(lock);
ThreadA[] tA=new ThreadA[2];
ThreadB[] tB=new ThreadB[2];
for(int i=0;i<2;i++){
tA[i]=new ThreadA(producer);
tA[i].setName("生产者"+(i+1));
tB[i]=new ThreadB(consumer);
tB[i].setName("消费者"+(i+1));
tA[i].start();
tB[i].start();
}
Thread.sleep(5000);
Thread[] threadArray=new Thread[Thread.currentThread().getThreadGroup().activeCount()];
Thread.currentThread().getThreadGroup().enumerate(threadArray);
for(int i=0;i<threadArray.length;i++){
System.out.println(threadArray[i].getName()+" "+threadArray[i].getState());
}
}
}
分析：造成这种结果就是由于notify();方法本身不具有指定性的通知哪一个具体的等待态的线程。所以造成随意通知一个处于等待

态的线程。并且假死的关键在于消费者没有被通知，所以只要保证任一时刻有生产者生产，有消费者消费就OK了。所以解决这种

假死的方式是使用notifyAll();方法，将所有的线程都唤醒，这样消费者还是生产者就都会被唤醒。

  1.7.3一生产与一消费---操作栈

public class MyStack {
private List<String> list=new ArrayList<String>();
public synchronized void push(){
try{
if(list.size() ==1){
this.wait();
}
list.add("anyString="+Math.random());
this.notify();
System.out.println("push="+list.size());
}catch(InterruptedException e){
e.printStackTrace();
}
}

public synchronized String pop(){
    String returnValue="";
    try{
        if(list.size() ==0){
            System.out.println("pop操作中的:"+Thread.currentThread().getName()+"线程呈wait状态");
            this.wait();
        }
        returnValue=""+list.get(0);
        list.remove(0);
        this.notify();
        System.out.println("pop="+list.size());
    }catch(InterruptedException e){
        e.printStackTrace();
    }
    return returnValue;
}

}

public class Producer {
private MyStack myStack;
public Producer(MyStack myStack) {
super();
this.myStack = myStack;
}
public void pushService(){
myStack.push();
}
}

public class Consumer {
private MyStack myStack;
public Consumer(MyStack myStack) {
super();
this.myStack = myStack;
}
public void popService(){
myStack.pop();
}
}

public class PThread extends Thread{
private Producer producer;
public PThread(Producer producer) {
super();
this.producer = producer;
}
@Override
public void run() {
while(true){
producer.pushService();
}
}
}

public class CThread extends Thread{
private Consumer consumer;
public CThread(Consumer consumer) {
super();
this.consumer = consumer;
}
@Override
public void run() {
while(true){
consumer.popService();
}
}
}

public class Test {
public static void main(String[] args) {
MyStack myStack=new MyStack();
Producer producer=new Producer(myStack);
Consumer consumer=new Consumer(myStack);
PThread pThread=new PThread(producer);
CThread cThread=new CThread(consumer);
pThread.start();
cThread.start();
}
}
结果：
push=1
pop=0
pop操作中的:Thread-1线程呈wait状态
push=1
pop=0
pop操作中的:Thread-1线程呈wait状态

。。。。。。。。

解释：为了有效的生产者生产一个产品，消费者则消费一个产品，如果生产者生产的产品未消费则等待，消费者没有可消费的产品

也等待，那么采用list集合来实现栈，同栈的长度是否为1来判断当前产品是否已经消费，如果长度为1则生产者等待，消费者消费

如果长度为0则生产者生产，消费者等待。这样就实现了所谓的一生产一消费操作栈。

 1.7.4一生产多消费---操作栈
对于生产者，消费者及其对应所在的线程都不变，只是消费者多了几个。

public class Test {
public static void main(String[] args) {
MyStack myStack=new MyStack();
Producer producer=new Producer(myStack);
Consumer consumer1=new Consumer(myStack);
Consumer consumer2=new Consumer(myStack);
Consumer consumer3=new Consumer(myStack);
Consumer consumer4=new Consumer(myStack);
Consumer consumer5=new Consumer(myStack);
PThread pThread=new PThread(producer);
pThread.start();
CThread cThread1=new CThread(consumer1);
CThread cThread2=new CThread(consumer2);
CThread cThread3=new CThread(consumer3);
CThread cThread4=new CThread(consumer4);
CThread cThread5=new CThread(consumer5);
cThread1.start();
cThread2.start();
cThread3.start();
cThread4.start();
cThread5.start();
}
}
结果：

解决办法：将Mystack中的if语句换成while语句即可，这样就没有异常了。而对于假死，还是由于消费者没有对应的生产者生产，

生产者没有对应的消费者消费的问题，所以采用notifyAll();都唤醒就ok 了。

同理对于多生产一消费和多生产多消费都可以手动去实现了，就是多增加了消费者或生产者，这儿就不写了。。。

1.8管道流实现线程间通信
java提供了PipedStream，管道流用于在不同线程之间直接传输数据，形象点就是在两个线程之间设立的管道传输。对应的肯定有

管道字节流和管道字符流了。

  1.8.1管道字节流：

public class OperateData {
public void writeData(PipedOutputStream out){
try{
String outData="我的女神！";
out.write(outData.getBytes());
out.close();
}catch(IOException e){
e.printStackTrace();
}
}
public void readData(PipedInputStream input){
try{
System.out.println("read :");
byte[] buf=new byte[20];
int length=0;
while((length=input.read(buf))!=-1){
System.out.println(new String(buf,0,length));
}
input.close();
}catch(IOException e){
e.printStackTrace();
}
}
}

public class WThread extends Thread{
private OperateData operateData;
private PipedOutputStream out;
public WThread(OperateData operateData, PipedOutputStream out) {
super();
this.operateData = operateData;
this.out = out;
}
@Override
public void run() {
operateData.writeData(out);
}
}

public class RThread extends Thread{
private OperateData operateData;
private PipedInputStream input;
public RThread(OperateData operateData, PipedInputStream input) {
super();
this.operateData = operateData;
this.input = input;
}
@Override
public void run() {
operateData.readData(input);
}
}

public class Test {
public static void main(String[] args) {
try{
OperateData operateData=new OperateData();
PipedInputStream input=new PipedInputStream();
PipedOutputStream out=new PipedOutputStream();
out.connect(input);
RThread rThread=new RThread(operateData,input);
rThread.start();
Thread.sleep(2000);
WThread wThread=new WThread(operateData,out);
wThread.start();
}catch(Exception e){
e.printStackTrace();
}
}
}

结果：
read :
我的女神！

2.避免线程提前结束之join方法的使用
join();方法的作用是等待线程对象销毁。并且join方法可以让线程排队运行。这个看起来有点像同步的意思。

2.1join与synchronized的区别：
join在内部使用wait();方法进行等待，而synchronized关键字使用的是 "对象监视器"原理做同步。

2.2join方法和异常：
在执行join的过程中，如果当前线程对象被中断，则当前线程出现异常。会报一个InterruptedException异常。

2.3join(long time)的使用：
方法join(long time);中的参数是设定等待的时间。

public class MyThread extends Thread{
@Override
public void run() {
try{
System.out.println("begin time="+System.currentTimeMillis());
Thread.sleep(5000);
}catch(InterruptedException e){
e.printStackTrace();
}
}
}

public class Test {
public static void main(String[] args) {
try{
MyThread myThread=new MyThread();
myThread.start();
myThread.join(2000);
System.out.println("end time="+System.currentTimeMillis());
}catch(InterruptedException e){
e.printStackTrace();
}
}
}
结果：
begin time=1474859755314
end time=1474859757314

2.4join(long time);方法和sleep(long time);方法的区别
方法join(long time);的功能在内部是使用wait(long time);方法来实现的，所以join(long time);方法具有释放锁的特点。而sleep(long);

 方法在线程睡眠时，并不释放锁。

2.5 join();方法后面的代码提前运行

public class ThreadB extends Thread{
@Override
public synchronized void run() {
try{
System.out.println("begin B ThreadName="+Thread.currentThread().getName()+":"+
System.currentTimeMillis());
Thread.sleep(5000);
System.out.println("end B ThreadName="+Thread.currentThread().getName()+":"+
System.currentTimeMillis());
}catch(InterruptedException e){
e.printStackTrace();
}
}
}

public class ThreadA extends Thread{
private ThreadB b;
public ThreadA(ThreadB b) {
super();
this.b = b;
}
@Override
public void run() {
try{
synchronized (b) {
System.out.println("begin A ThreadName="+Thread.currentThread().getName()+":"+
System.currentTimeMillis());
Thread.sleep(5000);
System.out.println("end A ThreadName="+Thread.currentThread().getName()+":"+
System.currentTimeMillis());
}
}catch(InterruptedException e){
e.printStackTrace();
}
}
}

public class Test {
public static void main(String[] args) {
try{
ThreadB b =new ThreadB();
ThreadA a =new ThreadA(b);
a.start();
b.start();
b.join(2000);
System.out.println(" main end="+System.currentTimeMillis());
}catch(InterruptedException e){
e.printStackTrace();
}
}
}

结果1：
begin A ThreadName=Thread-1:1474861031764
end A ThreadName=Thread-1:1474861036772
begin B ThreadName=Thread-0:1474861036772
main end=1474861036772
end B ThreadName=Thread-0:1474861041780

结果2：

begin A ThreadName=Thread-1:1474861153643
end A ThreadName=Thread-1:1474861158651
main end=1474861158651
begin B ThreadName=Thread-0:1474861158651
end B ThreadName=Thread-0:1474861163660

 分析：对于每次运行产生的不同结果，原因在于b.join(2000);执行后，抢到b锁之后，b线程等待2秒，然后将锁进行了释放，那么

 等到2秒执行完，就开始执行main end，然后b等待时间到了才获得锁，开始执行。或者再b线程等待2秒之后，然后将锁释放，b

 抢到锁打印Thread begin，然后这时main end是异步输出了；然后b打印了剩下了的Thread end。

3.ThreadLocal的使用
ThreadLocal类主要解决的就是每个线程绑定自己的值，就是用来存储每个线程的私有数据。

public class Tools {
public static ThreadLocal<String> tl=new ThreadLocal<String>();
}

public class ThreadA extends Thread{
@Override
public void run() {
try{
for(int i=0;i<100;i++){
Tools.tl.set("ThreadA"+(i+1));
System.out.println("ThreadA get Value="+Tools.tl.get());
Thread.sleep(200);
}
}catch(InterruptedException e){
e.printStackTrace();
}
}
}
public class ThreadB extends Thread{
@Override
public void run() {
try{
for(int i=0;i<100;i++){
Tools.tl.set("ThreadB"+(i+1));
System.out.println("ThreadB get Value="+Tools.tl.get());
Thread.sleep(200);
}
}catch(InterruptedException e){
e.printStackTrace();
}
}
}
public class Test {
public static void main(String[] args) {
try{
ThreadA a=new ThreadA();
ThreadB b=new ThreadB();
a.start();
b.start();
for(int i=0;i<10;i++){
Tools.tl.set("Main"+(i+1));
System.out.println("Main get Value="+Tools.tl.get());
Thread.sleep(200);
}
}catch(InterruptedException e){
e.printStackTrace();
}
}
}
结果：

。。。。。

解释：ThreadLocal类解决的是变量在不同线程间的隔离性，也就是不同线程拥有自己的值，不同的线程中的值是可以放入

ThreadLocal中进行保存的。

4.InheritableThreadLocal的使用
4.1值继承
使用这个类InheritableThreadLocal类可以让子线程从父线程中取得值。

public class InheritableThreadLocalExt extends InheritableThreadLocal<Long>{
@Override
protected Long initialValue() {
// TODO Auto-generated method stub
return new Date().getTime();
}
}

public class Tools {
public static InheritableThreadLocalExt tl=new InheritableThreadLocalExt();
}

public class ThreadA extends Thread{
@Override
public void run() {
try{
System.out.println("ThreadA get Value="+Tools.tl.get());
Thread.sleep(100);
}catch(InterruptedException e){
e.printStackTrace();
}
}
}

public class Test {
public static void main(String[] args) {
try{
System.out.println("在 Main线程中取值="+Tools.tl.get());
Thread.sleep(100);
Thread.sleep(5000);
ThreadA a=new ThreadA();
a.start();
}catch(InterruptedException e){
e.printStackTrace();
}
}
}
结果：
在 Main线程中取值=1474875161613
ThreadA get Value=1474875161613

4.2值继承再修改

public class InheritableThreadLocalExt extends InheritableThreadLocal<Long>{
@Override
protected Long initialValue() {
// TODO Auto-generated method stub
return new Date().getTime();
}
@Override
protected Long childValue(Long parentValue) {
// TODO Auto-generated method stub
return parentValue+100;//修改值
}
}
解释：如果子线程在获得值得同时，主线程将InheritableThreadLocal中的值进行更改，那么子线程仍取到的是旧值。
————————————————
版权声明：本文为CSDN博主「sunpy 孙培煜」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/ya_1249463314/article/details/52661833

Java多线程间通信

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读