并发编程（二）concurrent 工具类

一、CountDownLatch

经常用于监听某些初始化操作，等初始化执行完毕后，通知主线程继续工作。

import java.util.concurrent.CountDownLatch;

public class CountDownLatchTest extends Thread {

    private final static CountDownLatch countDown = new CountDownLatch(2); // (1)

    @Override

    public void run() {

        // 唤醒线程线程

        countDown.countDown(); // (2)

        System.out.println(Thread.currentThread().getName() + "执行完毕...");

    }

    public static void main(String[] args) {

        new Thread(new CountDownLatchTest()).start();

        new Thread(new CountDownLatchTest()).start();

        try {

            Thread.sleep(1000);

            countDown.await();  // (3)

            System.out.println(Thread.currentThread().getName() + "继续执行...");

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

    }

}

声明一个 CountDownLatch 对象，参数 2 表示被阻塞的线程需要被唤醒再次才能执行。
```
final CountDownLatch countDown = new CountDownLatch(2);
```
countDown() 调用两次后，主线程才会继续执行
```
countDown.countDown();
```
阻塞当前线程-main
```
countDown.await();
```

执行结果如下：

Thread-1执行完毕...

Thread-0执行完毕...

main继续执行...  // Thread-0, Thread-1 执行完成才会继续执行主线程

二、CyclicBarrier

假设有只有的一个场景：每个线程代表一个跑步运动员，当运动员都准备好后，才一起出发，只要有一个没有准备了，大家都等待。

import java.io.IOException;

import java.util.Random;

import java.util.concurrent.BrokenBarrierException;

import java.util.concurrent.CyclicBarrier;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

public class UseCyclicBarrier {

	static class Runner implements Runnable {

	    private CyclicBarrier barrier;

	    private String name;  

	    public Runner(CyclicBarrier barrier, String name) {

	        this.barrier = barrier;

	        this.name = name;

	    }

	    @Override

	    public void run() {

	        try {

	            Thread.sleep(1000 * (new Random()).nextInt(5));

	            System.out.println(name + " 准备OK.");

	            barrier.await(); //(1)

	        } catch (InterruptedException e) {

	            e.printStackTrace();

	        } catch (BrokenBarrierException e) {

	            e.printStackTrace();

	        }

	        System.out.println(name + " Go!!");

	    }

	} 

    public static void main(String[] args) throws IOException, InterruptedException {

        CyclicBarrier barrier = new CyclicBarrier(2);  // (2)

        ExecutorService executor = Executors.newFixedThreadPool(2);  

        executor.submit(new Thread(new Runner(barrier, "Thread-1")));

        executor.submit(new Thread(new Runner(barrier, "Thread-2")));

        executor.shutdown();

    }

}

await() 阻塞当前的线程。
```
barrier.await();
```
声明一个 CyclicBarrier 对象，参数 2 表示 barrier 必须有两个线程都准备好了才能执行。
```
CyclicBarrier barrier = new CyclicBarrier(2);
```

执行结果如下：

Thread-1 准备OK.

Thread-2 准备OK.

Thread-1 Go!!

Thread-2 Go!!

修改 CyclicBarrier barrier = new CyclicBarrier(3) 后这两个线程都会被阻塞，执行结果如下：
```
Thread-1 准备OK.

Thread-2 准备OK.
```

三、Future

future模式请参考这里

四、Semaphore

Semaphore 信号量非常适合高并发访问。

public class UseSemaphore {

    public static void main(String[] args) {

        // 线程池

        ExecutorService exec = Executors.newCachedThreadPool();

        // 只能5个线程同时访问

        final Semaphore semp = new Semaphore(5); // (1)

        // 模拟20个客户端访问

        for (int index = 0; index < 20; index++) {

            final int NO = index;

            Runnable run = new Runnable() {

                public void run() {

                    try {

                        // 获取许可

                        semp.acquire(); // (2)

                        System.out.println("Accessing: " + NO);

                        //模拟实际业务逻辑

                        Thread.sleep((long) (Math.random() * 10000));

                        // 访问完后，释放

                        semp.release(); // (3)

                    } catch (InterruptedException e) {

                    	;

                    }

                }

            };

            exec.execute(run);

        } 

        try {

			Thread.sleep(10);

		} catch (InterruptedException e) {

			e.printStackTrace();

		}

        System.out.println(semp.getQueueLength());

        // 退出线程池

        exec.shutdown();

    }

}

声明一个 Semaphore 对象，参数 5 表示最多有5个线程同时访问。
```
final Semaphore semp = new Semaphore(5);
```
semp.acquire() 获取 semp 对象，如果超过5个线程，那么其余的线程就会阻塞，直到有线程执行完毕。
semp.release() 释放 semp 对象，这样其余的线程就可以执行了。

补充：

PV(page view) 网站的总访问量，页面浏览量或点击量，用户每刷新一次就会记录一次。
UV(unique vistor) 访问网站的一台电脑客户端为一个访客。一般来讲，时间上以00:00~24:00之内相同的客户端记录一次。
QPS(query per second) 即每秒查询数，QPS 很大程度代表了系统业务的繁忙程度。一旦当前 QPS 超过所设定的预警阀值，可以考虑对集群扩容，以免压力过大导致宕机。
RT(response time) 即请求的响应时间，这个指标非常关键，直接说明客户端的体验，因此任何系统设计师都想降低 RT 时间。

对系统进行峰值评估，采用所谓的80/20原则，即80%的请求20%的时间到达：

QRS = (PV * 80%) / (24 * 60 * 60 * 20%)

五、ReentrantLock

import java.util.concurrent.locks.Lock;

import java.util.concurrent.locks.ReentrantLock;

public class ReentrantLockTest implements Runnable {

	private Lock lock = new ReentrantLock(); // (1)

	public void run(){

		try {

			lock.lock(); // (2)

			System.out.println(Thread.currentThread().getName() + "进入..");

			Thread.sleep(1000);

			System.out.println(Thread.currentThread().getName() + "退出..");

			Thread.sleep(1000);

		} catch (InterruptedException e) {

			e.printStackTrace();

		} finally {

			lock.unlock(); // (3)

		}

	}

	public static void main(String[] args) throws InterruptedException {

		final ReentrantLockTest ur = new ReentrantLockTest();

        for (int i = 0; i < 10; i++) {

            new Thread(ur).start();

        }

	}

}

ReentrantLock 一般用法：

private Lock lock = new ReentrantLock();

try {

    lock.lock();

    //do something

} finally {

    lock.unlock();

}

condition 使用方法，注意 condition 可以实例化多个：

Lock lock = new ReentrantLock();

Condition condition = lock.newCondition();

condition.await(); //阻塞线程，释放锁

condition.signal();//唤醒线程，不释放锁

公平锁(true)和非公平锁(false)，非公平锁执行效率比公平锁高
```
Lock lock = new ReentrantLock(boolean isFair);
```

读写锁，实现读写分离的锁，适用于读多写少的情况下(读读共享，读写互斥)

private ReentrantReadWriteLock rwlock = new ReentrantReadWriteLock(); // (1)

private ReadLock readLock = rwlock.readLock();    // (2)

private WriteLock writeLock = rwlock.writeLock(); // (3)

public void read(){

    try {

        readLock.lock();

        // do something

    } finally {

        readLock.unlock();

    }

}

public void write(){

    try {

        writeLock.lock();

        // do something

    } finally {

        writeLock.unlock();

    }

}

每天用心记录一点点。内容也许不重要，但习惯很重要！

秒客网

并发编程（二）concurrent 工具类

并发编程（二）concurrent 工具类

一、CountDownLatch

二、CyclicBarrier

三、Future

四、Semaphore

五、ReentrantLock

相关文章