java高并发编程(三)

时间:2022-04-25 15:43:56
java高并发主要有三块知识点:
synchronizer:同步器,在多个线程之间互相之间怎么进行通讯,同步等;
同步容器:jdk提供了同步性的容器,比如concurrentMap,concurrentList,BlockQueen等;
ThreadPool:线程池,executor,java在前两个的基础之上提供的线程池,很多实际中的问题可以着手解决了。
 
一、ReentrantLock              
jdk里面提供了一个新的锁,是手工锁,它是用来替代synchronized的,叫ReentrantLock,重入锁,其实synchronized也是可重入的,但是这把锁是和synchronized是有区别的,ReentrantLock是用新的同步方法写的时候经常用的一个工具;
复习之前讲的synchronized同步:
java高并发编程(三)java高并发编程(三)
/**
 * reentrantlock用于替代synchronized
 * 本例中由于m1锁定this,只有m1执行完毕的时候,m2才能执行
 * 这里是复习synchronized最原始的语义
 * @author mashibing
 */
package yxxy.c_020;

import java.util.concurrent.TimeUnit;

public class ReentrantLock1 {
    synchronized void m1() {
        for(int i=0; i<10; i++) {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(i);
        }
        
    }
    
    synchronized void m2() {
        System.out.println("m2 ...");
    }
    
    public static void main(String[] args) {
        ReentrantLock1 rl = new ReentrantLock1();
        new Thread(rl::m1).start();
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        new Thread(rl::m2).start();
    }
}
View Code

 

二、使用ReentrantLock完成同样功能      

/**
 * reentrantlock用于替代synchronized
 * 使用reentrantlock可以完成同样的功能
 * 需要注意的是,必须要必须要必须要手动释放锁(重要的事情说三遍)
 * 使用syn锁定的话如果遇到异常,jvm会自动释放锁,但是lock必须手动释放锁,因此经常在finally中进行锁的释放
 * @author mashibing
 */
package yxxy.c_020;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ReentrantLock2 {
    Lock lock = new ReentrantLock();

    void m1() {
        try {
            lock.lock(); //synchronized(this)
            for (int i = 0; i < 10; i++) {
                TimeUnit.SECONDS.sleep(1);

                System.out.println(i);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    void m2() {
        lock.lock();
        System.out.println("m2 ...");
        lock.unlock();
    }

    public static void main(String[] args) {
        ReentrantLock2 rl = new ReentrantLock2();
        new Thread(rl::m1).start();
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        new Thread(rl::m2).start();
    }
}

三、RenntrantLock的tryLock:          

java高并发编程(三)java高并发编程(三)
/**
 * 使用reentrantlock可以进行“尝试锁定”tryLock,这样无法锁定,或者在指定时间内无法锁定,线程可以决定是否继续等待
 * @author mashibing
 */
package yxxy.c_020;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ReentrantLock3 {
    Lock lock = new ReentrantLock();

    void m1() {
        try {
            lock.lock();
            for (int i = 0; i < 10; i++) {
                TimeUnit.SECONDS.sleep(1);

                System.out.println(i);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 使用tryLock进行尝试锁定,不管锁定与否,方法都将继续执行
     * 可以根据tryLock的返回值来判定是否锁定
     * 也可以指定tryLock的时间,由于tryLock(time)抛出异常,所以要注意unclock的处理,必须放到finally中
     */
    void m2() {
        /*
        boolean locked = lock.tryLock();
        System.out.println("m2 ..." + locked);
        if(locked) lock.unlock();
        */
        
        boolean locked = false;
        
        try {
            locked = lock.tryLock(5, TimeUnit.SECONDS);
            System.out.println("m2 ..." + locked);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            if(locked) lock.unlock();
        }
        
    }

    public static void main(String[] args) {
        ReentrantLock3 rl = new ReentrantLock3();
        new Thread(rl::m1).start();
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        new Thread(rl::m2).start();
    }
}
View Code

console:

java高并发编程(三)java高并发编程(三)
0
1
2
3
4
5
m2 ...false
6
7
8
9
View Code

 

四、ReentrantLock的lockInterruptibly方法:    

/**
 * 使用ReentrantLock还可以调用lockInterruptibly方法,可以对线程interrupt方法做出响应,
 * 在一个线程等待锁的过程中,可以被打断
 * 
 * @author mashibing
 */
package yxxy.c_020;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;

public class ReentrantLock4 {
        
    public static void main(String[] args) {
        Lock lock = new ReentrantLock();
        
        
        Thread t1 = new Thread(()->{
            try {
                lock.lock();
                System.out.println("t1 start");
                TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
                System.out.println("t1 end");
            } catch (InterruptedException e) {
                System.out.println("interrupted!");
            } finally {
                lock.unlock();
            }
        });
        t1.start();
        
        Thread t2 = new Thread(()->{
            try {
                //lock.lock();
                lock.lockInterruptibly(); //可以对interrupt()方法做出响应
                System.out.println("t2 start");
            } catch (InterruptedException e) {
                System.out.println("interrupted!");
            } finally {
                lock.unlock();
            }
        });
        t2.start();
        
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        t2.interrupt(); //打断线程2的等待
        
    }
}

console:

t1 start
interrupted!
Exception in thread "Thread-1" java.lang.IllegalMonitorStateException
    at java.util.concurrent.locks.ReentrantLock$Sync.tryRelease(ReentrantLock.java:151)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.release(AbstractQueuedSynchronizer.java:1261)
    at java.util.concurrent.locks.ReentrantLock.unlock(ReentrantLock.java:457)
    at yxxy.c_020.ReentrantLock4.lambda$1(ReentrantLock4.java:42)
    at java.lang.Thread.run(Thread.java:745)

t1线程牢牢的拿到锁之后,一直sleep不会释放,如果t2线程中的run方法使用lock.lock(),那么t2线程就会一直傻傻的等着这把锁,不能被其他线程打断;

而使用lockInterruptibly()方法是可以被打断的,主线程main调用t2.interrupt()来打断t2,告诉他是不会拿到这把锁的,别等了;

报错是因为lock.unlock()这个方法报错的,因为都没有拿到锁,无法unlock();是代码的问题,应该判断有锁,已经锁定的情况下才lock.unlock();

 

五、ReentrantLock还可以指定为公平锁            

默认的synchronized全都是不公平锁;

什么是公平锁,什么是不公平锁,假设很多个线程访问同一份资源的时候都要锁定,其中某一个线程A如果拿到了,他占有这把锁之后,其他人是都访问不了的,都得在那等着;什么时候一旦线程A释放了这把锁,那么剩下的线程中哪个线程得到这把锁,这事说不定,这个要看线程调度器自己去选哪个了,所以这叫竞争锁;也就是说等待的线程里面是没有公平性可言的;不过这种效率比较高,线程调度器不用计算到底哪个线程等的时间更长,所以默认的synchronized是非公平锁;
 
公平锁就是,谁等的时间长让谁得到那把锁。
 
/**
 * ReentrantLock还可以指定为公平锁
 * 
 * @author mashibing
 */
package yxxy.c_020;

import java.util.concurrent.locks.ReentrantLock;

public class ReentrantLock5 extends Thread {
        
    private static ReentrantLock lock = new ReentrantLock(); //参数为true表示为公平锁,请对比输出结果
    
    public void run() {
        for(int i=0; i<100; i++) {
            lock.lock();
            try{
                System.out.println(Thread.currentThread().getName()+"获得锁");
            }finally{
                lock.unlock();
            }
        }
    }
    public static void main(String[] args) {
        ReentrantLock5 rl=new ReentrantLock5();
        Thread th1=new Thread(rl);
        Thread th2=new Thread(rl);
        th1.start();
        th2.start();
    }
}

 

 

六、面试题:生产者消费者程序:          

要求:写一个固定容量同步容器,拥有put和get方法,以及getCount方法,能够支持2个生产者线程以及10个消费者线程的阻塞调用

同步容器:多个线程共同访问的时候,不能出问题,就是要加锁了,下面这个是阻塞式的同步容器;
代码:
java高并发编程(三)java高并发编程(三)
/**
 * 面试题:写一个固定容量同步容器,拥有put和get方法,以及getCount方法,
 * 能够支持2个生产者线程以及10个消费者线程的阻塞调用
 * 
 * 使用wait和notify/notifyAll来实现
 * 
 * @author mashibing
 */
package yxxy.c_021;

import java.util.LinkedList;
import java.util.concurrent.TimeUnit;

public class MyContainer1<T> {
    final private LinkedList<T> lists = new LinkedList<>();
    final private int MAX = 10; //最多10个元素
    private int count = 0;
    
    
    public synchronized void put(T t) {
        while(lists.size() == MAX) { //想想为什么用while而不是用if?
            try {
                this.wait(); //effective java
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        
        lists.add(t);
        ++count;
        this.notifyAll(); //通知消费者线程进行消费
    }
    
    public synchronized T get() {
        T t = null;
        while(lists.size() == 0) {
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        t = lists.removeFirst();
        count --;
        this.notifyAll(); //通知生产者进行生产
        return t;
    }
    
    public static void main(String[] args) {
        MyContainer1<String> c = new MyContainer1<>();
        //启动消费者线程
        for(int i=0; i<10; i++) {
            new Thread(()->{
                for(int j=0; j<5; j++) System.out.println(c.get());
            }, "c" + i).start();
        }
        
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        //启动生产者线程
        for(int i=0; i<2; i++) {
            new Thread(()->{
                for(int j=0; j<25; j++) c.put(Thread.currentThread().getName() + " " + j);
            }, "p" + i).start();
        }
    }
}
View Code

1.为什么用while而不用if?

假设容器中已经满了,如果用的是if,这个线程A发现list.size()==max已经满了,就this.wait()住了;
如果容器中被拿走了元素,线程A被叫醒了,它会从this.wait()开始继续往下运行,准备执行lists.add(),可是它被叫醒了之后还没有往里扔的时候,另外一个线程往list里面扔了一个,线程A拿到锁之后不再进行if判断,而是继续执行lists.add()就会出问题了;
如果用while,this.wait()继续往下执行的时候需要在while中再检查一遍,就不会出问题;
 
2.put()方法中为什么使用notifyAll而不是notify?
如果使用notify,notify是叫醒一个线程,那么就有可能叫醒的一个线程又是生产者,整个程序可能不动了,都wait住了;
 
七、使用Lock和Condition                      
使用wati和notify写线程程序的时候就像使用汇编语言一样,写起来会比较费劲;那么还可以用什么来写呢?
java高并发编程(三)java高并发编程(三)
/**
 * 面试题:写一个固定容量同步容器,拥有put和get方法,以及getCount方法,
 * 能够支持2个生产者线程以及10个消费者线程的阻塞调用
 * 
 * 使用Lock和Condition来实现
 * 对比两种方式,Condition的方式可以更加精确的指定哪些线程被唤醒
 * 
 * @author mashibing
 */
package yxxy.c_021;

import java.util.LinkedList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class MyContainer2<T> {
    final private LinkedList<T> lists = new LinkedList<>();
    final private int MAX = 10; //最多10个元素
    private int count = 0;
    
    private Lock lock = new ReentrantLock();
    private Condition producer = lock.newCondition();
    private Condition consumer = lock.newCondition();
    
    public void put(T t) {
        try {
            lock.lock();
            while(lists.size() == MAX) {
                producer.await();
            }
            
            lists.add(t);
            ++count;
            consumer.signalAll(); //通知消费者线程进行消费
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    
    public T get() {
        T t = null;
        try {
            lock.lock();
            while(lists.size() == 0) {
                consumer.await();
            }
            t = lists.removeFirst();
            count --;
            producer.signalAll(); //通知生产者进行生产
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
        return t;
    }
    
    public static void main(String[] args) {
        MyContainer2<String> c = new MyContainer2<>();
        //启动消费者线程
        for(int i=0; i<10; i++) {
            new Thread(()->{
                for(int j=0; j<5; j++){
                    System.out.println(c.get());
                }
            }, "c" + i).start();
        }
        
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        //启动生产者线程
        for(int i=0; i<2; i++) {
            new Thread(()->{
                for(int j=0; j<25; j++) {
                    c.put(Thread.currentThread().getName() + " " + j);
                }
            }, "p" + i).start();
        }
    }
}
View Code
使用lock和condition好处在于可以精确的通知那些线程被叫醒,哪些线程不必被叫醒,这个效率显然要比notifyAll把所有线程全叫醒要高很多。
 
 
 

 八、ThreadLocal                  

java高并发编程(三)java高并发编程(三)
/**
 * ThreadLocal线程局部变量
 */
package yxxy.c_022;

import java.util.concurrent.TimeUnit;

public class ThreadLocal1 {
    volatile static Person p = new Person();
    
    public static void main(String[] args) {
                
        new Thread(()->{
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            
            System.out.println(p.name);
        }).start();
        
        new Thread(()->{
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            p.name = "lisi";
        }).start();
    }
}

class Person {
    String name = "zhangsan";
}
View Code
现在这两个线程是互相影响的;第二个线程改了名字之后,第一个线程就能读的到了;
有的时候就想线程2的改变,不想让线程1知道,这时候怎么做?
 
java高并发编程(三)java高并发编程(三)
/**
 * ThreadLocal线程局部变量
 *
 * ThreadLocal是使用空间换时间,synchronized是使用时间换空间
 * 比如在hibernate中session就存在与ThreadLocal中,避免synchronized的使用
 *
 * 运行下面的程序,理解ThreadLocal
 */
package yxxy.c_022;

import java.util.concurrent.TimeUnit;

public class ThreadLocal2 {
    static ThreadLocal<Person> tl = new ThreadLocal<>();
    
    public static void main(String[] args) {
                
        new Thread(()->{
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            
            System.out.println(tl.get());
        }).start();
        
        new Thread(()->{
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            tl.set(new Person());
        }).start(); 
    }
    
    static class Person {
        String name = "zhangsan";
    }
}
View Code

console输出:null

 

ThreadLocal的意思就是,tl里面的变量,自己的线程自己用;你别的线程里要想用的话,不好意思你自己往里扔;不能用我线程里面放的东西;相当于每个线程都有自己的变量,互相之间不会产生冲突;
可以理解为person对象每个线程里面拷贝了一份,改的都是自己那份,都是自己线程本地的变量,所以空间换时间;ThreadLocal在效率上会更高一些;
有一些需要加锁的对象,如果它们在使用的时候自己进行的改变,自己维护这个状态,不用通知其他线程,那么这个时候可以使用ThreadLocal;