Java学习笔记11多线程工具类实战

编程

Java学习笔记11-多线程工具类实战

Runnable接口

Runnable是个接口,里面只有个run方法,Thread也实现Runnable接口,所以要实现多线程那你最终要实现Runnable接口,不多说直接上代码演示。

Runnable测试代码

public class RunnableDemo {

public static void main(String[] args) {

// 可以定义好一个类实现Runnable接口

RunnableImpl runnable = new RunnableImpl();

new Thread(runnable).start();

// 省事写法

new Thread(new Runnable() {

@Override

public void run() {

System.out.println(Thread.currentThread().getName() + "当前线程");

}

}).start();

// lambda函数式接口,更优雅

new Thread(() -> System.out.println(Thread.currentThread().getName() + "当前线程")).start();

}

}

class RunnableImpl implements Runnable {

@Override

public void run() {

// 具体执行逻辑

System.out.println(Thread.currentThread().getName() + "当前线程");

}

}

Callable / Future

Callable也是个接口,里面只有个call方法,与Runnable的区别是Callable配合Future可以得到返回值、抛异常等;Future表示异步计算的结果,提供了用于检查计算是否完成、等待计算完成以及获取结果的方法;常用的为FutureTask,实际实在run方法里执行了run,下面我们直接看代码。

Callable测试代码

import java.util.Random;

import java.util.concurrent.Callable;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.FutureTask;

public class CallableDemo {

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

Callable<String> callable = new Callable<String>() {

@Override

public String call() throws Exception {

Random random = new Random();

return "返回值为:" + random.nextInt(200);

}

};

FutureTask<String> futureTask = new FutureTask<>(callable);

new Thread(futureTask).start();

String result = futureTask.get(); // 阻塞,等待执行结果后继续

System.out.println(result);

}

}

看到Callable / Future的效果了吗,没错如果我们把它应用在实际业务中呢?我有一个idea!看代码~

FutureTask测试代码

import java.util.HashMap;

import java.util.Map;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.FutureTask;

import java.util.concurrent.TimeUnit;

public class MyFutureTaskDemo {

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

// 实际业务场景中会有一个方法 调用多次接口 / 查询多个数据库表 / 多个sql

// 如果顺序执行的话执行时间是累加的,但是要是使用FutureTask执行的话呢?执行时间是最慢的那个,这不效率就提升了~

// 下面举个栗子

// 这里可以用我们之前写的MyFutureTask,效果是一样的

FutureTask<String> futureTask1 = new FutureTask<>(() -> {

TimeUnit.SECONDS.sleep(3L);

return "任务1 接口执行完毕";

});

FutureTask<String> futureTask2 = new FutureTask<>(() -> {

TimeUnit.SECONDS.sleep(5L);

return "任务2 接口执行完毕";

});

FutureTask<String> futureTask3 = new FutureTask<>(() -> {

TimeUnit.SECONDS.sleep(2L);

return "任务3 接口执行完毕";

});

long time = System.currentTimeMillis();

new Thread(futureTask1).start();

new Thread(futureTask2).start();

new Thread(futureTask3).start();

Map<String, String> result = new HashMap<>();

result.put("result1", futureTask1.get());

result.put("result2", futureTask2.get());

result.put("result3", futureTask3.get());

System.out.println(result);

System.out.println("执行时间为:" + (System.currentTimeMillis() - time));

}

}

CountDownLatch使用

CountDownLatch也可以实现异步多线程处理,你可以理解它是一个倒计数器,等待计数器的数值为0就继续执行,也可以做并发测试,下面直接上测试代码~

CountDownLatch测试代码

import java.util.HashMap;

import java.util.Map;

import java.util.concurrent.CountDownLatch;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.TimeUnit;

public class CountDownLatchDemo {

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

ExecutorService executorService = Executors.newCachedThreadPool();

CountDownLatch count = new CountDownLatch(3);

Map<String, String> result = new HashMap<>();

long time = System.currentTimeMillis();

executorService.submit(() -> {

try {

TimeUnit.SECONDS.sleep(3L);

result.put("result1", "任务1 接口执行完毕");

} catch (InterruptedException e) {

e.printStackTrace();

} finally {

count.countDown();

}

});

executorService.submit(() -> {

try {

TimeUnit.SECONDS.sleep(5L);

result.put("result2", "任务2 接口执行完毕");

} catch (InterruptedException e) {

e.printStackTrace();

} finally {

count.countDown();

}

});

executorService.submit(() -> {

try {

TimeUnit.SECONDS.sleep(2L);

result.put("result3", "任务3 接口执行完毕");

} catch (InterruptedException e) {

e.printStackTrace();

} finally {

count.countDown();

}

});

count.await(); // 等待计数器归零

System.out.println(result);

System.out.println("执行时间为:" + (System.currentTimeMillis() - time));

executorService.shutdown();

}

}

CountDownLatch原理和我们之前讲的锁差不多,也可以用AQS实现(AQS请看前几节),下面我们写一个自己的CountDownLatch。

实现自己的CountDownLatch

import java.util.Iterator;

import java.util.concurrent.LinkedBlockingQueue;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.concurrent.locks.LockSupport;

public class MyCountDownLatch {

//MyAqs aqs = new MyAqs() {

//

// @Override

// public int tryAcquireShared() {

// return (this.getState().get() == 0) ? 1 : -1;

// }

//

// @Override

// public boolean tryReleaseShared() {

// return this.getState().decrementAndGet() == 0;

// }

//};

//

//public MyCountDownLatch(int count) {

// aqs.getState().set(count);

//}

//

//public void await() {

// aqs.acquireShared();

//}

//

//public void countDown() {

// aqs.releaseShared();

//}

volatile AtomicInteger count;

LinkedBlockingQueue<Thread> waiters = new LinkedBlockingQueue<>();

public MyCountDownLatch(int count) {

this.count = new AtomicInteger(count);

}

public void await() {

boolean addQ = true;

while (this.count.get() != 0) {

if (addQ) {

waiters.offer(Thread.currentThread());

addQ = false;

} else {

LockSupport.park(); // 收到 unpark 通知之后唤醒,继续循环

}

}

waiters.remove(Thread.currentThread()); // 从等待集合中移除线程

}

public void countDown() {

if (this.count.decrementAndGet() == 0) {

// 通知其他等待线程

Iterator<Thread> iterator = waiters.iterator();

while (iterator.hasNext()) {

Thread waiter = iterator.next();

LockSupport.unpark(waiter); // 唤醒线程继续

}

}

}

}

CyclicBarrier使用

CyclicBarrier又称为“线程栅栏”,创建对象时,指定栅栏线程数量;等到线程数量到达指定量时自动执行指定任务;和CountDownLatch重要区别在于,CyclicBarrier对象可多次触发执行;典型应用场景:

  1. 数据量较大,实现批量插入操作数据库
  2. 统计全年数据,12个线程统计每个月的,全部统计完后再汇总统计;

CyclicBarrier测试代码

import java.util.concurrent.CyclicBarrier;

import java.util.concurrent.LinkedBlockingQueue;

public class CyclicBarrierDemo {

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

LinkedBlockingQueue<String> sqls = new LinkedBlockingQueue<>();

// 任务1+2+3...1000 拆分为100个任务(1+..10, 11+20) -> 100线程去处理。

// 每当有4个线程处于await状态的时候,则会触发barrierAction执行

CyclicBarrier barrier = new CyclicBarrier(4, new Runnable() {

@Override

public void run() {

// 这是每满足4次数据库操作,就触发一次批量执行

System.out.println("有4个线程执行了,开始批量插入: " + Thread.currentThread().getName());

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

System.out.println(sqls.poll());

}

}

});

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

new Thread(() -> {

try {

sqls.add("data - " + Thread.currentThread().getName()); // 缓存起来

Thread.sleep(1000L); // 模拟数据库操作耗时

barrier.await(); // 等待栅栏打开,有4个线程都执行到这段代码的时候,才会继续往下执行

System.out.println(Thread.currentThread().getName() + "插入完毕");

} catch (Exception e) {

e.printStackTrace();

}

}).start();

}

Thread.sleep(2000);

}

}

Semaphore使用

Semaphore又称“信号量”,控制多个线程争抢许可;典型场景:令牌桶,代码并发处理限流。

Semaphore测试代码

import java.util.Random;

import java.util.concurrent.Semaphore;

public class SemaphoreDemo {

public static void main(String[] args) {

SemaphoreDemo semaphoreTest = new SemaphoreDemo();

int N = 9; // 客人数量

Semaphore semaphore = new Semaphore(5); // 手牌数量,限制请求数量

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

String vipNo = "vip-00" + i;

new Thread(() -> {

try {

semaphore.acquire(); // 获取令牌

semaphoreTest.service(vipNo);

semaphore.release(); // 释放令牌

} catch (InterruptedException e) {

e.printStackTrace();

}

}).start();

}

}

// 限流 控制5个线程 同时访问

public void service(String vipNo) throws InterruptedException {

System.out.println("楼上出来迎接贵宾一位,贵宾编号" + vipNo + ",...");

Thread.sleep(new Random().nextInt(3000));

System.out.println("欢送贵宾出门,贵宾编号" + vipNo);

}

}

实现自己的Semaphore

public class MySemaphore {

MyAqs aqs = new MyAqs() {

@Override

public int tryAcquireShared() { // 信号量获取, 数量 - 1

for (; ; ) {

int count = getState().get();

int n = count - 1;

if (count <= 0 || n < 0) {

return -1;

}

if (getState().compareAndSet(count, n)) {

return 1;

}

}

}

@Override

public boolean tryReleaseShared() { // 信号量释放, 数量 + 1

return this.getState().incrementAndGet() >= 0;

}

};

/**

* 设置令牌数量

*/

public MySemaphore(int count) {

aqs.getState().set(count);

}

/**

* 获取令牌

*/

public void acquire() {

aqs.acquireShared();

}

/**

* 释放令牌

*/

public void release() {

aqs.releaseShared();

}

}

Fork / Join框架使用

ForkJoinPool是ExecutorService接口的实现,它专为可以递归分解成小块的工作而设计。

Fork / Join框架将任务分配给线程池中的工作线程,充分利用多处理器的优势,提高程序性能。

使用Fork / Join框架的第一步是编写执行一部分工作的代码。类似伪代码如下:

如果(当前工作部分足够小)

​ 直接做这项工作

其他

​ 把当前工作拆分成两部分

​ 调用这两个部分并等待结果

将此代码包装在ForkJoinTask子类中,通常是RecursiveTask(可以返回结果)或RecursiveAction。

实现思路:

  1. 每个Worker线程都维护一个任务队列,即ForkJoinWorkerThread中的任务队列
  2. 任务队列是双向队列,这样可以同时实现LIFO和FIFO
  3. 子任务会被加入到原先任务所在Worker线程的任务队列
  4. Worker线程用LIFO的方法取出任务,后进队列的任务先取出来(子任务总是后加入队列,但是需要先执行)
  5. 当任务队列为空,会随即从其他的Worker的队列中拿走一个任务执行(工作窃取:steal work)
  6. 如果一个Worker线程遇到了join操作,而这时候正在处理其他任务,会等到这个任务结束。否则直接返回
  7. 如果一个Worker线程窃取任务失败,它会用yield或者sleep之类的方法休息一会儿,再尝试(如果所有线程都是空闲状态,即没有任务运行,那么该线程也会进入阻塞状态等待新任务的到来)

适用

使用尽可能少的线程池-在大多数情况下,最好的决定是为每个应用程序或系统使用一个线程池

如果不需要特定调整,请使用默认的公共线程池

使用合理的阀值将ForkJoinTask拆分为子任务

避免在ForkJoinTask中出现任何阻塞

适合数据处理、结果汇总、统计等场景;

java8实例:java.util.Arrays类用于parallelSort()方法

结语:工作窃取带来的性能提升偏理论,API的复杂性较高,实际研发中可控性来说不如其他API。

ForkJoinPool测试代码

import java.util.concurrent.*;

public class ForkJoinDemo {

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

// 本质是一个线程池,默认的线程数量:CPU的核数

// 默认情况下,并行线程数量等于可用处理器的数量

// ForkJoinPool与其他类型的ExecutorService的区别主要在于它使用了工作窃取:

// 池中的所有线程都试图查找和执行提交给池的任务和/或其他活动任务创建的任务

// (如果不存在工作,则最终阻塞等待工作)。

ForkJoinPool forkJoinPool = new ForkJoinPool();

// 适合数据处理、结果汇总、统计等场景

// 举个栗子计算1+2+3+…100W的结果,拆分多个线程进行计算,合并汇总返回

// 查询多个接口的数据,合并返回

// 其他例子,查数据库的多个表数据,分多次查询

// fork/join

// forkJoinPool.submit()

// 计算1+2+3+…100W的结果,这里只是举个例子

long time1 = System.currentTimeMillis();

// 提交可分解的RecursiveTask任务

ForkJoinTask<Long> forkJoinTask = forkJoinPool.submit(new SumTask(1, 1000000));

System.out.println("结果为:" + forkJoinTask.get() + " 执行时间" + (System.currentTimeMillis() - time1));

//forkJoinPool.shutdown();

// 实际执行Fork、Join、从队列存取等操作都需要消耗时间,请在合适的场景使用

long time2 = System.currentTimeMillis();

long sum = 0L;

for (int i = 1; i <= 1000000; i++) {

sum += i;

}

System.out.println("结果为:" + sum + " 执行时间" + (System.currentTimeMillis() - time2));

// 提交可分解的RecursiveAction任务

forkJoinPool.submit(new PrintAction(1, 1000));

//阻塞当前线程直到 ForkJoinPool 中所有的任务都执行结束

forkJoinPool.awaitTermination(2, TimeUnit.SECONDS);

forkJoinPool.shutdown();

}

}

class SumTask extends RecursiveTask<Long> {

private static final int THRESHOLD = 1000;

private int start; // 开始下标

private int end; // 结束下标

public SumTask(int start, int end) {

this.start = start;

this.end = end;

}

@Override

protected Long compute() {

// 如果当前任务的计算量在阈值范围内,则直接进行计算

if (end - start < THRESHOLD) {

return computeByUnit();

} else { // 如果当前任务的计算量超出阈值范围,则进行计算任务拆分

// 计算中间索引

int middle = (start + end) / 2;

//定义子任务-迭代思想

SumTask left = new SumTask(start, middle);

SumTask right = new SumTask(middle + 1, end);

//划分子任务-fork

left.fork();

right.fork();

//合并计算结果

return left.join() + right.join();

}

}

private long computeByUnit() {

long sum = 0L;

for (int i = start; i <= end; i++) {

sum += i;

}

return sum;

}

}

class PrintAction extends RecursiveAction {

private static final int THRESHOLD = 20;

private int start; // 开始下标

private int end; // 结束下标

public PrintAction(int start, int end) {

this.start = start;

this.end = end;

}

@Override

protected void compute() {

//当end-start的值小于MAX时,开始打印

if ((end - start) < THRESHOLD) {

computeByUnit();

} else {

// 将大任务分解成两个小任务

int middle = (start + end) / 2;

PrintAction left = new PrintAction(start, middle);

PrintAction right = new PrintAction(middle + 1, end);

left.fork();

right.fork();

}

}

private void computeByUnit() {

for (int i = start; i <= end; i++) {

System.out.println(Thread.currentThread().getName() + "的值:" + i);

}

}

}

以上是 Java学习笔记11多线程工具类实战 的全部内容, 来源链接: utcz.com/z/511195.html

回到顶部