Java线程间的共享和协作，详细过程如下文，一眼懂

逃离我推掉我的手 2024-05-06 21:32 5阅读 0赞

## 线程的共享 ##

### synchronized内置锁 ###

Java 支持多个线程同时访问一个对象或者对象的成员变量，关键字synchronized 可以修饰方法或者以同步块的形式来进行使用，它主要确保多个线程在同一个时刻，只能有一个线程处于方法或者同步块中，它保证了线程对变量访问的可见性和排他性，又称为内置锁机制。 **对象锁和类锁：** 对象锁是用于对象实例方法，或者一个对象实例上的，类锁是用于类的静态方法或者一个类的 class 对象上的。我们知道，类的对象实例可以有很多个，但是每个类只有一个 class 对象，所以不同对象实例的对象锁是互不干扰的，但是每个类只有一个类锁。 但是有一点必须注意的是，其实类锁只是一个概念上的东西，并不是真实存在的，类锁其实锁的是每个类的对应的 class 对象。类锁和对象锁之间也是互不干扰的。

代码示例：

*类说明：synchronized关键字的使用方法
     */
    public class SynTest {
    
    	private long count =0;
    	private Object obj = new Object();//作为一个锁
    
    	public long getCount() {
    		return count;
    	}
    
    	public void setCount(long count) {
    		this.count = count;
    	}
    
    	/*用在同步块上*/
    	public void incCount(){
    		synchronized (obj){
    			count++;
    		}
    	}
    
    	/*用在方法上*/
    	public synchronized void incCount2(){
    			count++;
    	}
    
    	/*用在同步块上，但是锁的是当前类的对象实例*/
    	public void incCount3(){
    		synchronized (this){
    			count++;
    		}
    	}
    
    	//线程
    	private static class Count extends Thread{
    
    		private SynTest simplOper;
    
    		public Count(SynTest simplOper) {
    			this.simplOper = simplOper;
    		}
    
    		@Override
    		public void run() {
    			for(int i=0;i<10000;i++){
    				simplOper.incCount();//count = count+10000
    			}
    		}
    	}
    
    	public static void main(String[] args) throws InterruptedException {
    		SynTest simplOper = new SynTest();
    		//启动两个线程
    		Count count1 = new Count(simplOper);
    		Count count2 = new Count(simplOper);
    		count1.start();
    		count2.start();
    		Thread.sleep(50);
    		System.out.println(simplOper.count);//20000
    	}
    }

/**
     *类说明：锁的实例不一样，也是可以并行的
     */
    public class DiffInstance {
    	
        private static class InstanceSyn implements Runnable{
            private DiffInstance diffInstance;
    
            public InstanceSyn(DiffInstance diffInstance) {
                this.diffInstance = diffInstance;
            }
    
            @Override
            public void run() {
                System.out.println("TestInstance is running..."+ diffInstance);
                diffInstance.instance();
            }
        }
    
        private static class Instance2Syn implements Runnable{
            private DiffInstance diffInstance;
    
            public Instance2Syn(DiffInstance diffInstance) {
                this.diffInstance = diffInstance;
            }
            @Override
            public void run() {
                System.out.println("TestInstance2 is running..."+ diffInstance);
                diffInstance.instance2();
            }
        }
    
        private synchronized void instance(){
            SleepTools.second(3);
            System.out.println("synInstance is going..."+this.toString());
            SleepTools.second(3);
            System.out.println("synInstance ended "+this.toString());
        }
    
        private synchronized void instance2(){
            SleepTools.second(3);
            System.out.println("synInstance2 is going..."+this.toString());
            SleepTools.second(3);
            System.out.println("synInstance2 ended "+this.toString());
        }
    
        public static void main(String[] args) {
            DiffInstance instance1 = new DiffInstance();
            Thread t3 = new Thread(new Instance2Syn(instance1));
            DiffInstance instance2 = new DiffInstance();
            Thread t4 = new Thread(new InstanceSyn(instance1));
            //先执行完一个才会执行另外一个
            t3.start();
            t4.start();
            SleepTools.second(1);
        }
    }

![format_png][]

/**
     *类说明：演示实例锁和类锁是不同的，两者可以并行
     */
    public class InstanceAndClass {
    	
        private static class SynClass extends Thread{
            @Override
            public void run() {
                System.out.println("TestClass is running...");
                synClass();
            }
        }
    
        private static class InstanceSyn implements Runnable{
            private InstanceAndClass SynClassAndInstance;
    
            public InstanceSyn(InstanceAndClass SynClassAndInstance) {
                this.SynClassAndInstance = SynClassAndInstance;
            }
    
            @Override
            public void run() {
                System.out.println("TestInstance is running..."+SynClassAndInstance);
                SynClassAndInstance.instance();
            }
        }
    
        private synchronized void instance(){
            SleepTools.second(1);
            System.out.println("synInstance is going..."+this.toString());
            SleepTools.second(1);
            System.out.println("synInstance ended "+this.toString());
        }
    
        private static synchronized void synClass(){
            SleepTools.second(1);
            System.out.println("synClass going...");
            SleepTools.second(1);
            System.out.println("synClass end");
        }
    
        public static void main(String[] args) {
            InstanceAndClass synClassAndInstance = new InstanceAndClass();
            Thread t1 = new SynClass();
            Thread t2 = new Thread(new InstanceSyn(synClassAndInstance));
            t2.start();
            SleepTools.second(1);
            t1.start();
        }
    }

![format_png 1][]

/**
     *类说明：类锁和锁static变量也是不同的 可以并行 
     */
    public class StaticAndClass {
    	
        private static class SynClass extends Thread{
            @Override
            public void run() {
                System.out.println(currentThread().getName()
                        +":SynClass is running...");
                synClass();
            }
        }
    
        private static class SynStatic extends Thread{
            @Override
            public void run() {
                System.out.println(currentThread().getName()
                        +"SynStatic is running...");
                synStatic();
            }
        }
    
        private static synchronized void synClass(){
            System.out.println(Thread.currentThread().getName()
                    +"synClass going...");
            SleepTools.second(1);
            System.out.println(Thread.currentThread().getName()
                    +"synClass end");
        }
    
        private static Object obj = new Object();
        private static void synStatic(){
            synchronized (obj){
                System.out.println(Thread.currentThread().getName()
                        +"synStatic going...");
                SleepTools.second(1);
                System.out.println(Thread.currentThread().getName()
                        +"synStatic end");
            }
        }
    
        public static void main(String[] args) {
            StaticAndClass synClassAndInstance = new StaticAndClass();
            Thread t1 = new SynClass();
            //Thread t2 = new SynStatic();
            Thread t2 = new SynClass();
            t2.start();
            SleepTools.second(1);
            t1.start();
        }
    }

![format_png 2][]

### 错误的加锁和原因分析 ###

原因：虽然我们对 i 进行了加锁，但是

![format_png 3][]

但是当我们反编译这个类的 class 文件后，可以看到 i++实际是， 本质上是返回了一个新的 Integer 对象。也就是每个线程实际加锁的是不同的 Integer 对象。

### volatile，最轻量的同步机制 ###

volatile 保证了不同线程对这个变量进行操作时的可见性，即一个线程修改了某个变量的值，这新值对其他线程来说是立即可见的。

![format_png 4][]

不加 volatile 时，子线程无法感知主线程修改了 ready 的值，从而不会退出循环，而加了 volatile 后，子线程可以感知主线程修改了 ready 的值，迅速退出循环。

/**
     * 类说明：演示Volatile的提供的可见性
     */
    public class VolatileCase {
        private volatile static boolean ready;
        private static int number;
    
        //
        private static class PrintThread extends Thread{
            @Override
            public void run() {
                System.out.println("PrintThread is running.......");
                while(!ready);//无限循环
                System.out.println("number = "+number);
            }
        }
    
        public static void main(String[] args) {
            new PrintThread().start();
            SleepTools.second(1);
            number = 51;//如果没有加volatile关键字则主线程都结束了也没有打印number的值，加了关键值后打印出来的值就是主线程修改的值
            ready = true;
            SleepTools.second(5);
            System.out.println("main is ended!");
        }
    }

但是 volatile 不能保证数据在多个线程下同时写时的线程安全。

/**
     * 类说明：
     */
    public class NotSafe {
        private volatile long count =0;
    
        public long getCount() {
            return count;
        }
    
        public void setCount(long count) {
            this.count = count;
        }
    
        //count进行累加
        public void incCount(){
            count++;
        }
    
        //线程
        private static class Count extends Thread{
    
            private NotSafe simplOper;
    
            public Count(NotSafe simplOper) {
                this.simplOper = simplOper;
            }
    
            @Override
            public void run() {
                for(int i=0;i<10000;i++){
                    simplOper.incCount();
                }
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            NotSafe simplOper = new NotSafe();
            //启动两个线程
            Count count1 = new Count(simplOper);
            Count count2 = new Count(simplOper);
            count1.start();
            count2.start();
            Thread.sleep(50);
            System.out.println(simplOper.count);//20000?
        }
    }

volatile 最适用的场景：一个线程写，多个线程读。

## 线程间的协作 ##

线程之间相互配合，完成某项工作，比如：一个线程修改了一个对象的值，而另一个线程感知到了变化，然后进行相应的操作，整个过程开始于一个线程，而最终执行又是另一个线程。前者是生产者，后者就是消费者，这种模式隔离了“做什么”（what）和“怎么做”（How），简单的办法是让消费者线程不断地循环检查变量是否符合预期在 while 循环中设置不满足的条件，如果条件满足则退出 while 循环，从而完成消费者的工作。却存在如下问题： 1） 难以确保及时性。 2）难以降低开销。如果降低睡眠的时间，比如休眠 1 毫秒，这样消费者能更加迅速地发现条件变化，但是却可能消耗更多的处理器资源，造成了无端的浪费。

### 等待/通知机制 ###

是指一个线程 A 调用了对象 O 的 wait()方法进入等待状态，而另一个线程 B调用了对象 O 的 notify()或者 notifyAll()方法，线程 A 收到通知后从对象 O 的 wait()方法返回，进而执行后续操作。上述两个线程通过对象 O 来完成交互，而对象上的 wait()和notify/notifyAll()的关系就如同开关信号一样，用来完成等待方和通知方之间的交互工作。

**notify()：** 通知一个在对象上等待的线程,使其从 wait 方法返回,而返回的前提是该线程获取到了对象的锁，没有获得锁的线程重新进入 WAITING 状态。

**notifyAll()：** 通知所有等待在该对象上的线程

**wait()** 调用该方法的线程进入 WAITING 状态,只有等待另外线程的通知或被中断才会返回.需要注意,调用 wait()方法后,会释放对象的锁

**wait(long)** 超时等待一段时间,这里的参数时间是毫秒,也就是等待长达 n 毫秒,如果没有通知就超时返回

**wait (long,int)** 对于超时时间更细粒度的控制,可以达到纳秒

#### **等待和通知的标准范式** ####

等待方遵循如下原则。

1）获取对象的锁。

2）如果条件不满足，那么调用对象的 wait()方法，被通知后仍要检查条件。

3）条件满足则执行对应的逻辑。

![format_png 5][]

通知方遵循如下原则。

1）获得对象的锁。

2）改变条件。

3）通知所有等待在对象上的线程。

![format_png 6][]

**在调用 wait（）、notify()系列方法之前，线程必须要获得该对象的对象级别锁，即只能在同步方法或同步块中调用 wait（）方法、notify()系列方法**，进入 wait（）方法后，当前线程释放锁，在从 wait（）返回前，线程与其他线程竞争重新获得锁，执行 notify()系列方法的线程退出调用了 notifyAll 的 synchronized代码块的时候后，他们就会去竞争。如果其中一个线程获得了该对象锁，它就会继续往下执行，在它退出 synchronized 代码块，释放锁后，其他的已经被唤醒的线程将会继续竞争获取该锁，一直进行下去，直到所有被唤醒的线程都执行完毕。

#### **notify 和 notifyAll 应该用谁** ####

尽可能用 notifyall()，谨慎使用 notify()，因为 notify()只会唤醒一个线程，我们无法确保被唤醒的这个线程一定就是我们需要唤醒的线程

代码示例：

/**
     *类说明：快递实体类
     */
    public class Express {
        public final static String CITY = "ShangHai";
        private int km;/*快递运输里程数*/
        private String site;/*快递到达地点*/
    
        public Express() {
        }
    
        public Express(int km, String site) {
            this.km = km;
            this.site = site;
        }
    
        /* 变化公里数，然后通知处于wait状态并需要处理公里数的线程进行业务处理*/
        public synchronized void changeKm(){
            this.km = 101;
            notify();
        }
    
        /* 变化地点，然后通知处于wait状态并需要处理地点的线程进行业务处理*/
        public  synchronized  void changeSite(){
            this.site = "BeiJing";
            notifyAll();
        }
    
        /*线程等待公里的变化*/
        public synchronized void waitKm(){
            while(this.km<100){
                try {
                    wait();
                    System.out.println("Check Site thread["
                                    +Thread.currentThread().getId()
                            +"] is be notified");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("the Km is "+this.km+",I will change db");
        }
    
        /*线程等待目的地的变化*/
        public synchronized void waitSite(){
            while(this.site.equals(CITY)){//快递到达目的地
                try {
                    wait();
                    System.out.println("Check Site thread["+Thread.currentThread().getId()
                    		+"] is be notified");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("the site is "+this.site+",I will call user");
        }
    }

/**
     *类说明：测试wait/notify/notifyAll
     */
    public class TestWN {
        private static Express express = new Express(0,Express.CITY);
    
        /*检查里程数变化的线程,不满足条件，线程一直等待*/
        private static class CheckKm extends Thread{
            @Override
            public void run() {
            	express.waitKm();
            }
        }
    
        /*检查地点变化的线程,不满足条件，线程一直等待*/
        private static class CheckSite extends Thread{
            @Override
            public void run() {
            	express.waitSite();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            for(int i=0;i<3;i++){
                new CheckSite().start();
            }
            for(int i=0;i<3;i++){
                new CheckKm().start();
            }
    
            Thread.sleep(1000);
            express.changeKm();//快递地点变化
        }
    }

![format_png 7][]

#### 等待超时模式实现一个连接池 ####

调用场景：调用一个方法时等待一段时间（一般来说是给定一个时间段），如果该方法能够在给定的时间段之内得到结果，那么将结果立刻返回，反之，超时返回默认结果 假设等待时间段是 T，那么可以推断出在当前时间 now+T 之后就会超时 等待持续时间：REMAINING=T。 超时时间：FUTURE=now+T。

/**
     *类说明：连接池的实现
     */
    public class DBPool {
    
        /*容器，存放连接*/
        private static LinkedList<Connection> pool = new LinkedList<Connection>();
    
        /*限制了池的大小=20*/
        public DBPool(int initialSize) {
            if (initialSize > 0) {
                for (int i = 0; i < initialSize; i++) {
                    pool.addLast(SqlConnectImpl.fetchConnection());
                }
            }
        }
    
        /*释放连接,通知其他的等待连接的线程*/
        public void releaseConnection(Connection connection) {
            if (connection != null) {
                synchronized (pool){
                    pool.addLast(connection);
                    //通知其他等待连接的线程
                    pool.notifyAll();
                }
            }
        }
    
        /*获取*/
        // 在mills内无法获取到连接，将会返回null 1S
        public Connection fetchConnection(long mills)
                throws InterruptedException {
            synchronized (pool){
                //永不超时
                if(mills<=0){
                    while(pool.isEmpty()){
                        pool.wait();
                    }
                    return pool.removeFirst();
                }else{
                    /*超时时刻*/
                    long future = System.currentTimeMillis()+mills;
                    /*等待时长*/
                    long remaining = mills;
                    while(pool.isEmpty()&&remaining>0){
                        pool.wait(remaining);
                        /*唤醒一次，重新计算等待时长*/
                        remaining = future-System.currentTimeMillis();
                    }
                    Connection connection = null;
                    if(!pool.isEmpty()){
                        connection = pool.removeFirst();
                    }
                    return connection;
                }
            }
    
        }
    }

/**
     *类说明：
     */
    public class DBPoolTest {
        static DBPool pool  = new DBPool(10);
        // 控制器:控制main线程将会等待所有Woker结束后才能继续执行
        static CountDownLatch end;
    
        public static void main(String[] args) throws Exception {
        	// 线程数量
            int threadCount = 50;
            end = new CountDownLatch(threadCount);
            int count = 20;//每个线程的操作次数
            AtomicInteger got = new AtomicInteger();//计数器：统计可以拿到连接的线程
            AtomicInteger notGot = new AtomicInteger();//计数器：统计没有拿到连接的线程
            for (int i = 0; i < threadCount; i++) {
                Thread thread = new Thread(new Worker(count, got, notGot), 
                		"worker_"+i);
                thread.start();
            }
            end.await();// main线程在此处等待
            System.out.println("总共尝试了: " + (threadCount * count));
            System.out.println("拿到连接的次数：  " + got);
            System.out.println("没能连接的次数： " + notGot);
        }
    
        static class Worker implements Runnable {
            int           count;
            AtomicInteger got;
            AtomicInteger notGot;
    
            public Worker(int count, AtomicInteger got,
                                   AtomicInteger notGot) {
                this.count = count;
                this.got = got;
                this.notGot = notGot;
            }
    
            public void run() {
                while (count > 0) {
                    try {
                        // 从线程池中获取连接，如果1000ms内无法获取到，将会返回null
                        // 分别统计连接获取的数量got和未获取到的数量notGot
                        Connection connection = pool.fetchConnection(1000);
                        if (connection != null) {
                            try {
                                connection.createStatement();
    //                            PreparedStatement preparedStatement
    //                                    = connection.prepareStatement("");
    //                            preparedStatement.execute();
                                connection.commit();
                            } finally {
                                pool.releaseConnection(connection);
                                got.incrementAndGet();
                            }
                        } else {
                            notGot.incrementAndGet();
                            System.out.println(Thread.currentThread().getName()
                            		+"等待超时!");
                        }
                    } catch (Exception ex) {
                    } finally {
                        count--;
                    }
                }
                end.countDown();
            }
        }
    }

/**
     *类说明：
     */
    public class SqlConnectImpl implements Connection{
    	
    	/*拿一个数据库连接*/
        public static final Connection fetchConnection(){
            return new SqlConnectImpl();
        }
        .........

![format_png 8][]

客户端获取连接的过程被设定为等待超时的模式，也就是在 1000 毫秒内如果无法获取到可用连接，将会返回给客户端一个 null。设定连接池的大小为 10个，然后通过调节客户端的线程数来模拟无法获取连接的场景。 它通过构造函数初始化连接的最大上限，通过一个双向队列来维护连接，调用方需要先调用 fetchConnection(long)方法来指定在多少毫秒内超时获取连接，当连接使用完成后，需要调用 releaseConnection(Connection)方法将连接放回线程池

[format_png]: https://img-blog.csdnimg.cn/img_convert/af854734f344fcef5ca4b364fc029b36.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/a0bb6dc0000440283dbfc26fdc43c3bd.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/49da3c5e6cd317e7935a624577a4b0ed.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/29295f5420dce86c5fc3fadce59be1c1.webp?x-oss-process=image/format,png
[format_png 4]: https://img-blog.csdnimg.cn/img_convert/b636d94eafd6f900fef042b6cc76b00f.webp?x-oss-process=image/format,png
[format_png 5]: https://img-blog.csdnimg.cn/img_convert/61309301d2c607e59300c5bbe5a3105b.webp?x-oss-process=image/format,png
[format_png 6]: https://img-blog.csdnimg.cn/img_convert/ee91d8d62b125bdcaa2082440066fa21.webp?x-oss-process=image/format,png
[format_png 7]: https://img-blog.csdnimg.cn/img_convert/9a99864cd3add9ecd3747b6ea4606bf1.webp?x-oss-process=image/format,png
[format_png 8]: https://img-blog.csdnimg.cn/img_convert/a5b70c42068e9046aaaf1a359be80154.webp?x-oss-process=image/format,png