多线程(六)

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实现顺序执行