参考:https://www.cnblogs.com/yjd_hycf_space/p/7526608.html
Java中有两种创建线程的方式:
创建方式一
- 继承Thread类(在java.lang包中),并重写该类的run()方法,其中run()方法即线程需要执行的任务代码。
- 然后new出这个类对象。这表示创建线程对象。
- 调用start()方法开启线程来执行任务(start()方法会调用run()以便执行任务)。
class MyThread extends Thread { String name; String gender; MyThread(String name,String gender){ this.name = name; this.gender = gender; } public void run(){ int i = 0; while(i<=20) { System.out.println(Thread.currentThread().getName()); i++; } } } public class CreateThread { public static void main(String[] args) { MyThread mt1 = new MyThread("张三","Male"); MyThread mt2 = new MyThread("李四","Female"); mt1.start(); mt2.start(); System.out.println("main thread over"); } }
每个线程被cpu调度是随机的,执行时间也是随机的,所以即使mt1先开启任务,但mt2可能会比mt1线程先执行,也可能更先消亡。
创建线程方式二:
- 实现Runnable接口,并重写run()方法。
- 创建子类对象。
- 创建Thread对象来创建线程对象,并将实现了Runnable接口的对象作为参数传递给Thread()构造方法。
- 调用start()方法开启线程来执行run()中的任务。
class MyThread implements Runnable { String name; String gender; MyThread(String name,String gender){ this.name = name; this.gender = gender; } public void run(){ int i = 0; while(i<=200) { System.out.println(Thread.currentThread().getName()); i++; } } } public class CreateThread2 { public static void main(String[] args) { //创建子类对象 MyThread mt = new MyThread("小张","Male"); //创建线程对象 Thread th1 = new Thread(mt); Thread th2 = new Thread(mt); th1.start(); th2.start(); System.out.println("main thread over"); } }
实现Runnable接口比继承Thread类所具有的优势:
1):适合多个相同的程序代码的线程去处理同一个资源
2):可以避免java中的单继承的限制
3):增加程序的健壮性,代码可以被多个线程共享,代码和数据独立
4):线程池只能放入实现Runable或callable类线程,不能直接放入继承Thread的类
线程常用方法:
Thread类中的方法:
isAlive()
:判断线程是否还活着。活着的概念是指是否消亡了,对于运行态、就绪态、睡眠态的线程都是活着的状态。currentThread()
:返回值为Thread,返回当前线程对象。getName()
:获取当前线程的线程名称。setName()
:设置线程名称。给线程命名还可以使用构造方法Thread(String thread_name)
或Thread(Runnable r,String thread_name)
。getPriority()
:获取线程优先级。优先级范围值为1-10(默认值为5),相邻值之间的差距对cpu调度的影响很小。分三个优先级可较大地区分cpu的调度。setPriority()
:设置线程优先级。run()
:封装的是线程开启后要执行的任务代码。如果run()中没有任何代码,则线程不做任何事情。start()
:开启线程并让线程开始执行run()中的任务。toString()
:返回线程的名称、优先级和线程组。sleep(long millis)
:让线程睡眠多少毫秒。join(t1)
:将线程t1合并到当前线程,并等待线程t1执行完毕后才继续执行当前线程。即让t1线程强制插队到当前线程的前面并等待t1完成。yield()
:将当前正在执行的线程退让出去,以让就绪队列中的其他线程有更大的几率被cpu调度。即强制自己放弃cpu,并将自己放入就绪队列。由于自己也在就绪队列中,所以即使此刻自己放弃了cpu,下一次还是可能会立即被cpu选中调度。但毕竟给了机会给其它就绪态线程,所以其他就绪态线程被选中的几率要更大一些。
Object类中的方法:
wait()
:线程进入某个线程池中并进入睡眠态。等待notify()或notifyAll()的唤醒。notify()
:从某个线程池中随机唤醒一个睡眠态的线程。notifyAll()
:唤醒某个线程池中所有的睡眠态线程。
Lock类中:
lock()
:获取锁(互斥锁)。unlock()
:释放锁。newCondition()
:创建关联此lock对象的Condition对象。
Condition类中:
await()
:和wait()一样。signal()
:和notify()一样。signalAll()
:和notifyAll()一样。
join():指等待线程终止:
join是Thread类的一个方法,启动线程后直接调用,作用是:“等待该线程终止”,该线程是指的主线程等待子线程的终止。也就是在子线程调用了join()方法后面的代码,只有等到子线程结束了才能执行,也就是主线程需要等待子线程执行完成之后再结束。
public class Demo{ public static void main(String[] args) { System.out.println(Thread.currentThread().getName()+"主线程运行开始!"); Thread1 mTh1=new Thread1("A"); Thread1 mTh2=new Thread1("B"); mTh1.start(); mTh2.start(); try { mTh1.join(); } catch (InterruptedException e) { e.printStackTrace(); } try { mTh2.join(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread().getName()+ "主线程运行结束!"); } }
常遇到问题:
1、死锁:线程全睡眠了无法被唤醒,导致程序卡死在某一处无法再执行下去。典型的是两个同步线程,线程1持有A锁,且等待B锁,但线程2持有B锁且等待A锁,这样的僵局会造成死锁。但需要注意的是,死锁并非都是因为僵局,只要两边的线程都无法继续向下执行代码(或者两边的线程池都无法被唤醒,这是等价的概念,因为锁等待也会让进程进入睡眠态),则都是死锁。
线程池:
Executors创建线程池 ,只需要调用Executors
中相应的便捷方法即可,
各个创建线程池便捷方法列表:
newFixedThreadPool(int nThreads) |
创建固定大小的线程池 |
newSingleThreadExecutor() |
创建只有一个线程的线程池 |
newCachedThreadPool() |
创建一个不限线程数上限的线程池,任何提交的任务都将立即执行 |
实际各便捷方法是调用了ThreadPoolExecutor
的构造方法(定时任务使用的是ScheduledThreadPoolExecutor
)
// Java线程池的完整构造函数 public ThreadPoolExecutor( int corePoolSize, // 线程池长期维持的线程数,即使线程处于Idle状态,也不会回收。 int maximumPoolSize, // 线程数的上限 long keepAliveTime, // 超过corePoolSize的线程的idle时长, TimeUnit unit, // 超过这个时间,多余的线程会被回收。 BlockingQueue<Runnable> workQueue, // 任务的排队队列 ThreadFactory threadFactory, // 新线程的产生方式 RejectedExecutionHandler handler) // 拒绝策略 }
Runnable和Callable
可以向线程池提交的任务有两种:Runnable
和Callable
,二者的区别如下:
- 方法签名不同,
void Runnable.run()
,V Callable.call() throws Exception
- 是否允许有返回值,
Callable
允许有返回值 - 是否允许抛出异常,
Callable
允许抛出异常。
三种提交任务的方式:
提交方式 | 是否关心返回结果 |
---|---|
Future<T> submit(Callable<T> task) |
是 |
void execute(Runnable command) |
否 |
Future<?> submit(Runnable task) |
否,虽然返回Future,但是其get()方法总是返回null |
正确使用线程池:
1、避免无界队列(OOM)
为避免OOM,我们应该自己实现ThreadPoolExecutor
的构造方法手动指定队列的最大长度
int poolSize = Runtime.getRuntime().availableProcessors() * 2;
BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(512);// 使用有界队列,避免OOM
RejectedExecutionHandler policy = new ThreadPoolExecutor.DiscardPolicy();
ExecutorService executorService = new ThreadPoolExecutor(poolSize, poolSize, 0, TimeUnit.SECONDS, queue, policy);
2、明确拒绝任务时的行为
线程池给我们提供了几种常见的拒绝策略:
拒绝策略 | 拒绝行为 |
---|---|
AbortPolicy |
抛出RejectedExecutionException |
DiscardPolicy | 什么也不做,直接忽略 |
DiscardOldestPolicy |
丢弃执行队列中最老的任务,尝试为当前提交的任务腾出位置 |
CallerRunsPolicy | 直接由提交任务者执行这个任务 |
线程池默认的拒绝行为是AbortPolicy
,也就是抛出RejectedExecutionHandler
异常,该异常是非受检异常,很容易忘记捕获。如果不关心任务被拒绝的事件,可以将拒绝策略设置成DiscardPolicy
,这样多余的任务会悄悄的被忽略。
3、获取处理结果和异常
线程池的处理结果、以及处理过程中的异常都被包装到Future
中,并在调用Future.get()
方法时获取,执行过程中的异常会被包装成ExecutionException
,submit()
方法本身不会传递结果和任务执行过程中的异常。获取执行结果的代码可以这样写:
ExecutorService executorService = Executors.newFixedThreadPool(4); Future<Object> future = executorService.submit(new Callable<Object>() { @Override public Object call() throws Exception { throw new RuntimeException("exception in call~");// 该异常会在调用Future.get()时传递给调用者 } }); try { Object result = future.get(); } catch (InterruptedException e) {
} catch (ExecutionException e) { e.printStackTrace(); }
4.1获取单个结果
通过submit()
向线程池提交任务后会返回一个Future
,调用V Future.get()
方法能够阻塞等待执行结果,该方法可以指定等待的超时时间。
4.2获取多个结果
如果向线程池提交了多个任务,要获取这些任务的执行结果,可以依次调用Future.get()
获得。但对于这种场景,我们更应该使用ExecutorCompletionService,该类的take()
方法总是阻塞等待某一个任务完成,然后返回该任务的Future
对象。向CompletionService
批量提交任务后,只需调用相同次数的CompletionService.take()
方法,就能获取所有任务的执行结果,获取顺序是任意的,取决于任务的完成顺序:
public void solve(Executor executor, Collection<Callable<Result>> solvers) throws InterruptedException, ExecutionException { CompletionService<Result> ecs = new ExecutorCompletionService<Result>(executor);// 构造器 for (Callable<Result> s : solvers)// 提交所有任务 ecs.submit(s); int n = solvers.size(); for (int i = 0; i < n; ++i) {// 获取每一个完成的任务 Result r = ecs.take().get(); if (r != null) use(r); } }
5.1单个任务的超时时间
V Future.get(long timeout, TimeUnit unit)
方法可以指定等待的超时时间,超时未完成会抛出TimeoutException
。
5.2多个任务的超时时间
等待多个任务完成,并设置最大等待时间,可以通过CountDownLatch完成:
public void testLatch(ExecutorService executorService, List<Runnable> tasks) throws InterruptedException{ CountDownLatch latch = new CountDownLatch(tasks.size()); for(Runnable r : tasks){ executorService.submit(new Runnable() { @Override public void run() { try{ r.run(); }finally { latch.countDown();// countDown } } }); } latch.await(10, TimeUnit.SECONDS); // 指定超时时间 }
实例一 :多线程统计
import java.text.SimpleDateFormat; import java.util.Date; import java.util.concurrent.CountDownLatch; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; /** * 多任务并行+线程池统计 */ public class StatsDemo { final static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); final static String startTime = sdf.format(new Date()); /** * IO密集型任务 = 一般为2*CPU核心数(常出现于线程中:数据库数据交互、文件上传下载、网络数据传输等等) * CPU密集型任务 = 一般为CPU核心数+1(常出现于线程中:复杂算法) * 混合型任务 = 视机器配置和复杂度自测而定 */ private static int corePoolSize = Runtime.getRuntime().availableProcessors(); /** * public ThreadPoolExecutor( * int corePoolSize, //用于指定核心线程数量 * int maximumPoolSize, //指定最大线程数 * long keepAliveTime, //指定线程空闲后的最大存活时间 * TimeUnit unit, //指定线程空闲后的最大存活时间 * BlockingQueue<Runnable> workQueue //则是线程池的缓冲队列,还未执行的线程会在队列中等待 监控队列长度,确保队列有界 * ) * 不当的线程池大小会使得处理速度变慢,稳定性下降,并且导致内存泄露。如果配置的线程过少,则队列会持续变大,消耗过多内存。 * 而过多的线程又会 由于频繁的上下文切换导致整个系统的速度变缓——殊途而同归。 * 队列的长度至关重要,它必须得是有界的,这样如果线程池不堪重负了它可以暂时拒绝掉新的请求。 * ExecutorService 默认的实现是一个无界的 * LinkedBlockingQueue。 */ private static ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, corePoolSize+1, 10l, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(1000)); public static void main(String[] args) throws InterruptedException { CountDownLatch latch = new CountDownLatch(5); //使用execute方法 executor.execute(new Stats("A", 1000, latch)); executor.execute(new Stats("B", 1000, latch)); executor.execute(new Stats("C", 1000, latch)); executor.execute(new Stats("D", 1000, latch)); executor.execute(new Stats("E", 1000, latch)); latch.await();// 等待所有人任务结束 System.out.println("所有的统计任务执行完成:" + sdf.format(new Date())); } static class Stats implements Runnable { String statsName; int runTime; CountDownLatch latch; public Stats(String statsName, int runTime, CountDownLatch latch) { this.statsName = statsName; this.runTime = runTime; this.latch = latch; } public void run() { try { //模拟任务执行时间 Thread.sleep(runTime); latch.countDown();//单次任务结束,计数器减一 } catch (InterruptedException e) { e.printStackTrace(); } } } }
Future的案例
import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.concurrent.*; /** * Future的使用 */ public class StatsDemo { final static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); final static String startTime = sdf.format(new Date()); private static int corePoolSize = Runtime.getRuntime().availableProcessors(); private static ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, corePoolSize+1, 10l, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(1000)); public static void main(String[] args) throws InterruptedException { List<Future<String>> resultList = new ArrayList<Future<String>>(); //使用submit提交异步任务,并且获取返回值为future resultList.add(executor.submit(new Stats("A", 100))); resultList.add(executor.submit(new Stats("B", 500))); resultList.add(executor.submit(new Stats("C", 100))); resultList.add(executor.submit(new Stats("D", 1000))); resultList.add(executor.submit(new Stats("E", 100))); //遍历任务的结果 for (Future<String> fs : resultList) { try { System.out.println(fs.get());//打印各个线任务执行的结果,调用future.get() 阻塞主线程,获取异步任务的返回结果 } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } finally { //启动一次顺序关闭,执行以前提交的任务,但不接受新任务。如果已经关闭,则调用没有其他作用。 executor.shutdown(); } } System.out.println("所有任务执行完成:" + sdf.format(new Date())); } static class Stats implements Callable<String> { String statsName; int runTime; public Stats(String statsName, int runTime) { this.statsName = statsName; this.runTime = runTime; } public String call() { try { //模拟任务执行时间 Thread.sleep(runTime); } catch (InterruptedException e) { e.printStackTrace(); } return call(); } }