Java多线程之线程池、前后台线程、发令枪、信号量、Future

在 上张贴 由 admin发表回复

一、Java线程池使用

创建线程是一个耗资源的过程。创建大连线程会导致系统内存溢出。

所以为了更合理使用线程，通常我们使用线程池在解决

1、简单使用

jdk提供一个工具类Executors可以方便创建线程池

//核心线程0，最新线程Integer.MAX_VALUE，空闲线程存活时间60s ExecutorService threadPool1=Executors.newCachedThreadPool(); //核心线程10，最大线程10，空闲线程存活时间0 ExecutorService threadPool2=Executors.newFixedThreadPool(10); //核心线程1，最大线程1，空闲线程存活时间0 ExecutorService threadPool3=Executors.newSingleThreadExecutor();

2、自定义线程池

自定义线程池重点在于自定义以下几个参数

  • 线程池名称
  • 初始值
  • 最大值
  • 空闲线程保留时间

1）使用jdk原生代码

ThreadFactory namedThreadFactory = new ThreadFactory() { private final AtomicInteger threadNumber = new AtomicInteger(1); @Override public Thread newThread(Runnable r) { Thread t = new Thread(Thread.currentThread().getThreadGroup(), r, "threadPoolDemo_" + threadNumber.getAndIncrement(), 0); if (t.isDaemon()) t.setDaemon(false); if (t.getPriority() != Thread.NORM_PRIORITY) t.setPriority(Thread.NORM_PRIORITY); return t; } }; ExecutorService threadPool = new ThreadPoolExecutor(0, 100, 60L, TimeUnit.MILLISECONDS, new SynchronousQueue<>(), namedThreadFactory);

上面这段代码表示最大只能100个线程，超过就会拒绝。 正常情况，如果使用SynchronousQueue，那么maximumPoolSize就要设置到足够大

ThreadFactory namedThreadFactory = new ThreadFactory() { private final AtomicInteger threadNumber = new AtomicInteger(1); @Override public Thread newThread(Runnable r) { Thread t = new Thread(Thread.currentThread().getThreadGroup(), r, "threadPoolDemo_" + threadNumber.getAndIncrement(), 0); if (t.isDaemon()) t.setDaemon(false); if (t.getPriority() != Thread.NORM_PRIORITY) t.setPriority(Thread.NORM_PRIORITY); return t; } }; ExecutorService threadPool = new ThreadPoolExecutor(10, 10, 60L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), namedThreadFactory);

上面这段代码表示最大只能10个线程，超过部分会放在队列中等待执行

2）使用google工具类

import com.google.common.util.concurrent.ThreadFactoryBuilder; public void initThreadPool() { ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("loopPrecreate-pool-%d").build(); ExecutorService threadPool = new ThreadPoolExecutor(5, 50, 60L, TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>(), namedThreadFactory); }

上面这段代码表示最大只能50个线程，超过就会拒绝

3）注意：

  1. 如果使用SynchronousQueue，那么maximumPoolSize就要设置到足够大，并发请求时maximumPoolSize会逐渐增加到maximumPoolSize，超过max会拒绝
  2. 如果使用LinkedBlockingQueue，那么corePoolSize和maximumPoolSize要设置成一样，maximumPoolSize不会自动扩展

3、使用示例

并发100个线程池

int N = 100; CountDownLatch latch = new CountDownLatch(N); CountDownLatch lock = new CountDownLatch(1); for (int i = 0; i < N; i++) { Thread.sleep(100L); threadPool.submit(()->{ try { lock.await(); System.out.println("---> start " + Thread.currentThread().getName()); Thread.sleep(5000L); System.out.println("---> end " + Thread.currentThread().getName()); latch.countDown(); } catch (Exception e) { e.printStackTrace(); } }); } lock.countDown(); latch.await(); System.out.println("--end all ---");

二、CountDownLatch 介绍

类似发令枪，CountDownLatch的构造函数可以传入一个数字，代表倒计时多少个数。

主要方法：

  • countDown：每调用一次countDown()，倒计时就会减少1，
  • await：当倒计时到0时，await()方法就会停止阻塞

1、示例代码

public static void main(String[] args) throws InterruptedException { int N = 10; CountDownLatch latch = new CountDownLatch(N); CountDownLatch lock = new CountDownLatch(1); for (int i = 0; i < N; i++) { Thread.sleep(100L); new Thread(() -> { try { lock.await(); System.out.println("---> start " + Thread.currentThread().getName()); Thread.sleep(1000L); System.out.println("---> end " + Thread.currentThread().getName()); latch.countDown(); } catch (Exception e) { e.printStackTrace(); } }).start(); } lock.countDown(); latch.await(); }

三、什么是Future

Future是一个未来对象，里面保存这线程处理结果，它像一个提货凭证，拿着它你可以随时去提取结果

1、什么时候使用

在两种情况下，离开Future几乎很难办。

  • 一种情况是拆分订单，比如你的应用收到一个批量订单，此时如果要求最快的处理订单，那么需要并发处理，并发的结果如何收集，这个问题如果自己去编程将非常繁琐，此时可以使用CompletionService解决这个问题。CompletionService将Future收集到一个队列里，可以按结果处理完成的先后顺序进队。
  • 另外一种情况是，如果你需要并发去查询一些东西（比如爬虫），并发查询只要有一个结果返回，你就认为查询到了，并且结束查询，这时也需要用CompletionService和Future来解决。

使用过程既可以用CompletionService，也可以自己维护一个list将Future添加进去。

区别是：CompletionService是先完成的先返回，自己维护的list就是按顺序一个个取值

示例代码：

public class FutureDemo { private static class Task implements Callable<String> { @Override public String call() throws Exception { // 模拟真实事务的处理过程，这个过程是非常耗时的。 Thread.sleep(5000); return "call return " + Thread.currentThread().getName(); } } public static void main(String[] args) throws Exception { FutureDemo demo =new FutureDemo(); demo.testComplete(); // demo.testFutrue(); } private void testFutrue() throws Exception{ List<Future<String>> futures = new ArrayList<Future<String>>(); ExecutorService executorService = Executors.newCachedThreadPool(); System.out.println("已经提交资源申请"); for (int i = 0; i < 10; i++) { futures.add(executorService.submit(new Task())); } for (Future<String> future : futures) { if (!future.isDone()) { System.out.println("资源还没有准备好"); } System.out.println(future.get()); } executorService.shutdown(); } private void testComplete() throws Exception { ExecutorService executorService = Executors.newCachedThreadPool(); CompletionService<String> ecs = new ExecutorCompletionService<>(executorService); System.out.println("已经提交资源申请"); int lenth =10; for (int i = 0; i < lenth; i++) { ecs.submit(new Task()); } //CompletionService会按处理完后顺序返回结果 List<String> res =new ArrayList<>(); for(int i = 0;i<lenth;i++ ){ Future<String> f = ecs.take(); System.out.println(f.get()); } executorService.shutdown(); } }

四、Java前台线程和后台线程

1、前台线程

定义：普通线程又可以称为用户线程，只完成用户自己想要完成的任务，不提供公共服务。

通过new Thread()产生的线程默认都是前台线程。 前台线程在程序执行完成后，才会自己退出，不会随着主线程的退出而退出。假设main线程已经退出了，但是该线程还未执行完成，还会继续执行直到线程自己退出。

java线程池启动的是前台线程

2、后台线程

定义：指在程序运行的时候在后台提供一种通用服务的线程，并且这种线程并不属于程序中不可或缺的部分。

后台线程又称守护线程，在前台线程全部退出后，后台线程会随着一起退出，并且不会执行未执行的finally语句。

在线程start之前通过setDaemon(true)方法可以将当前线程设置为后台线程

Thread t1 =new Thread(()->{ try{ System.out.println(Thread.currentThread().getName()); //如果前台线程全部退出，后台线程会一起退出，不会一直休眠下去 Thread.sleep(10000); } catch (Exception e) { e.printStackTrace(); } finally { System.out.println(Thread.currentThread().getName()+"out"); } }); t1.setDaemon(true);//设置为后台线程 t1.start();

3、使用场景

  • jvm的垃圾回收器其实就是一个后台线程
  • 和主业务无关的服务线程

五、信号量 Semaphore 重点语法

  • availablePermits()：判断是否有权限
  • acquire()：获得一个信号，阻塞线程，可被中断
  • release()：释放一个信号
  • acquireUninterruptibly()：获得一个信号，阻塞线程，不可被中断

代码示例

1、每次只允许2个线程执行

public class SemaphoreThread { public static void main(String[] args) { Semaphore semaphore = new Semaphore(2); // 建立一个缓存线程池 ExecutorService es = Executors.newCachedThreadPool(); for (int i = 0; i < 10; i++) { es.submit(new Thread(() -> { String name = Thread.currentThread().getName(); try { semaphore.acquire(); System.out.println(LocalDateTime.now().toLocalTime()+" "+name + "开始执行"); Thread.sleep(1000); semaphore.release(); } catch (InterruptedException e) { e.printStackTrace(); } })); } // 关闭线程池 es.shutdown(); } }

输出：

13:42:11.763 pool-1-thread-2开始执行 13:42:11.763 pool-1-thread-1开始执行 13:42:12.764 pool-1-thread-3开始执行 13:42:12.764 pool-1-thread-5开始执行 13:42:13.764 pool-1-thread-6开始执行 13:42:13.764 pool-1-thread-4开始执行 13:42:14.764 pool-1-thread-9开始执行 13:42:14.764 pool-1-thread-10开始执行 13:42:15.765 pool-1-thread-7开始执行 13:42:15.765 pool-1-thread-8开始执行

从输出日志可以看出，每秒只有2个线程在执行

2、控制两个线程执行顺序

public class SemaphoreThread { public static void main(String[] args) { // 设置公平锁 Semaphore semaphore = new Semaphore(1,true); // 建立一个缓存线程池 ExecutorService es = Executors.newCachedThreadPool(); es.submit( new Thread(() -> { String name = Thread.currentThread().getName(); for (int i = 0; i < 10; i++) { try { semaphore.acquire(); System.out.println(LocalDateTime.now().toLocalTime()+" "+name + "正在工作"); Thread.sleep(1000); semaphore.release(); } catch (InterruptedException e) { e.printStackTrace(); } } })); es.submit( new Thread(() -> { String name = Thread.currentThread().getName(); for (int i = 0; i < 10; i++) { try { semaphore.acquire(); System.out.println(LocalDateTime.now().toLocalTime()+" "+name + "正在工作"); Thread.sleep(100); semaphore.release(); } catch (InterruptedException e) { e.printStackTrace(); } } })); // 关闭线程池 es.shutdown(); } }

输出：

13:45:11.661 pool-1-thread-1正在工作 13:45:12.662 pool-1-thread-2正在工作 13:45:12.763 pool-1-thread-1正在工作 13:45:13.763 pool-1-thread-2正在工作 13:45:13.863 pool-1-thread-1正在工作 13:45:14.864 pool-1-thread-2正在工作 13:45:14.965 pool-1-thread-1正在工作 13:45:15.965 pool-1-thread-2正在工作 13:45:16.066 pool-1-thread-1正在工作 13:45:17.067 pool-1-thread-2正在工作 13:45:17.167 pool-1-thread-1正在工作 13:45:18.168 pool-1-thread-2正在工作 13:45:18.268 pool-1-thread-1正在工作 13:45:19.268 pool-1-thread-2正在工作 13:45:19.368 pool-1-thread-1正在工作 13:45:20.369 pool-1-thread-2正在工作 13:45:20.470 pool-1-thread-1正在工作 13:45:21.470 pool-1-thread-2正在工作 13:45:21.570 pool-1-thread-1正在工作 13:45:22.571 pool-1-thread-2正在工作

从输出日志可以看出，两个线程交替执行，这里的关键是设置了信号量公平锁

发表回复

您的邮箱地址不会被公开。必填项已用 * 标注