linuxsir首页 LinuxSir.Org | Linux、BSD、Solaris、Unix | 开源传万世,因有我参与欢迎您!
网站首页 | 设为首页 | 加入收藏
您所在的位置:主页 > Linux基础建设 >

Java并发编程高级内容介绍

时间:2017-03-13  来源:未知  作者:linuxsir首页

计数器:CountDownLatch

CountDownLatch类似于一个计数器,和Atomic类比较相近,操作是原子的,即多个线程同时只能有一个可以去操作。CountDownLatch对象设置一个初始的数字作为计数值,任何调用这个对象上的await()方法都会阻塞,直到这个计数器的计数值被其他的线程调用countDown()减为0为止。典型的应用场景就是:有一个任务想要往下执行,但必须要等到其他的任务执行完毕后才可以继续往下执行。例如在Zookeeper的使用过程中,由于客户端与服务器建立连接是异步调用的,因此主线程需要await()阻塞直至异步回调countDown()完成。

代码示例:

public class CountDownLatchTest {
 
 
    public static void main(String[] args) {
 
        final CountDownLatch countDownLatch = new CountDownLatch(2);
 
        Thread work1 = new Thread(new Runnable() {
            @Override
            public void run() {
 
                System.out.println(Thread.currentThread() + " doing work...start");
 
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
 
                System.out.println(Thread.currentThread() + " doing work...end ");
                countDownLatch.countDown();
            }
        },"work1");
 
 
        Thread work2 = new Thread(new Runnable() {
            @Override
            public void run() {
 
                System.out.println(Thread.currentThread().getName() + " doing work...start");
 
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
 
                System.out.println(Thread.currentThread().getName() + " doing work...end ");
                countDownLatch.countDown();
            }
        },"work2");
 
        work1.start();
        work2.start();
 
        try {
            countDownLatch.await();
            System.out.println("all workers finish ");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
 
 
    }
 
 
}

齐步走:CyclicBarrier

Barrier的意思是栅栏,就是让一组线程相互等待,直至所有线程都到齐了,那么就可以齐步走。Cyclic是循环的意思,就是说Barrier可以循环使用。CyclicBarrier主要的方法就是await(),较CountDownLatch的await()虽然都是阻塞,但是CyclicBarrier.await()有返回值int,即当前线程是第几个到达这个Barrier的线程。

构造CyclicBarrier时指定计数值,await() 方法每被调用一次,计数便会减少1,并阻塞住当前线程。当计数减至0时,阻塞解除,所有在此 CyclicBarrier 上面阻塞的线程开始运行。在这之后,如果再次调用 await() 方法,计数就又会变成 N-1,新一轮重新开始。在构造方法上还可以传递一个Runnable对象,阻塞解除时这个Runnable会得到运行。

CyclicBarrier有点“不见不散”的味道,想一想,如果某个成员因某种原因来不了Barrier这个地方,那么我们一直等待吗?实际中,如果来不了理应通知其他成员,别等了,回家吧!注意到CyclicBarrier.await()独有的BrokenBarrierException异常

代码示例:

public class CyclicBarrierTest {
 
    public static void main(String[] args) {
 
        final CyclicBarrier cyclicBarrier = new CyclicBarrier(2, new Runnable() {
            @Override
            public void run() {
                System.out.println("都准备好啦!");
            }
        });
 
        Thread runman1 = new Thread(new Runnable() {
            @Override
            public void run() {
 
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
 
                try {
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread().getName() + "i am ok");
 
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
        },"runman1");
 
 
        Thread runman2 = new Thread(new Runnable() {
            @Override
            public void run() {
 
                try {
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread().getName() + "i am ok");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
        },"runman2");
 
        runman1.start();
        runman2.start();
 
    }
 
}

Callable And Future

在我以前的博客《Java Future模式实现》中有介绍Future模式,Future模式非常适合在处理耗时很长的业务逻辑,可以有效的减少系统的响应时间,提高系统的吞吐量。JDK其实已经为我们提供了API实现,我们来看一段代码即可:

public class FutureTest {
 
 
    public static void main(String[] args) {
 
        FutureTask<String> futureTask = new FutureTask<String>(new Callable<String>() {
            @Override
            public String call() throws Exception {
 
                Thread.sleep(2000);
                return "ok";
            }
        });
 
        ExecutorService es = Executors.newFixedThreadPool(1);
        es.submit(futureTask);
        System.out.println("开启线程去异步处理,主线程继续往下执行!");
        try {
 
            System.out.println("取得异步处理结果:" + futureTask.get());
 
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
 
 
    }
 
}

注意到线程池执行任务,可以利用2个方法:



submit和execute有什么区别呢?从入参和结果类型就知道了。

信号量:Semaphore

Semaphore实现的功能就类似厕所有5个坑,假如有10个人要上厕所,那么显然同时只能有5个人占用厕所,当5个人中的任何一个人让开后,其中等待的另外5个人中又有一个人可以占用了。另外等待的5个人中可以是随机获得优先机会,也可以是按照先来后到的顺序获得机会,这取决于构造Semaphore对象时传入的fair参数选项。

Semaphore可以控制某个资源可被同时访问的个数(构造方法传入),通过 acquire() 获取一个许可,如果没有就等待,而 release() 释放一个许可。

代码示例:

public static void main(String[] args) {
 
    final Semaphore semaphore = new Semaphore(5);
 
    for(int i = 0 ; i < 6 ; i++){
 
        new Thread(new Runnable() {
            @Override
            public void run() {
 
                try {
                    semaphore.acquire();
 
                    System.out.println(Thread.currentThread().getName() + " 运行...");
 
                    Thread.sleep(1000);
 
                    semaphore.release();
 
                    System.out.println(Thread.currentThread().getName() + " 结束...");
 
 
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
 
            }
        },String.valueOf(i)).start();
 
 
    }
 
}

Condition

JDK由原始的synchronized发展到Lock,以类的方式提供锁机制,发展出重入锁、读写锁,以类的形式存在自然功能更加强大灵活,比如可以tryLock进行锁的嗅探。在synchronized代码块中我们可以使用wait/notify/notifyAll来进行线程的协同工作,那么JDK也发展了这一块,即Condition。Condition.await类似于wait,Condition.signal/signalAll类似于notify/nofityAll。下面我们简单实现一个Condition版的生产者/消费者。

处理核心:Handler

public class Handler {
 
    //容器
    private LinkedList<String> linkedList = new LinkedList<String>();
 
    //限制
    private int MAX_SIZE = 3;
 
    //锁
    private Lock lock = new ReentrantLock();
 
    //condition  实际上,可以new多个condition,这里暂且只是用给一个
    private Condition condition = lock.newCondition();
 
    public void put(String bread){
 
        try{
            lock.lock();
 
            if(linkedList.size() == MAX_SIZE){
                System.out.println("容器已满");
                condition.await();
            }
 
            linkedList.add(bread);
            System.out.println("放入面包" + bread);
            condition.signalAll();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
 
    }
 
    public void eat(){
 
        try{
            lock.lock();
 
            if(linkedList.size() == 0){
                System.out.println("容器为空");
                condition.await();
            }
 
            String bread = linkedList.removeFirst();
            System.out.println("吃掉一个面包" + bread);
            condition.signalAll();
 
        }catch(Exception e){
 
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
 
 
    }
 
}

生产者:Produce

public class Produce implements Runnable{
 
    private Handler handler;
 
    public Produce(Handler handler) {
        this.handler = handler;
    }
 
    @Override
    public void run() {
 
        for(int i = 0 ; i < 10 ; i++){
            try {
                Thread.sleep(new Random().nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            handler.put(String.valueOf(i));
        }
 
    }
}

消费者:Consume

public class Consume implements Runnable{
 
    private Handler handler;
 
    public Consume(Handler handler) {
        this.handler = handler;
    }
 
    @Override
    public void run() {
        for (int i = 0 ; i < 10 ; i++){
            try {
                Thread.sleep(new Random().nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            handler.eat();
        }
    }
}

Main:

public class Main {
 
    public static void main(String[] args) {
 
        Handler handler = new Handler();
 
        Produce produce = new Produce(handler);
        Consume consume = new Consume(handler);
 
        new Thread(consume).start();
        new Thread(produce).start();
        new Thread(produce).start();
    }
}

友情链接