多線程(六)

1.lock

用lock實現同步:

public class MyService {
    private Lock lock=new ReentrantLock();
    public void testMethod(){
        lock.lock();
        for (int i = 0; i <5; i++) {
            System.out.println("ThreadName="+Thread.currentThread().getName()+"("+(i+1)+")");
        }
        lock.unlock();
    }
}

public class MyThread extends Thread{
    private MyService myService;
    public MyThread(MyService myService){
        this.myService=myService;
    }

    @Override
    public void run() {
        myService.testMethod();
    }
}

public class Run {
    public static void main(String[] args) {
        MyService service=new MyService();
        MyThread a1=new MyThread(service);
        MyThread a2=new MyThread(service);
        MyThread a3=new MyThread(service);
        MyThread a4=new MyThread(service);
        MyThread a5=new MyThread(service);
        a1.start();
        a2.start();
        a3.start();
        a4.start();
        a5.start();
    }
/*ThreadName=Thread-0(1)
ThreadName=Thread-0(2)
ThreadName=Thread-0(3)
ThreadName=Thread-0(4)
ThreadName=Thread-0(5)
ThreadName=Thread-2(1)
ThreadName=Thread-2(2)
ThreadName=Thread-2(3)
ThreadName=Thread-2(4)
ThreadName=Thread-2(5)
ThreadName=Thread-4(1)
ThreadName=Thread-4(2)
ThreadName=Thread-4(3)
ThreadName=Thread-4(4)
ThreadName=Thread-4(5)
ThreadName=Thread-1(1)
ThreadName=Thread-1(2)
ThreadName=Thread-1(3)
ThreadName=Thread-1(4)
ThreadName=Thread-1(5)
ThreadName=Thread-3(1)
ThreadName=Thread-3(2)
ThreadName=Thread-3(3)
ThreadName=Thread-3(4)
ThreadName=Thread-3(5)
*/實現了同步
}

2.關鍵字synchronized與wait()和notify/notifyAll()方法相結合但是實現等待/通知模式,類ReentrantLock也能夠實現一樣的功能,但須要藉助Condition對象.

notify是隨機通知的,可是類ReentrantLock結合Condition類是能夠實現"選擇性通知的"

public class ThreadA extends Thread{
    private MyService myService;
    public ThreadA(MyService myService){
        this.myService=myService;
    }

    @Override
    public void run() {
        myService.await();
    }
}

public class MyService {
    private Lock lock=new ReentrantLock();
    private Condition condition=lock.newCondition();//對象監視器
    public void await(){
        try {
            condition.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public class Run {
    public static void main(String[] args) {
        MyService service=new MyService();
        ThreadA a =new ThreadA(service);
        a.start();
    }
/*Exception in thread "Thread-0" java.lang.IllegalMonitorStateException
    at java.util.concurrent.locks.ReentrantLock$Sync.tryRelease(ReentrantLock.java:155)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.release(AbstractQueuedSynchronizer.java:1260)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.fullyRelease(AbstractQueuedSynchronizer.java:1723)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2040)
    at lock.MyService.await(MyService.java:15)
    at lock.ThreadA.run(ThreadA.java:14)
*/
}

總結:監視器報錯,解決方法是在condition.await()方法調用以前調用lock.lock();

修改:

public class MyService {
    private Lock lock=new ReentrantLock();
    private Condition condition=lock.newCondition();//對象監視器
    public void await(){
        try {
            lock.lock();
            System.out.println("A");
            condition.await();
            System.out.println("B");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
/*OUTPUT:A
*/
}

lock的喚醒實現:

public class ThreadA extends Thread{
    private MyService myService;
    public ThreadA(MyService myService){
        this.myService=myService;
    }

    @Override
    public void run() {
        myService.await();
    }
}

public class MyService {
    private Lock lock=new ReentrantLock();
    private Condition condition=lock.newCondition();//對象監視器
    public void await(){
        try {
            lock.lock();
            System.out.println("await 時間爲"+System.currentTimeMillis());
            condition.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    public void signal(){
        try{
            lock.lock();
            System.out.println("signal時間爲"+System.currentTimeMillis());
            condition.signal();
        }finally {
            lock.unlock();
        }
    }
}

public class Run {
    public static void main(String[] args) throws InterruptedException {
        MyService service=new MyService();
        ThreadA a =new ThreadA(service);
        a.start();
        Thread.sleep(3000);
        service.signal();
    }
/*
await 時間爲1516676645558
signal時間爲1516676648557
*/
}

4.多個condition實現通知所有線程

public class MyService {
    private Lock lock=new ReentrantLock();
    private Condition condition=lock.newCondition();//對象監視器
    public void awaitA(){
        try {
            lock.lock();
            System.out.println("begin awaitA 時間爲"+System.currentTimeMillis()+" ThreadName="+Thread.currentThread().getName());
            condition.await();
            System.out.println("end awaitA 時間爲"+System.currentTimeMillis()+" ThreadName="+Thread.currentThread().getName());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }
    public void awaitB(){
        try {
            lock.lock();
            System.out.println("begin awaitB 時間爲"+System.currentTimeMillis()+" ThreadName="+Thread.currentThread().getName());
            condition.await();
            System.out.println("end awaitB 時間爲"+System.currentTimeMillis()+" ThreadName="+Thread.currentThread().getName());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }
    public void signalAll(){
        try{
            lock.lock();
            System.out.println("signalAll時間爲"+System.currentTimeMillis()+" ThreadName="+Thread.currentThread().getName());
            condition.signalAll();
        }finally {
            lock.unlock();
        }
    }
}

public class ThreadA extends Thread{
    private MyService myService;
    public ThreadA(MyService myService){
        this.myService=myService;
    }

    @Override
    public void run() {
        myService.awaitA();
    }
}

public class ThreadB extends Thread{
    private MyService myService;
    public ThreadB(MyService myService){
        this.myService=myService;
    }

    @Override
    public void run() {
        myService.awaitB();
    }
}

public class Run {
    public static void main(String[] args) throws InterruptedException {
        MyService service=new MyService();
        ThreadA a =new ThreadA(service);
        a.setName("A");
        a.start();
        ThreadB b =new ThreadB(service);
        b.setName("B");
        b.start();
        Thread.sleep(3000);
        service.signalAll();
    }
}

5.多個condition實現通知部分線程

public class MyService {
    private Lock lock=new ReentrantLock();
    private Condition conditionA=lock.newCondition();//A對象監視器
    private Condition conditionB=lock.newCondition();//B對象監視器
    public void awaitA(){
        try {
            lock.lock();
            System.out.println("begin awaitA 時間爲"+System.currentTimeMillis()+" ThreadName="+Thread.currentThread().getName());
            conditionA.await();
            System.out.println("end awaitA 時間爲"+System.currentTimeMillis()+" ThreadName="+Thread.currentThread().getName());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }
    public void awaitB(){
        try {
            lock.lock();
            System.out.println("begin awaitB 時間爲"+System.currentTimeMillis()+" ThreadName="+Thread.currentThread().getName());
            conditionB.await();
            System.out.println("end awaitB 時間爲"+System.currentTimeMillis()+" ThreadName="+Thread.currentThread().getName());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }
    public void signalAll_A(){
        try{
            lock.lock();
            System.out.println("signalAll時間爲"+System.currentTimeMillis()+" ThreadName="+Thread.currentThread().getName());
            conditionA.signalAll();
        }finally {
            lock.unlock();
        }
    }
    public void signalAll_B(){
        try{
            lock.lock();
            System.out.println("signalAll時間爲"+System.currentTimeMillis()+" ThreadName="+Thread.currentThread().getName());
            conditionB.signalAll();
        }finally {
            lock.unlock();
        }
    }
}

public class ThreadA extends Thread{
    private MyService myService;
    public ThreadA(MyService myService){
        this.myService=myService;
    }

    @Override
    public void run() {
        myService.awaitA();
    }
}

public class ThreadB extends Thread{
    private MyService myService;
    public ThreadB(MyService myService){
        this.myService=myService;
    }

    @Override
    public void run() {
        myService.awaitB();
    }
}

public class Run {
    public static void main(String[] args) throws InterruptedException {
        MyService service=new MyService();
        ThreadA a =new ThreadA(service);
        a.setName("A");
        a.start();
        ThreadB b =new ThreadB(service);
        b.setName("B");
        b.start();
        Thread.sleep(3000);
        service.signalAll_A();
    }
}

6.實現生產者消費者模式:一對一交替打印

public class MyService {
    private Lock lock=new ReentrantLock();
    private Condition condition=lock.newCondition();//對象監視器
    private boolean hasValue=false;
    public void set(){
        try{
            lock.lock();
            while (hasValue==true){
                condition.await();
            }
            System.out.println("打印*");
            hasValue=true;
            condition.signal();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    public void get(){
        try {
            lock.lock();
            while(hasValue==false){
                condition.await();
            }
            System.out.println("打印-");
            hasValue=false;
            condition.signal();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public class ThreadA extends Thread{
    private MyService myService;
    public ThreadA(MyService myService){
        this.myService=myService;
    }

    @Override
    public void run() {
        for (int i = 0; i <Integer.MAX_VALUE ; i++) {
            myService.set();
        }
    }
}

public class ThreadB extends Thread{
    private MyService myService;
    public ThreadB(MyService myService){
        this.myService=myService;
    }

    @Override
    public void run() {
        for (int i = 0; i <Integer.MAX_VALUE ; i++) {
            myService.get();
        }
    }
}

public class Run {
    public static void main(String[] args) throws InterruptedException {
        MyService service=new MyService();
        ThreadA a =new ThreadA(service);
        a.start();
        ThreadB b =new ThreadB(service);
        b.start();
    }
/*
打印-
打印*
打印-
打印*
打印-
打印*
*/
}

7.實現生產者消費者模式:多對多交替打印

public class MyService {
    private Lock lock=new ReentrantLock();
    private Condition condition=lock.newCondition();//對象監視器
    private boolean hasValue=false;
    public void set(){
        try{
            lock.lock();
            while (hasValue==true){
                System.out.println("有可能**連續");
                condition.await();
            }
            System.out.println("打印*");
            hasValue=true;
            condition.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    public void get(){
        try {
            lock.lock();
            while(hasValue==false){
                System.out.println("有可能--連續");
                condition.await();
            }
            System.out.println("打印-");
            hasValue=false;
            condition.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public class ThreadA extends Thread{
    private MyService myService;
    public ThreadA(MyService myService){
        this.myService=myService;
    }

    @Override
    public void run() {
        for (int i = 0; i <Integer.MAX_VALUE ; i++) {
            myService.set();
        }
    }
}

public class ThreadB extends Thread{
    private MyService myService;
    public ThreadB(MyService myService){
        this.myService=myService;
    }

    @Override
    public void run() {
        for (int i = 0; i <Integer.MAX_VALUE ; i++) {
            myService.get();
        }
    }
}

public class Run {
    public static void main(String[] args) throws InterruptedException {
        MyService service=new MyService();
        ThreadA[] threadA= new ThreadA[10];
        ThreadB[] threadB= new ThreadB[10];
        for (int i = 0; i < 10; i++) {
            threadA[i]=new ThreadA(service);
            threadB[i]=new ThreadB(service);
            threadA[i].start();
            threadB[i].start();
        }
    }
/*
//signal會出現假死
...
打印*
有可能**連續
打印-
有可能--連續
有可能--連續
打印*
有可能**連續
有可能**連續
...
*/
}

8.公平鎖和非公平鎖

鎖lock分爲公平鎖和非公平鎖,公平鎖表示線程獲取鎖的順序是按照線程加鎖的順序來分配的,即先來先得,非公平鎖是一種獲取鎖的搶佔機制,是隨機得到鎖的.默認是非公平鎖.

9.getHoldCount() getQueueLength() getWaitQueueLength()

int getHoldCount() 查詢當前線程保持此鎖定的個數,也就是調用lock方法的次數.

public class Service {
    private ReentrantLock lock=new ReentrantLock();
    public void serviceMethod1(){
        try{
            lock.lock();
            System.out.println("serviceMethod1 getHoldCount="+lock.getHoldCount());
            serviceMethod2();
        }finally {
            lock.unlock();
        }

    }
    public void serviceMethod2(){
        try{
            lock.lock();
            System.out.println("serviceMethod2 getHoldCount="+lock.getHoldCount());
        }finally {
            lock.unlock();
        }
    }
}

public class Run {
    public static void main(String[] args) throws InterruptedException {
        Service service=new Service();
        service.serviceMethod1();
    }
}

getQueueLength() 返回正在等待獲取此鎖定的線程估計數

public class Service {
    private ReentrantLock lock=new ReentrantLock();
    public void serviceMethod1(){
        try{
            lock.lock();
            System.out.println("ThreadName="+Thread.currentThread().getName()+"進入方法");
            Thread.sleep(Integer.MAX_VALUE);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }

    }

    public static void main(String[] args) throws InterruptedException {
        final Service service=new Service();
        Runnable runnable=new Runnable() {
            @Override
            public void run() {
                service.serviceMethod1();
            }
        };
        Thread[] threadArray=new Thread[10];
        for (int i = 0; i <10; i++) {
            threadArray[i]=new Thread(runnable);
        }
        for (int i = 0; i <10 ; i++) {
            threadArray[i].start();
        }
        Thread.sleep(2000);
        System.out.println("有線程數:"+service.lock.getQueueLength()+"個在等待獲取鎖");
    }
/*
ThreadName=Thread-0進入方法
有線程數:9個在等待獲取鎖
*/
}

getWaitQueueLength(Condition condition)返回等待與此鎖定相關的給定條件Condition的線程估計數,好比有5個線程,每一個線程都執行同一個Condition對象的await()方法,則調用以後返回5

public class Service {
    private ReentrantLock lock=new ReentrantLock();
    private Condition condition=lock.newCondition();
    public void waitMethod(){
        try{
            lock.lock();
            condition.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    public void notifyMethod(){
        try{
            lock.lock();
            System.out.println("有"+lock.getWaitQueueLength(condition)+"個線程正在等待newCondition");
            condition.signal();
        } finally {
            lock.unlock();
        }
    }
}

public class Run {
    public static void main(String[] args) throws InterruptedException {
        final Service service=new Service();
        Runnable runnable=new Runnable() {
            @Override
            public void run() {
                service.waitMethod();
            }
        };
        Thread[] threadArray=new Thread[10];
        for (int i = 0; i <10 ; i++) {
            threadArray[i]=new Thread(runnable);
        }
        for (int i = 0; i <10 ; i++) {
            threadArray[i].start();
        }
        Thread.sleep(2000);
        service.notifyMethod();
    }
/*有10個線程正在等待newCondition*/
}

10 hasQueuedThread() hasQueuedThreads() hasWaiters()

boolean hasQueuedThread(Thread thread) 查詢指定的線程是否正在等待獲取此鎖定

boolean hasQueuedThreads() 查詢是否有線程正在等待獲取此鎖定

public class Service {
    public ReentrantLock lock=new ReentrantLock();
    private Condition condition=lock.newCondition();
    public void waitMethod(){
        try{
            lock.lock();
            Thread.sleep(Integer.MAX_VALUE);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
}

public class Run {
    public static void main(String[] args) throws InterruptedException {
        final Service service=new Service();
        Runnable runnable=new Runnable() {
            @Override
            public void run() {
                service.waitMethod();
            }
        };
        Thread threadA=new Thread(runnable);
        threadA.start();
        threadA.sleep(500);
        Thread threadB=new Thread(runnable);
        threadB.start();
        threadB.sleep(500);
        System.out.println(service.lock.hasQueuedThread(threadA));
        System.out.println(service.lock.hasQueuedThread(threadB));
        System.out.println(service.lock.hasQueuedThreads());
    }
/*
false
true
true
*/
}

hasWaiters(Condition condition)查詢是否有線程正在等待與此鎖定有關的condition條件

public class Service {
    public ReentrantLock lock=new ReentrantLock();
    private Condition condition=lock.newCondition();
    public void waitMethod(){
        try{
            lock.lock();
            condition.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    public void notifyMethod(){
        try{
            lock.lock();
            System.out.println("有沒有線程正在等待condition?"+lock.hasWaiters(condition)+" 線程數是多少?"+lock.getWaitQueueLength(condition));
        } finally {
            lock.unlock();
        }
    }
}

public class Run {
    public static void main(String[] args) throws InterruptedException {
        final Service service=new Service();
        Runnable runnable=new Runnable() {
            @Override
            public void run() {
                service.waitMethod();
            }
        };
        Thread[] threadArray=new Thread[10];
        for (int i = 0; i <10 ; i++) {
            threadArray[i]=new Thread(runnable);
        }
        for (int i = 0; i <10; i++) {
            threadArray[i].start();
        }
        Thread.sleep(2000);
        service.notifyMethod();
    }
/*
有沒有線程正在等待condition?true 線程數是多少?10
*/
}

11.isFair() isHeldByCurrentThread() isLocked()

isFair()獲取公平鎖狀態

isHeldByCurrentThread()查詢當前線程是否保持此鎖定(相似於isInterrupted())

public class Service {
    private ReentrantLock lock;
    public Service(boolean isFair){
        super();
        lock=new ReentrantLock(isFair);
    }
    public void serviceMethod(){
        try{
            System.out.println(lock.isHeldByCurrentThread());
            lock.lock();
            System.out.println(lock.isHeldByCurrentThread());
        }finally {
            lock.unlock();
        }
    }
}

public class Run {
    public static void main(String[] args) throws InterruptedException {
        final Service service=new Service(true);
        Runnable runnable=new Runnable() {
            @Override
            public void run() {
                service.serviceMethod();
            }
        };
        Thread thread=new Thread(runnable);
        thread.start();
    }
/*
false
true
*/
}

isLocked() 查詢此鎖是否被任意線程保持

public class Service {
    private ReentrantLock lock;
    public Service(boolean isFair){
        super();
        lock=new ReentrantLock(isFair);
    }
    public void serviceMethod(){
        try{
            System.out.println(lock.isLocked());
            lock.lock();
            System.out.println(lock.isLocked());
        }finally {
            lock.unlock();
        }
    }
}

public class Run {
    public static void main(String[] args) throws InterruptedException {
        final Service service=new Service(true);
        Runnable runnable=new Runnable() {
            @Override
            public void run() {
                service.serviceMethod();
            }
        };
        Thread thread=new Thread(runnable);
        thread.start();
    }
/*
false
true
*/
}

12.lockInterruptibly() tryLock() tryLock(long timeout,TimeUnit unit)

lockInterruptibly() 當前線程未被中斷,則獲取鎖定,若是已經被中斷則出現異常

public class Service {
    public ReentrantLock lock=new ReentrantLock();
    private Condition condition=lock.newCondition();
    public void waitMethod(){
        try{
            lock.lockInterruptibly();
            System.out.println("lock begin "+Thread.currentThread().getName());
            for (int i = 0; i < Integer.MAX_VALUE/10; i++) {
                String newString=new String();
                Math.random();
            }
            System.out.println("lock end "+Thread.currentThread().getName());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            if(lock.isHeldByCurrentThread()){
                lock.unlock();
            }
        }
    }
}

public class Run {
    public static void main(String[] args) throws InterruptedException {
        final Service service=new Service();
        Runnable runnable=new Runnable() {
            @Override
            public void run() {
                service.waitMethod();
            }
        };
        Thread threadA=new Thread(runnable);
        threadA.setName("A");
        threadA.start();
        Thread.sleep(500);
        Thread threadB=new Thread(runnable);
        threadB.setName("B");
        threadB.start();
        threadB.interrupt();//打標記
        System.out.println("main end");
    }
/*
lock begin A
main end
java.lang.InterruptedException
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireInterruptibly(AbstractQueuedSynchronizer.java:1219)
    at java.util.concurrent.locks.ReentrantLock.lockInterruptibly(ReentrantLock.java:340)
    at lock.Service.waitMethod(Service.java:14)
    at lock.Run$1.run(Run.java:12)
    at java.lang.Thread.run(Thread.java:722)
*/
}

boolean tryLock() 僅在調用時鎖定未被另外一個線程保持的狀況下,才獲取該鎖定

public class Service {
    public ReentrantLock lock=new ReentrantLock();
    private Condition condition=lock.newCondition();
    public void waitMethod(){
        if(lock.tryLock()){
            System.out.println(Thread.currentThread().getName()+"得到鎖");
        }else{
            System.out.println(Thread.currentThread().getName()+"沒有得到鎖");
        }
    }
}

public class Run {
    public static void main(String[] args) throws InterruptedException {
        final Service service=new Service();
        Runnable runnable=new Runnable() {
            @Override
            public void run() {
                service.waitMethod();
            }
        };
        Thread threadA=new Thread(runnable);
        threadA.setName("A");
        threadA.start();
        Thread.sleep(500);
        Thread threadB=new Thread(runnable);
        threadB.setName("B");
        threadB.start();
    }
/*
A得到鎖
B沒得到鎖
*/
}

tryLock(long timeout,TimeUnit unit) 若是鎖定在給定等待時間內沒有被另外一個線程保持,且當前線程未被中斷,則獲取該鎖定.

13.awaitUtil() 線程在等待時間內,能夠被其餘線程提早喚醒

public class Service {
    public ReentrantLock lock=new ReentrantLock();
    private Condition condition=lock.newCondition();
    public void waitMethod(){
        try{
            Calendar calendar=Calendar.getInstance();
            calendar.add(Calendar.SECOND,10);
            lock.lock();
            System.out.println("wait begin timer="+System.currentTimeMillis());
            condition.awaitUntil(calendar.getTime());
            System.out.println("wait end timer="+System.currentTimeMillis());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            if(lock.isHeldByCurrentThread()){
                lock.unlock();
            }
        }
    }
    public void notifyMethod(){
        try{
            Calendar calendar=Calendar.getInstance();
            calendar.add(Calendar.SECOND,10);
            lock.lock();
            System.out.println("notify begin timer="+System.currentTimeMillis());
            condition.signalAll();
            System.out.println("notify end timer="+System.currentTimeMillis());
        } finally {
            if(lock.isHeldByCurrentThread()){
                lock.unlock();
            }
        }
    }
}

public class ThreadA extends Thread{
    private Service service;
    public ThreadA(Service service){
        this.service=service;
    }

    @Override
    public void run() {
        service.waitMethod();
    }
}

public class ThreadB extends Thread{
    private Service service;
    public ThreadB(Service service){
        this.service=service;
    }

    @Override
    public void run() {
        service.notifyMethod();
    }
}

public class Run {
    public static void main(String[] args) throws InterruptedException {
        Service service=new Service();
        ThreadA threadA=new ThreadA(service);
        threadA.start();
        ThreadB threadB=new ThreadB(service);
        threadB.start();
    }
/*
wait begin timer=1516776325733
notify begin timer=1516776325733
notify end timer=1516776325733
wait end timer=1516776325733
*/
}

14.使用Condition實現順序執行