您的位置: 首页  >  文章  >  并发编程--线程池原理

并发编程--线程池原理

分类: 文章 • 2024-11-23 16:22:16

并发编程--线程池原理

阻塞队列和非阻塞队列

 

并发编程--线程池原理

ConcurrentLinkedQueue类

适用于高并发场景下的队列，通过无锁的方式，实现了高并发状态下的高性能，通常ConcurrentLinkedQueue性能好于BlockingQueue.它是一个基于链接节点的无界线程安全队列。该队列的元素遵循先进先出的原则。头是最先加入的，尾是最近加入的，该队列不允许null元素。

方法：

add 和offer() 都是加入元素的方法(在ConcurrentLinkedQueue中这俩个方法没有任何区别)

poll() 和peek() 都是取头元素节点，区别在于前者会删除元素，后者不会。

 

ConcurrentLinkedQueue<String> que = new ConcurrentLinkedQueue<>();
que.offer("A1");
que.offer("B1");
que.offer("C1");

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

System.out.println(que.size());

结果：

并发编程--线程池原理

BlockingQueue接口

阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。这两个附加的操作是：

在队列为空时，获取元素的线程会等待队列变为非空。

当队列满时，存储元素的线程会等待队列可用。

 

 

阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元素的线程，消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器，而消费者也只从容器里拿元素。

BlockingQueue即阻塞队列，从阻塞这个词可以看出，在某些情况下对阻塞队列的访问可能会造成阻塞。被阻塞的情况主要有如下两种：

1. 当队列满了的时候进行入队列操作

2. 当队列空了的时候进行出队列操作

因此，当一个线程试图对一个已经满了的队列进行入队列操作时，它将会被阻塞，除非有另一个线程做了出队列操作；同样，当一个线程试图对一个空队列进行出队列操作时，它将会被阻塞，除非有另一个线程进行了入队列操作。

在Concurrent包中，BlockingQueue很好的解决了多线程中，如何高效安全“传输”数据的问题。通过这些高效并且线程安全的队列类，为我们快速搭建高质量的多线程程序带来极大的便利。下文详细介绍BlockingQueue家庭中的所有成员，包括他们各自的功能以及常见使用场景：

1、ArrayBlockingQueue

ArrayBlockingQueue是一个有边界的阻塞队列，它的内部实现是一个数组。有边界的意思是它的容量是有限的，我们必须在其初始化的时候指定它的容量大小，容量大小一旦指定就不可改变。

ArrayBlockingQueue是以先进先出的方式存储数据，最新插入的对象是尾部，最新移出的对象是头部。

2、LinkedBlockingQueue

LinkedBlockingQueue阻塞队列大小的配置是可选的，如果我们初始化时指定一个大小，它就是有边界的，如果不指定，它就是无边界的。说是无边界，其实是采用了默认大小为Integer.MAX_VALUE的容量 。它的内部实现是一个链表。

LinkedBlockingQueue 也是以先进先出的方式存储数据，最新插入的对象是尾部，最新移出的对象是头部。

3、PriorityBlockingQueue

PriorityBlockingQueue是一个没有边界的队列，它的排序规则和 java.util.PriorityQueue一样。需要注意，PriorityBlockingQueue中允许插入null对象。

所有插入PriorityBlockingQueue的对象必须实现 java.lang.Comparable接口，队列优先级的排序规则就是按照我们对这个接口的实现来定义的。另外，我们可以从PriorityBlockingQueue获得一个迭代器Iterator，但这个迭代器并不保证按照优先级顺序进行迭代。

4、SynchronousQueue

SynchronousQueue队列内部仅允许容纳一个元素。当一个线程插入一个元素后会被阻塞，除非这个元素被另一个线程消费。

线程池

线程池的好处：

第一：降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
第二：提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。
第三：提高线程的可管理性。线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一分配、调优和监控。但是，要做到合理利用。

线程池的作用：

线程池是为突然大量爆发的线程设计的，通过有限的几个固定线程为大量的操作服务，减少了创建和销毁线程所需的时间，从而提高效率。如果一个线程的时间非常长，就没必要用线程池了(不是不能作长时间操作，而是不宜。)，况且我们还不能控制线程池中线程的开始、挂起、和中止。

 

各个类之间的关系：

Executor类是最顶层的接口，它的最顶层实现是ThreadPoolExecutor类。

Executors类是个工厂类，负责创建各种线程池。Executors工厂类中提供的newScheduledThreadPool、newFixedThreadPool、newCachedThreadPool方法其实也只是ThreadPoolExecutor的构造函数参数不同而已。

1、newCachedThreadPool

创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。示例代码如下：

ExecutorService executorService= Executors.newCachedThreadPool();
for(int i = 0;i<10;i++){
    final int temp = i;
    executorService.execute(new Runnable() {
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName()+":"+temp);
        }
    });
}
executorService.shutdown();

线程池为无限大，当执行第二个任务时第一个任务已经完成，会复用执行第一个任务的线程，而不用每次新建线程。

2、newFixedThreadPool

创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。

ExecutorService executorService = Executors.newFixedThreadPool(3);
for(int i = 0;i <10;i++){
    final int temp = i;
    executorService.execute(new Runnable() {
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName()+":"+temp);
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    });
}
executorService.shutdown();

因为线程池大小为3，每个任务输出index后sleep 2秒，所以每两秒打印3个数字。定长线程池的大小最好根据系统资源进行设置。如Runtime.getRuntime().availableProcessors()

3、newScheduledThreadPool

创建一个定长线程池，支持定时及周期性任务执行。延迟执行示例代码如下：

ScheduledExecutorService newScheduledThreadPool = Executors.newScheduledThreadPool(5);
		for (int i = 0; i < 10; i++) {
			final int temp = i;
			newScheduledThreadPool.schedule(new Runnable() {
				public void run() {
					System.out.println("i:" + temp);
				}
			}, 3, TimeUnit.SECONDS);
}
executorService.shutdown();

表示延迟3秒之后线程池开始执行

4、newSingleThreadExecutor

创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。示例代码如下：

ExecutorService executorService = Executors.newSingleThreadExecutor();
for(int i = 0;i <10;i++){
    final int temp = i;
    executorService.execute(new Runnable() {
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName()+":"+temp);
        }
    });
}
executorService.shutdown();

注意: 结果依次输出，相当于顺序执行各个任务。

线程池原理剖析

提交一个任务到线程池中，线程池的处理流程如下：

1、判断线程池里的核心线程是否都在执行任务，如果不是（核心线程空闲或者还有核心线程没有被创建）则创建一个新的工作线程来执行任务。如果核心线程都在执行任务，则进入下个流程。

2、线程池判断工作队列是否已满，如果工作队列没有满，则将新提交的任务存储在这个工作队列里。如果工作队列满了，则进入下个流程。

3、判断线程池里的线程是否都处于工作状态，如果没有，则创建一个新的工作线程来执行任务。如果已经满了，则交给饱和策略来处理这个任务。

并发编程--线程池原理

自定义线程池

如果当前线程池中的线程数目小于corePoolSize，则每来一个任务，就会创建一个线程去执行这个任务；

如果当前线程池中的线程数目>=corePoolSize，则每来一个任务，会尝试将其添加到任务缓存队列当中，若添加成功，则该任务会等待空闲线程将其取出去执行；若添加失败（一般来说是任务缓存队列已满），则会尝试创建新的线程去执行这个任务；

如果队列已经满了，则在总线程数不大于maximumPoolSize的前提下，则创建新的线程

如果当前线程池中的线程数目达到maximumPoolSize，则会采取任务拒绝策略进行处理；

如果线程池中的线程数量大于 corePoolSize时，如果某线程空闲时间超过keepAliveTime，线程将被终止，直至线程池中的线程数目不大于corePoolSize； 如果允许为核心池中的线程设置存活时间，那么核心池中的线程空闲时间超过keepAliveTime，线程也会被终止。

public class Test0007 {

	public static void main(String[] args) {
		ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 2, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(3));
		for (int i = 1; i <= 6; i++) {
			TaskThred t1 = new TaskThred("任务" + i);
			executor.execute(t1);
		}
		executor.shutdown();
	}
}

class TaskThred implements Runnable {
	private String taskName;

	public TaskThred(String taskName) {
		this.taskName = taskName;
	}

	@Override
	public void run() {
		System.out.println(Thread.currentThread().getName()+taskName);
	}

}

 

1、CPU密集

CPU密集的意思是该任务需要大量的运算，而没有阻塞，CPU一直全速运行。

CPU密集任务只有在真正的多核CPU上才可能得到加速(通过多线程)，而在单核CPU上，无论你开几个模拟的多线程，该任务都不可能得到加速，因为CPU总的运算能力就那些。

2、IO密集

IO密集型，即该任务需要大量的IO，即大量的阻塞。在单线程上运行IO密集型的任务会导致浪费大量的CPU运算能力浪费在等待。所以在IO密集型任务中使用多线程可以大大的加速程序运行，即时在单核CPU上，这种加速主要就是利用了被浪费掉的阻塞时间。

3、配置

要想合理的配置线程池的大小，首先得分析任务的特性，可以从以下几个角度分析：

1. 任务的性质：CPU密集型任务、IO密集型任务、混合型任务。

2. 任务的优先级：高、中、低。

3. 任务的执行时间：长、中、短。

4. 任务的依赖性：是否依赖其他系统资源，如数据库连接等。

 

对于不同性质的任务来说，CPU密集型任务应配置尽可能小的线程，如配置CPU个数+1的线程数，IO密集型任务应配置尽可能多的线程，因为IO操作不占用CPU，不要让CPU闲下来，应加大线程数量，如配置两倍CPU个数+1，而对于混合型的任务，如果可以拆分，拆分成IO密集型和CPU密集型分别处理，前提是两者运行的时间是差不多的，如果处理时间相差很大，则没必要拆分了。

若任务对其他系统资源有依赖，如某个任务依赖数据库的连接返回的结果，这时候等待的时间越长，则CPU空闲的时间越长，那么线程数量应设置得越大，才能更好的利用CPU。

Callable

Java提供了使用Callable和Future来实现获取任务结果的操作。Callable用来执行任务，产生结果，而Future用来获得结果。

public class Main {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        Future<String> submit = executorService.submit(new TestCallable());
        System.out.println("主线程开始");

        executorService.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    String s = submit.get();
                    System.out.println(s);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }
        });
        executorService.shutdown();
        System.out.println("主线程结束");
    }
}
class TestCallable implements Callable<String>{

    @Override
    public String call() throws Exception {
        System.out.println("正在执行任务。。。。。开始");
        Thread.sleep(3000);
        System.out.println("正在执行任务。。。。。结束");
        return "执行完成";
    }
}

Future模式

去除了主函数的等待时间，并使得原本需要等待的时间段可以用于处理其他业务逻辑。

 

处理逻辑：

1、首先我们有一个公共的返回数据的接口。

public interface Data {
   public abstract String getRequest();
}

2、真实的处理事件的部分

public class RealData implements Data {
   private String result;

   public RealData(String data) {
      System.out.println("正在使用data:" + data + "网络请求数据,耗时操作需要等待.");
      try {
         Thread.sleep(3000);
      } catch (Exception e) {

      }
      System.out.println("操作完毕,获取结果...");
      result = "result";
   }

   @Override
   public String getRequest() {
      return result;
   }
}

3、要返回的一个包装类，通过这个类能够拿到我们需要的返回值

public class FurureData implements Data {

   public volatile static boolean ISFLAG = false;
   private RealData realData;

   public synchronized void setRealData(RealData realData) {
      // 如果已经获取到结果，直接返回
      if (ISFLAG) {
         return;
      }
      // 如果没有获取到数据,传递真是对象
      this.realData = realData;
      ISFLAG = true;
      // 进行通知
      notify();
   }

   @Override
   public synchronized String getRequest() {
      while (!ISFLAG) {
         try {
            wait();
         } catch (Exception e) {

         }
      }
      // 获取到数据,直接返回
      return realData.getRequest();
   }
}

4、包装我们真实业务在Client里，并且将返回数据封装在FutureData中返回

public class FutureClient {

   public Data request(String queryStr) {
      FurureData furureData = new FurureData();
      new Thread(new Runnable() {

         @Override
         public void run() {
            RealData realData = new RealData(queryStr);
            furureData.setRealData(realData);
         }
      }).start();
      return furureData;

   }

5、调用

public class UseMain {
    public static void main(String[] args) {
        FutureClient futureClient = new FutureClient();
        Data request = futureClient.request("11111111111");
        System.out.println("请求发送成功!");
        System.out.println("执行其他任务...");
        String result = request.getRequest();
        System.out.println("获取到结果..." + result);
    }
}

 

相关推荐