提供了编程的基础技术教程

zl程序教程

您现在的位置是：首页 >  大数据

当前栏目

【并发编程】Executor框架

并发框架编程 Executor
2023-06-13 09:13:57 时间

为了更好的控制多线程 jdk提供了一套线程框架Executor 帮助开发人员有效地进行线程控制 他们都在java.util.concurrent包中 是jdk并发包的核心 一个重要的类Executors 他扮演线程工厂的角色 通过Executors创建特定功能的线程池 Executors创建线程池方法：

  1. newFixedThreadPool() 方法： 该方法返回一个固定数量的线程池 该方法的线程数始终不变 当有一个任务提交时 若线程池中空闲 则立即执行 若没有 则会被暂存在一个任务队列中等待有空闲的线程去执行
  2. newSingleThreadExecutor()方法：创建一个线程的线程池 若空闲则执行 若没有空闲线程则暂缓在任务队列中。
  3. newCachedThreadPool() 方法：返回一个可根据实际情况调整线程个数的线程池 不限制最大线程数量 若用空闲的线程则执行任务 若无任务则不创建线程 并且每一个空闲线程会在60秒后自动收回。
  4. newScheduledThreadPool()方法： 该方法返回一个SchededExecutorService对象 但该线程池可以指定线程的数量。

newScheduledThreadPool()实例：

import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

class Temp extends Thread {
    public void run() {
        System.out.println("run");
    }
}

public class ScheduledJob {
	
    public static void main(String args[]) throws Exception {
    
    	Temp command = new Temp();
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        
        ScheduledFuture<?> scheduleTask = scheduler.scheduleWithFixedDelay(command, 5, 1, TimeUnit.SECONDS);
    
    }
}

这个程序意思是5秒后执行 每一秒输出一次run 类似于自动任务

如果Executors工厂类无法满足我们的需求 可以自己去创建自定义的线程池 其实Executors工厂类里面的创建线程方法其内部实现均是用了ThreadPoolExecutor这个类 这个类可以自定义线程

构造函数：

 /**
   
     * @param corePoolSize 核心线程数
     * @param maximumPoolSize 线程池允许的最大线程池数
     * @param keepAliveTime 线程等待存活时间
     * @param unit 存活时间的单位
     * @param workQueue 线程等待队列
     * @param handler 线程阻塞后的策略 
     */
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              RejectedExecutionHandler handler) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), handler);
    }

构造方法中的队列是什么类型比较关键

使用有界队列： ArrayBlockingQueue 若有新的任务需要执行 如果线程池实际线程数小于corePoolSize 则优先创建线程 若大于corePoolSize 则会将任务加入队列 若队列已满则总线程不大于maximumPoolSize的前提下 创建新的线程 若线程数大于maximumPoolSize 则执行拒绝策略 或其他自定义拒绝策略

无界的任务队列： LinkedBlockingQueue 与有界队列相比 除非系统资源耗尽 否则无界的任务队列不存在任务入队失败的情况 当有新任务到来 系统的线程数小于corePoolSize时 则新建线程执行任务 当达到corePoolSize 后 就不会继续增加线程 若后续仍有新的任务加入 而又没有空闲的线程资源 则任务直接进入队列等待 若任务创建和处理速度差异很大 无界队列会保持快速增长 直到耗尽系统内存。

有界队列 自定义拒绝策略示例：

public class MyTask implements Runnable {

	private int taskId;
	private String taskName;
	
	public MyTask(int taskId, String taskName){
		this.taskId = taskId;
		this.taskName = taskName;
	}
	
	public int getTaskId() {
		return taskId;
	}

	public void setTaskId(int taskId) {
		this.taskId = taskId;
	}

	public String getTaskName() {
		return taskName;
	}

	public void setTaskName(String taskName) {
		this.taskName = taskName;
	}

	@Override
	public void run() {
		try {
			System.out.println("run taskId =" + this.taskId);
			Thread.sleep(5*1000);
			//System.out.println("end taskId =" + this.taskId);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}		
	}
	
	public String toString(){
		return Integer.toString(this.taskId);
	}

}

自定义拒绝策略类：

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

public class MyRejected implements RejectedExecutionHandler{

	
	public MyRejected(){
	}
	
	@Override
	public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
		System.out.println("自定义处理..");
		System.out.println("当前被拒绝任务为：" + r.toString());
		

	}

}

UseThreadPoolExecutor1 类：

package com.bjsxt.height.concurrent018;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;



public class UseThreadPoolExecutor1 {


	public static void main(String[] args) {
		/**
		 * 在使用有界队列时，若有新的任务需要执行，如果线程池实际线程数小于corePoolSize，则优先创建线程，
		 * 若大于corePoolSize，则会将任务加入队列，
		 * 若队列已满，则在总线程数不大于maximumPoolSize的前提下，创建新的线程，
		 * 若线程数大于maximumPoolSize，则执行拒绝策略。或其他自定义方式。
		 * 
		 */	
		ThreadPoolExecutor pool = new ThreadPoolExecutor(
				1, 				//coreSize
				2, 				//MaxSize
				60, 			//60
				TimeUnit.SECONDS, 
				new ArrayBlockingQueue<Runnable>(3)			//指定一种队列 （有界队列）
				//new LinkedBlockingQueue<Runnable>()
				, new MyRejected()
				//, new DiscardOldestPolicy()
				);
		
		MyTask mt1 = new MyTask(1, "任务1");
		MyTask mt2 = new MyTask(2, "任务2");
		MyTask mt3 = new MyTask(3, "任务3");
		MyTask mt4 = new MyTask(4, "任务4");
		MyTask mt5 = new MyTask(5, "任务5");
		MyTask mt6 = new MyTask(6, "任务6");
		
		pool.execute(mt1);
		pool.execute(mt2);
		pool.execute(mt3);
		pool.execute(mt4);
		pool.execute(mt5);
		pool.execute(mt6);
		
		pool.shutdown();
		
	}
}

运行结果：

可以看出 一共6个任务 超过最大线程数 触发自定义拒绝策略

为什么1和5先执行 234在后面的 因为核心线程数是1 1进入后调用线程执行 234 就被放入队列中 5进来队列已满 当前线程数未超过最大线程数：2个 于是创建新线程执行5 任务6进来直接超过了最大线程数 触发自定义拒绝策略 1和5执行完后再执行234

相关文章