多线程&&线程池
多线程
多线程的实现方式主要有三种
- extends Thread
- implements Runnable
- implements Callable
实现接口 VS 继承 Thread
实现接口会更好一些,因为:
- Java 不支持多重继承,因此继承了 Thread 类就无法继承其它类,但是可以实现多个接口;
- 类可能只要求可执行就行,继承整个 Thread 类开销过大。
继承 Thread 的代码
public class ExtendsThread extends Thread {
/**
* 线程名称
*/
private String threadName;
/**
* 构造函数
*
* @param threadName 线程名称
*/
public ExtendsThread(String threadName) {
this.threadName = threadName;
}
/**
* 重写run方式
*/
@Override
public void run() {
System.out.println(threadName);
}
}
public class ThreadMain {
public static void main(String[] args) {
ExtendsThread extendsThread = new ExtendsThread("1");
extendsThread.start();
extendsThread = new ExtendsThread("2");
extendsThread.start();
}
}
实现Runnable接口
public class ThreadMain {
public static void main(String[] args) {
ThreadImplRunnable threadImplRunnable = new ThreadImplRunnable("Thread1");
new Thread(threadImplRunnable).start();
threadImplRunnable = new ThreadImplRunnable("Thread2");
new Thread(threadImplRunnable).start();
System.out.println("end");
}
}
public class ThreadImplRunnable implements Runnable {
/**
* 线程名称
*/
private String threadName;
/**
* 构造函数
*
* @param threadName 线程名称
*/
public ThreadImplRunnable(String threadName) {
this.threadName = threadName;
}
/**
* 重写run方法
*/
@Override
public void run() {
System.out.println(threadName);
}
}
end
Thread1
Thread2
实现Callable接口主要是可以获取到返回值,futureTask.get() 获取返回的值
public class ThreadImplCallable<T> implements Callable<T> {
/**
* 线程名称
*/
private String threadName;
/**
* 构造函数
*
* @param threadName 线程名称
*/
public ThreadImplCallable(String threadName) {
this.threadName = threadName;
}
/**
* 重写call方法
*/
@Override
public T call() throws Exception {
System.out.println(threadName);
return (T)threadName;
}
}
public class ThreadMain {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Callable<String> threadImplCallable = new ThreadImplCallable<>("Thread1");
FutureTask<String> futureTask = new FutureTask<>(threadImplCallable);
Thread thread = new Thread(futureTask);
thread.start();
String rValue = futureTask.get();
System.out.println("Thread1 return value is " + rValue);
threadImplCallable = new ThreadImplCallable<>("Thread2");
futureTask = new FutureTask<>(threadImplCallable);
thread = new Thread(futureTask);
thread.start();
rValue = futureTask.get();
System.out.println("Thread2 return value is " + rValue);
System.out.println("end");
}
}
Thread1
Thread1 return value is Thread1
Thread2
Thread2 return value is Thread2
end
线程池
java 线程池的顶级接口是 Executor, 只定义了一个方法 execute,ExecutorService继承接口Executor接口,并定义了其他的相关的接口,ExecutorService下面有一个继承的抽象类AbstractExecutorService和实现了定时调度的接口ScheduledExecutorService,而又有两个类继承抽象类,分别是ThreadPoolExecutor和ForkJoinPool,而ScheduledThreadPoolExecutor则实现接口实现ScheduledExecutorService接口和继承ThreadPoolExecutor类,基本的关系就是这样的,而且基本常用的封装方法都在Executors类中,只要 Executors.xxxx 就可以应用对应的封装的方法。
Executor
Executor 管理多个异步任务的执行,而无需程序员显式地管理线程的生命周期。这里的异步是指多个任务的执行互不干扰,不需要进行同步操作。
线程池种类
一个单线程的线程池。这个线程池只有一个线程在工作,也就是相当于单线程串行执行所有任务。如果这个唯一的线程因为异常结束,那么会有一个新的线程来替代它。此线程池保证所有任务的执行顺序按照任务的提交顺序执行。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPool {
public static void main(String[] args) {
ExecutorService pool = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
pool.execute(() -> {
System.out.println(Thread.currentThread().getName() + "\t开始发车啦....");
});
}
}
}
pool-1-thread-1 开始发车啦....
pool-1-thread-1 开始发车啦....
pool-1-thread-1 开始发车啦....
pool-1-thread-1 开始发车啦....
pool-1-thread-1 开始发车啦....
pool-1-thread-1 开始发车啦....
pool-1-thread-1 开始发车啦....
pool-1-thread-1 开始发车啦....
pool-1-thread-1 开始发车啦....
pool-1-thread-1 开始发车啦....
创建固定大小的线程池。每次提交一个任务就创建一个线程,直到线程达到线程池的最大大小。线程池的大小一旦达到最大值就会保持不变,如果某个线程因为执行异常而结束,那么线程池会补充一个新线程。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPool {
public static void main(String[] args) {
ExecutorService pool = Executors.newFixedThreadPool(10);
for (int i = 0; i < 10; i++) {
pool.execute(() -> {
System.out.println(Thread.currentThread().getName() + "\t开始发车啦....");
});
}
}
}
pool-1-thread-1 开始发车啦....
pool-1-thread-4 开始发车啦....
pool-1-thread-3 开始发车啦....
pool-1-thread-2 开始发车啦....
pool-1-thread-6 开始发车啦....
pool-1-thread-7 开始发车啦....
pool-1-thread-5 开始发车啦....
pool-1-thread-8 开始发车啦....
pool-1-thread-9 开始发车啦....
pool-1-thread-10 开始发车啦....
创建一个可缓存的线程池。如果线程池的大小超过了处理任务所需要的线程,那么就会回收部分空闲的线程,当任务数增加时,此线程池又添加新线程来处理任务。
public static void main(String[] args) {
ExecutorService pool = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
pool.execute(() -> {
System.out.println(Thread.currentThread().getName() + "\t开始发车啦....");
});
}
}
pool-1-thread-1 开始发车啦....
pool-1-thread-2 开始发车啦....
pool-1-thread-2 开始发车啦....
pool-1-thread-3 开始发车啦....
pool-1-thread-3 开始发车啦....
pool-1-thread-1 开始发车啦....
pool-1-thread-7 开始发车啦....
pool-1-thread-5 开始发车啦....
pool-1-thread-6 开始发车啦....
pool-1-thread-4 开始发车啦....
此线程池支持定时以及周期性执行任务的需求。
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class ThreadPool {
public static void main(String[] args) {
ScheduledExecutorService pool = Executors.newScheduledThreadPool(10);
for (int i = 0; i < 10; i++) {
pool.schedule(() -> {
System.out.println(Thread.currentThread().getName() + "\t开始发车啦....");
}, 10, TimeUnit.SECONDS);
}
}
}
pool-1-thread-1 开始发车啦....
pool-1-thread-2 开始发车啦....
pool-1-thread-1 开始发车啦....
pool-1-thread-3 开始发车啦....
pool-1-thread-7 开始发车啦....
pool-1-thread-8 开始发车啦....
pool-1-thread-6 开始发车啦....
pool-1-thread-5 开始发车啦....
pool-1-thread-2 开始发车啦....
pool-1-thread-4 开始发车啦....
上面演示的是延迟10秒执行任务,如果想要执行周期性的任务可以用下面的方式,每秒执行一次。
public static void main(String[] args) {
ScheduledExecutorService pool = Executors.newScheduledThreadPool(10);
//pool.scheduleWithFixedDelay也可以
pool.scheduleAtFixedRate(() -> {
System.out.println(Thread.currentThread().getName() + "\t开始发车啦....");
}, 1, 1, TimeUnit.SECONDS);
}
pool-1-thread-1 开始发车啦....
pool-1-thread-1 开始发车啦....
pool-1-thread-2 开始发车啦....
pool-1-thread-1 开始发车啦....
pool-1-thread-3 开始发车啦....
pool-1-thread-2 开始发车啦....
pool-1-thread-4 开始发车啦....
pool-1-thread-1 开始发车啦....
pool-1-thread-5 开始发车啦....
pool-1-thread-3 开始发车啦....
pool-1-thread-3 开始发车啦....
pool-1-thread-3 开始发车啦....
pool-1-thread-3 开始发车啦....
pool-1-thread-3 开始发车啦....
pool-1-thread-3 开始发车啦....
pool-1-thread-3 开始发车啦....
pool-1-thread-9 开始发车啦....
pool-1-thread-9 开始发车啦....
newWorkStealingPool是jdk1.8才有的,会根据所需的并行层次来动态创建和关闭线程,通过使用多个队列减少竞争,底层用的ForkJoinPool来实现的。ForkJoinPool的优势在于,可以充分利用多cpu,多核cpu的优势,把一个任务拆分成多个“小任务”,把多个“小任务”放到多个处理器核心上并行执行;当多个“小任务”执行完成之后,再将这些执行结果合并起来即可。
newWorkStealingPool适合使用在很耗时的操作,但是newWorkStealingPool不是ThreadPoolExecutor的扩展,它是新的线程池类ForkJoinPool的扩展,但是都是在统一的一个Executors类中实现,由于能够合理的使用CPU进行对任务操作(并行操作),所以适合使用在很耗时的任务中
package org.jeecg.modules.exam.entity;
import org.junit.Test;
import java.util.concurrent.*;
public class WorkStealingPoolTest {
// 线程数
private static final int threads = 10;
// 用于计数线程是否执行完成
CountDownLatch countDownLatch = new CountDownLatch(threads);
/**
* newFixedThreadPool execute
*
* @throws ExecutionException
* @throws InterruptedException
*/
@Test
public void test1() throws ExecutionException, InterruptedException {
System.out.println("---- start ----");
ExecutorService executorService = Executors.newWorkStealingPool();
for (int i = 0; i < threads; i++) {
executorService.execute(() -> {
try {
System.out.println(Thread.currentThread().getName());
} catch (Exception e) {
System.out.println(e);
} finally {
countDownLatch.countDown();
}
});
}
countDownLatch.await();
System.out.println("---- end ----");
}
/**
* newFixedThreadPool submit submit
*/
@Test
public void test2() throws InterruptedException {
System.out.println("---- start ----");
ExecutorService executorService = Executors.newWorkStealingPool();
for (int i = 0; i < threads; i++) {
// Callable 带返回值
executorService.submit(new Thread(new Runnable() {
@Override
public void run() {
try {
System.out.println(Thread.currentThread().getName());
} catch (Exception e) {
e.printStackTrace();
} finally {
countDownLatch.countDown();
}
}
}));
}
countDownLatch.await();
System.out.println("---- end ----");
}
/**
* newFixedThreadPool submit Callable
*
* @throws ExecutionException
* @throws InterruptedException
*/
@Test
public void test3() throws ExecutionException, InterruptedException {
System.out.println("---- start ----");
ExecutorService executorService = Executors.newWorkStealingPool();
for (int i = 0; i < threads; i++) {
// Runnable 带返回值
FutureTask<?> futureTask = new FutureTask<>(new Callable<String>() {
/**
* call
* @return currentThreadName
*/
@Override
public String call() {
return Thread.currentThread().getName();
}
});
executorService.submit(new Thread(futureTask));
System.out.println(futureTask.get());
}
System.out.println("---- end ----");
}
}
还没有评论,来说两句吧...