并发编程专题三 -JAVA 线程的并发工具类，面试要掌握这几个关键点
int mid = (fromIndex+toIndex)/2;
SumTask left = new SumTask(src,fromIndex,mid);
SumTask right = new SumTask(src,mid+1,toIndex);
invokeAll(left,right);
return left.join()+right.join();
}
}
}
public static void main(String[] args) {
ForkJoinPool pool = new ForkJoinPool();
int[] src = MakeArray.makeArray();
SumTask inn
erFind = new SumTask(src,0,src.length-1);
long start = System.currentTimeMillis();
pool.invoke(innerFind);//同步调用
System.out.println("Task is Running.....");
System.out.println("The count is "+innerFind.join()
+" spend time:"+(System.currentTimeMillis()-start)+"ms");
}
}
public class MakeArray {
//数组长度
public static final int ARRAY_LENGTH = 100000000;
public static int[] makeArray() {
//new 一个随机数发生器
Random r = new Random();
int[] result = new int[ARRAY_LENGTH];
for(int i=0;i<ARRAY_LENGTH;i++){
//用随机数填充数组
result[i] = r.nextInt(ARRAY_LENGTH*3);
}
return result;
}
}
二、常用的并发工具类
==========
1、CountDownLatch
作用：是一组线程等待其他的线程完成工作以后在执行，加强版 join
await 用来等待，countDown 负责计数器的减一
代码示例：
import java.util.concurrent.CountDownLatch;
/**
@Auther: BlackKingW
@Date: 2019/4/14 12:09
@Description:
*/
public class UseCountDownLatch {
static CountDownLatch latch = new CountDownLatch(6);
//初始化线程(只有一步，有 4 个)
private static class InitThread implements Runnable{
@Override
public void run() {
System.out.println("Thread_"+Thread.currentThread().getId()
+" ready init work......");
latch.countDown();//初始化线程完成工作了，countDown 方法只扣减一次；
for(int i =0;i<2;i++) {
System.out.println("Thread_"+Thread.currentThread().getId()
+" ........continue do its work");
}
}
}
//业务线程
private static class BusiThread implements Runnable{
@Override
public void run() {
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
for(int i =0;i<3;i++) {
System.out.println("BusiThread_"+Thread.currentThread().getId()
+" do business-----");
}
}
}
public static void main(String[] args) throws InterruptedException {
//单独的初始化线程,初始化分为 2 步，需要扣减两次
new Thread(new Runnable() {
@Override
public void run() {
SleepTools.ms(1); //休眠 1s
System.out.println("Thread_"+Thread.currentThread().getId()
+" ready init work step 1st......");
latch.countDown();//每完成一步初始化工作，扣减一次
System.out.println("begin step 2nd.......");
SleepTools.ms(1);
System.out.println("Thread_"+Thread.currentThread().getId()
+" ready init work step 2nd......");
latch.countDown();//每完成一步初始化工作，扣减一次
}
}).start();
new Thread(new BusiThread()).start();
for(int i=0;i<=3;i++){
Thread thread = new Thread(new InitThread());
thread.start();
}
latch.await();
System.out.println("Main do ites work........");
}
}
2、CyclicBarrier
让一组线程达到某个屏障，被阻塞，一直到组内最后一个线程达到屏障时，屏障开放，所有被阻塞的线程会继续运行 CyclicBarrier(int parties)
CyclicBarrier(int parties, Runnable barrierAction)，屏障开放，barrierAction 定义的任务会执行
代码示例：
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CyclicBarrier;
/**
@Auther: BlackKingW
@Date: 2019/4/14 12:09
@Description:
*/
public class UseCyclicBarrier {
private static CyclicBarrier barrier
= new CyclicBarrier(5,new CollectThread());
private static ConcurrentHashMap<String,Long> resultMap
= new ConcurrentHashMap<>();//存放子线程工作结果的容器
public static void main(String[] args) {
for(int i=0;i<=4;i++){
Thread thread = new Thread(new SubThread());
thread.start();
}
}
//负责屏障开放以后的工作
private static class CollectThread implements Runnable{
@Override
public void run() {
StringBuilder result = new StringBuilder();
for(Map.Entry<String,Long> workResult:resultMap.entrySet()){
result.append("["+workResult.getValue()+"]");
}
System.out.println(" the result = "+ result);
System.out.println("do other business........");
}
}
//工作线程
private static class SubThread implements Runnable{
@Override
public void run() {
long id = Thread.currentThread().getId();//线程本身的处理结果
resultMap.put(Thread.currentThread().getId()+"",id);
Random r = new Random();//随机决定工作线程的是否睡眠
try {
if(r.nextBoolean()) {
Thread.sleep(2000+id);
System.out.println("Thread_"+id+" ....do something ");
}
System.out.println(id+"....is await");
barrier.await();
Thread.sleep(1000+id);
System.out.println("Thread_"+id+" ....do its business ");
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
CountDownLatch 和 CyclicBarrier 辨析
1、countdownlatch 放行由第三者控制，CyclicBarrier 放行由一组线程本身控制
2、countdownlatch 放行条件》=线程数，CyclicBarrier 放行条件=线程数
3、Semaphore
控制同时访问某个特定资源的线程数量，用在流量控制
import java.sql.Connection;
import java.util.LinkedList;
import java.util.concurrent.Semaphore;
/**
@Auther: BlackKingW
@Date: 2019/4/14 12:09
@Description:
*/
public class DBPoolSemaphore {
private final static int POOL_SIZE = 10;
private final Semaphore useful,useless;//useful 表示可用的数据库连接，useless 表示已用的数据库连接
public DBPoolSemaphore() {
this. useful = new Semaphore(POOL_SIZE);
this.useless = new Semaphore(0);
}
//存放数据库连接的容器
private static LinkedList<Connection> pool = new LinkedList<Connection>();
//初始化池
static {
for (int i = 0; i < POOL_SIZE; i++) {
pool.addLast(SqlConnectImpl.fetchConnection());
}
}
/归还连接/
public void returnConnect(Connection connection) throws InterruptedException {
if(connection!=null) {
System.out.println("当前有"+useful.getQueueLength()+"个线程等待数据库连接！！"
+"可用连接数:"+useful.availablePermits());
useless.acquire();
synchronized (pool) {
pool.addLast(connection);
}
useful.release();
}
}
/从池子拿连接/
public Connection takeConnect() throws InterruptedException {
useful.acquire();
Connection conn;
synchronized (pool) {
conn = pool.removeFirst();
}
useless.release();
return conn;
}
}
4、Exchange
两个线程间的数据交换，?用的比较少
还未添加个人签名 2021.03.18 加入
还未添加个人简介
评论