Java多线程技术研究（二）-线程同步，通信及ThreadLocal

矫情吗；* 2022-06-14 11:45 219阅读 0赞

本篇博客主要介绍Java多线程之间的同步与通信，以及ThreadLocal。

**一、线程同步**  
在多线程环境中，可能会有两个甚至更多的线程试图同时访问一个有限的资源（代码，数据库等）。我们把多线程访问同一代码，产生不确定的结果，称为是**线程不安全的**，否则称之为**线程安全的**。对于String类就是线程安全的，而对于HashMap类是线程不安全的。  
下面看一段代码：

package com.wygu.multiThread.synchro;
    
    public class Bank { 
        int money = 100;
        public Bank(int money){
            this.money = money;
        }
    
        public int getLeftMoney(int withDraw){
            if(withDraw<0){
                return -1;
            }else if(this.money<0){
                return -2;
            }else if(this.money<withDraw){
                return -3;
            }else{
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                this.money = this.money-withDraw;
                System.out.println(Thread.currentThread().getName()+"取走金额："+withDraw+",银行剩余金额为："+this.money);
                return this.money;
            }
        }
    }
    package com.wygu.multiThread.synchro;
    
    public class ThreadSynchronise implements Runnable{ 
        Bank bank = null;
        public ThreadSynchronise(Bank bank){
            this.bank = bank;
        }
    
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName()+"start!");
            for(int i=0;i<10;i++){
                if(bank.getLeftMoney((i+1)*10)<0){
                    break;
                }
            }
            System.out.println(Thread.currentThread().getName()+"start!");  
        }
    
    }
    package com.wygu.multiThread.synchro;
    
    public class Main { 
    
        public static void main(String[] args) {
            Bank bank = new Bank(80);
            ThreadSynchronise runnableA = new ThreadSynchronise(bank);
            ThreadSynchronise runnableB = new ThreadSynchronise(bank);
            Thread threadA = new Thread(runnableA,"User A-->");
            Thread threadB = new Thread(runnableB,"User B-->");
            threadA.start();
            threadB.start();
        }
    
    }

程序运行结果为：  
User B–>start!  
User A–>start!  
User A–>取走金额：10,银行剩余金额为：70  
User B–>取走金额：10,银行剩余金额为：70  
User B–>取走金额：20,银行剩余金额为：30  
User A–>取走金额：20,银行剩余金额为：30  
User B–>取走金额：30,银行剩余金额为：-30  
User B–>start!  
User A–>取走金额：30,银行剩余金额为：-30  
User A–>start!

程序中定义了Bank类，线程UserA和线程UserB共享一个Bank的实例bank，两个线程各自去银行取钱，我们发现出现取了钱后，剩余的金额竟然相同。是什么原因导致出现上述不正常的运行结果呢，这是因为，多线程可以同时调用方法getLeftMoney()，每个线程跳过判断以后都会休息一会，无法保证先调用方法的线程先执行完，导致可能会出现脏数据的情形。换句话说，对于上述对象bank不是线程安全的。

如何编写线程安全的代码或者把Java提供的类变成线程安全的，我们可以通过加入锁的机制实现线程同步的安全。Java语言提供了volatile和synchronized两个关键字来保证线程之间操作的有序性，volatile 是因为其本身包含“禁止指令重排序”的语义，synchronized 是由“一个变量在同一个时刻只允许一条线程对其进行 lock 操作”这条规则获得的，此规则决定了持有同一个对象锁的两个同步块只能顺序执行。

**1、synchronized**

synchronized修饰的对象有以下几种：  
（1） 修饰一个代码块，被修饰的代码块称为同步语句块，其作用的范围是\{\}括起来的代码，作用的对象是调用这个代码块的对象；  
（2） 修饰一个方法，被修饰的方法称为同步方法，其作用的范围是整个方法，作用的对象是调用这个方法的对象；  
（3） 修饰一个静态的方法，其作用的范围是整个静态方法，作用的对象是调用这个类的所有对象；  
（4） 修饰一个类，其作用的范围是synchronized后面()括起来的部分，作用的对象是这个类的所有对象。

针对上述的代码出现线程不安全的情况，可以通过下述两种方法实现线程安全的，Bank类中的方法getLeftMoney()修改为：  
**方法一：**

//方法之前加入关键字 synchronized 修饰getLeftMoney()方法
        public synchronized int getLeftMoney(int withDraw){
            if(withDraw<0){
                return -1;
            }else if(this.money<0){
                return -2;
            }else if(this.money<withDraw){
                return -3;
            }else{
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                this.money = this.money-withDraw;
                System.out.println(Thread.currentThread().getName()+"取走金额："+withDraw+",银行剩余金额为："+this.money);
                return this.money;
            }
        }

**方法二：**

//利用synchronized 修饰整个代码块或者说修饰一个类
    public  int getLeftMoney(int withDraw){
            synchronized (this) {
                if(withDraw<0){
                    return -1;
                }else if(this.money<0){
                    return -2;
                }else if(this.money<withDraw){
                    return -3;
                }else{
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    this.money = this.money-withDraw;
                    System.out.println(Thread.currentThread().getName()+"取走金额："+withDraw+",银行剩余金额为："+this.money);
                    return this.money;
                }
            }   
        }

执行结果为：  
User A–>start!  
User B–>start!  
User A–>取走金额：10,银行剩余金额为：70  
User B–>取走金额：10,银行剩余金额为：60  
User A–>取走金额：20,银行剩余金额为：40  
User B–>取走金额：20,银行剩余金额为：20  
User A–>start!  
User B–>start!  
程序运行正常

**2、volatile**  
Java语言提供了一种相对于synchronized 稍弱的同步机制volatile，主要用于修饰变量，确保此变量对所有的线程均是可见的。volatile变量不会被缓存在线程的工作内存中，而是直接储存子主存中，从而保证了读取到的volatile类型的变量时总会是最新写入的。  
此外，在访问voliatile变量时不会执行加锁操作，因而不会是线程发生阻塞，进而相对于synchronized 不会出现死锁的情形（使用synchronized 修饰多个方法，多线程方法可能会出现死锁的情形）。  
多线程访问单例：

package com.wygu.thread.study;
    
    public class SingleInstanceBasic { 
        private static SingleInstanceBasic instance = null;
        //构造方法私有化，保证外部无法创建该类的实例
        private SingleInstanceBasic(){}
    
        public synchronized static SingleInstanceBasic getInstance(){
            if(null == instance){
                instance = new SingleInstanceBasic();
            }
            return instance;
        }
    }

上述方式中存在性能上的问题，因为只有第一次执行getInstance()时，才真正需要同步。换句话，第一次之后的每次调用该方法，同步都是一种累赘。我们可以利用volatile和synchronized 实现双重锁机制。

package com.wygu.thread.study;
    
    public class SingleInstanceImprove { 
        private volatile static SingleInstanceImprove instance= null;
        //构造方法私有化，保证外部无法创建该类的实例
        private SingleInstanceImprove(){}
    
        public static SingleInstanceImprove getInstance(){
            if(null == instance){
       //检查实例，如果不存在，就进入同步区块
                synchronized (SingleInstanceImprove.class) {
       //只有第一次执行时才会执行到此处
                    if(null == instance){
       //再检查一次
                        instance = new SingleInstanceImprove();
                    }
                }
            }
            return instance;
        }
    }

可以看到利用volatile修饰变量确保其对所有的线程都是可见的。那么有这样一个问题，static关键字修饰的变量，会单独存放在静态存储区中，而且保证只有一个副本。

**static和volatile有什么区别呢？**  
volatile修饰的变量保证了主存中保存的变量总会是最新写入的，但是static变量可能在线程工作内存中存在本地缓存的值，导致主存中的值不一定是最新的。因而可以使用volatile和static共同修饰变量，从而强制线程每次需要从主存中读取全局值，每次写入值时直接写入到主存中。

**注意：**使用volatile 修饰的变量不是线程安全的，只是保证了变量的可见性，无法保证多种的操作的原子性，比如一个线程从主存中读入该变量后，另一个线程发生了写的操作，从而导致出现了脏读。

**二、线程通信**

**1、共享内存机制**  
Java中共享内存是通过多线程同步机制实现的，比如共享Runnable的实例，具体事例如下：

package wygu.multiThread.study;
    
    public class MultiThreadShare implements Runnable{ 
    
        private volatile int breakFast=10;
    
        @Override
        public void run() {
            for(int i=0;breakFast>0;i++){
                System.out.println(Thread.currentThread().getName()+"---->"+breakFast--);
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
    
        }
    
        public static void main(String[] args) {
            MultiThreadShare mThreadShare = new MultiThreadShare();
            Thread threadG = new Thread(mThreadShare, "KFC窗口1");
            Thread threadH = new Thread(mThreadShare, "KFC窗口2");
            Thread threadI = new Thread(mThreadShare, "KFC窗口3");
            threadG.start();
            threadH.start();
            threadI.start();
        }
    
    }

程序运行结果为：  
KFC窗口1—->10  
KFC窗口3—->9  
KFC窗口1—->8  
KFC窗口3—->7  
KFC窗口1—->6  
KFC窗口3—->5  
KFC窗口2—->4  
KFC窗口1—->3  
KFC窗口2—->2  
KFC窗口3—->1

**2、wait()/notify()/notifyAll()机制**  
下面利用生产者/消费者模式示例说明如何通过wait()/notify()/notifyAll()机制。  
**生产者**

package wygu.multiThread.study;
    
    public class Producer extends Thread{ 
        private ShareResource shareResource = null;
        public Producer(ShareResource shareResource) {
            this.shareResource = shareResource;
        }
    
        @Override
        public void run(){
            for(int i=0;i<5;i++){
                shareResource.set(String.valueOf(i));
            }
        }
    }

**消费者**

package wygu.multiThread.study;
    
    public class Consumer extends Thread{ 
    
        private ShareResource shareResource = null;
        public Consumer(ShareResource shareResource) {
            this.shareResource = shareResource;
        }   
        @Override
        public void run(){
            for(int i=0;i<5;i++){
                shareResource.get();
            }
        }
    }

**共享资源池**

package wygu.multiThread.study;
    
    import java.util.LinkedList;
    
    public class ShareResource {
        private int maxSize = 3;
        private LinkedList<String> resCatchList = new LinkedList<String>();
    
        public void get(){
            synchronized (this) {
                while(resCatchList.isEmpty()){
                    try {
                        System.out.println(Thread.currentThread().getName()+":释放对象锁，CPU");
                        wait();
                        System.out.println(Thread.currentThread().getName()+":重新获得对象锁，CPU");
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
                System.out.println(Thread.currentThread().getName()+"获得资源---->"+resCatchList.poll());
                notify();
            }
    
        }
    
        public  void set(String resource){
            synchronized (this) {
                while(maxSize==resCatchList.size()){
                    try {
                        System.out.println(Thread.currentThread().getName()+":释放对象锁，释放CPU");
                        wait();
                        System.out.println(Thread.currentThread().getName()+":重新获得对象锁，CPU");
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
                System.out.println(Thread.currentThread().getName()+"放入新资源---->"+resource);
                resCatchList.push(resource);
                notify();
            }
        }
    }

**测试程序及运行结果**

package wygu.multiThread.study;
    
    public class Main {
        public static void main(String [] argv){
            ShareResource shareResource = new ShareResource();
            Producer producer = new Producer(shareResource);
            producer.setName("Producer");
            Consumer consumer = new Consumer(shareResource);
            consumer.setName("Consumer");
            producer.start();
            consumer.start();
        }
    
    }

Consumer:释放对象锁，CPU  
Producer放入新资源—->0  
Consumer:重新获得对象锁，CPU  
Consumer获得资源—->0  
Consumer:释放对象锁，CPU  
Producer放入新资源—->1  
Consumer:重新获得对象锁，CPU  
Consumer获得资源—->1  
Consumer:释放对象锁，CPU  
Producer放入新资源—->2  
Consumer:重新获得对象锁，CPU  
Consumer获得资源—->2  
Consumer:释放对象锁，CPU  
Producer放入新资源—->3  
Consumer:重新获得对象锁，CPU  
Consumer获得资源—->3  
Consumer:释放对象锁，CPU  
Producer放入新资源—->4  
Consumer:重新获得对象锁，CPU  
Consumer获得资源—->4

**3、管道流通信机制**

管道流过程：生产数据者*（生产者）向管道中输出数据，读数据者*（消费者）从管道中读取数据。此外，输入管道流和输出管道流之间通过方法connection()建立连接。具体事例如下：

package wygu.multiThread.Pipe;
    
    import java.io.IOException;
    import java.io.PipedInputStream;
    //消费者
    public class ReadThread extends Thread{ 
        private PipedInputStream pipedInput;
        public ReadThread(PipedInputStream pipedInput) {
            this.pipedInput = pipedInput;
        }
        @Override
        public void run(){
            System.out.println(Thread.currentThread().getName()+"-->当前管道没有数据，阻塞中...");  
            byte[] buf = new byte[1024];  
            int len;
            try {
                len = pipedInput.read(buf);
                 System.out.println(Thread.currentThread().getName()+"-->读取管道中的数据："+new String(buf,0,len));    
                 pipedInput.close(); 
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }          
        }
    }
    
    //生产者
    package wygu.multiThread.Pipe;
    
    import java.io.IOException;
    import java.io.PipedOutputStream;
    public class WriteThread extends Thread{ 
        private PipedOutputStream pipedOutput;
        public WriteThread(PipedOutputStream pipedOutput) {
            this.pipedOutput = pipedOutput;
        }
        @Override
        public void run(){
            System.out.println(Thread.currentThread().getName()+"-->开始将数据写入管道中...");   
            try {
                Thread.sleep(5000);
                pipedOutput.write(new String("Hello World !!").getBytes());
                pipedOutput.close(); 
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    
    }
    
    //主程序
    package wygu.multiThread.Pipe;
    
    import java.io.IOException;
    import java.io.PipedInputStream;
    import java.io.PipedOutputStream;
    
    public class Main { 
    
        public static void main(String[] args) {
            PipedInputStream pInputStream = new PipedInputStream();
            PipedOutputStream pOutputStream = new PipedOutputStream();
            try {
                pInputStream.connect(pOutputStream); //连接管道的输入流和输出流
            } catch (IOException e) {
                e.printStackTrace();
            }
            ReadThread readThread = new ReadThread(pInputStream);
            WriteThread writeThread = new WriteThread(pOutputStream);
            readThread.setName("ReadThread");
            writeThread.setName("writeThread");
            readThread.start();
            writeThread.start();
        }
    }

程序运行结果为：  
ReadThread–>当前管道没有数据，阻塞中…  
writeThread–>开始将数据写入管道中…  
ReadThread–>读取管道中的数据：Hello World !!

**三、ThreadLocal（线程本地变量）**

在Web开发过程中，每个外部请求到服务器后，服务器会创建一个Thread去处理该请求。在处理请求的过程中，可能会出现许多报错信息，希望在线程执行快结束时再打印出来。一种简单的方式就是定义一个public static变量或者一个单例，然而这种方式无法解决多线程并发问题。那么是否存在一种本地变量，它的生命周期和线程的生命周期是一样的呢。早在JDK 1.2的版本中就提供Java.lang.ThreadLocal，ThreadLocal为解决多线程程序的并发问题提供了一种新的思路，并且在JDK 1.5后提供了对泛型的支持。

ThreadLocal很容易让人望文生义，认为是一种Thread（本地线程）。实际上ThreadLocal是Thread的的一个变量，称它为ThreadLocalVariable可能更合适一点。  
**1、 每个线程都有自己的局部变量**  
每个线程都有一个独立于其他线程的上下文来保存这个变量，一个线程的本地变量对其他线程是不可见的。当线程结束后，对应该线程的局部变量将自动被垃圾回收。因而显示的调用remove()方法不是必须的操作，当然显示调用可以加快内存的清理速度。  
**2、独立于变量的初始化副本**  
ThreadLocal可以给一个初始值，而每个线程都会获得这个初始化值的一个副本，这样才能保证不同的线程都有一份拷贝。  
**3、状态与某一个线程相关联**  
ThreadLocal不是用于解决共享变量的问题的，不是为了协调线程同步而存在，而是为了方便每个线程处理自己的状态而引入的一个机制。  
下面是处理多线程打印报错信息的部分代码：

package com.wygu.threadlocal;
    
    import java.util.ArrayList;
    
    public class ErrorInfoUtil {
    
        public static class ThreadShareList {
            private static final ThreadLocal<ArrayList<String>> threadLocal = new ThreadLocal<ArrayList<String>>();  
            public static ArrayList<String> getCurrList() {  
                // 获取当前线程内共享的arrayList 
                ArrayList<String> arrayList = threadLocal.get();  
                if(null==arrayList) {  
                    arrayList = new ArrayList<String>();
                    threadLocal.set(arrayList);  
                }       
                return arrayList;  
            }  
        }
    
        //将详细的报错信息填充线程本地变量中
        public static void fillErrorInfo(String errorReason){
            String respStringInfo = null;
            StackTraceElement[] stack = (new Throwable()).getStackTrace();  
            if(null!=stack && 1<stack.length){
                respStringInfo = errorReason+" | "+stack[1].getClassName()+" | "+stack[1].getMethodName()+
                        " | "+stack[1].getLineNumber();
            }       
            ThreadShareList.getCurrList().add(respStringInfo);
        }
    
        //线程即将执行结束之前，打印所有的错误信息
        public static void dumpTraceError(){
            System.out.println("Error Information Begin...");
            ArrayList<String> arrayList = ThreadShareList.getCurrList();
            int errInfoIndex=1;
            for(String string : arrayList){
                System.out.println("["+(errInfoIndex++)+"]: "+string);
            }
            System.out.println("Error Information End...");
            ThreadShareList.getCurrList().clear();
        }   
    }

**贵在精**