Fork me on GitHub

线程池

发表于 | | 阅读次数:

每一个线程的启动和结束都是比较消耗时间和占用资源的。
如果在系统中用到了很多的线程，大量的启动和结束动作会导致系统的性能变卡，响应变慢。
为了解决这个问题，引入线程池这种设计思想。
线程池的模式很像生产者消费者模式，消费的对象是一个一个的能够运行的任务。

线程池设计思路

线程池的思路和生产者消费者模型是很接近的。
1、准备一个任务容器
2、一次性启动10个消费者线程
3、刚开始任务容器时是空的，所以线程都wait在上面。
4、直到一个外部线程往这个任务容器中扔了一个”任务”，就会有一个消费者线程被唤醒notify。
5、这个消费者线程取出”任务”，并且执行这个任务，执行完毕后，继续等待下一次任务的到来。
6、如果短时间内，有较多的任务加入，那么就会有多个线程别唤醒，去执行这些任务。
11

开发一个自定义的线程池

这是一个自定义的线程池，虽然不够完善和健壮，但是已经足以说明线程池的工作原理。
缓慢的给这个线程池添加任务，会看到有多条线程来执行这些任务。
线程7执行完毕任务后，又回到池子里，下一次任务来的时候，线程7又来执行新的任务。
ThreadPool.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
import java.util.LinkedList;
  
public class ThreadPool {
  
    // 线程池大小
    int threadPoolSize;
  
    // 任务容器
    LinkedList<Runnable> tasks = new LinkedList<Runnable>();
  
    // 试图消费任务的线程
  
    public ThreadPool() {
        threadPoolSize = 10;
  
        // 启动10个任务消费者线程
        synchronized (tasks) {
            for (int i = 0; i < threadPoolSize; i++) {
                new TaskConsumeThread("任务消费者线程 " + i).start();
            }
        }
    }
  
    public void add(Runnable r) {
        synchronized (tasks) {
            tasks.add(r);
            // 唤醒等待的任务消费者线程
            tasks.notifyAll();
        }
    }
  
    class TaskConsumeThread extends Thread {
        public TaskConsumeThread(String name) {
            super(name);
        }
  
        Runnable task;
  
        public void run() {
            System.out.println("启动： " + this.getName());
            while (true) {
                synchronized (tasks) {
                    while (tasks.isEmpty()) {
                        try {
                            tasks.wait();
                        } catch (InterruptedException e) {
                            // TODO Auto-generated catch block
                            e.printStackTrace();
                        }
                    }
                    task = tasks.removeLast();
                    // 允许添加任务的线程可以继续添加任务
                    tasks.notifyAll();
  
                }
                System.out.println(this.getName() + " 获取到任务，并执行");
                task.run();
            }
        }
    }
}

TestThread.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
public class TestThread {
       
    public static void main(String[] args) {
        ThreadPool pool = new ThreadPool();
  
        for (int i = 0; i < 5; i++) {
            Runnable task = new Runnable() {
                @Override
                public void run() {
                    //System.out.println("执行任务");
                    //任务可能是打印一句话
                    //可能是访问文件
                    //可能是做排序
                }
            };
             
            pool.add(task);
             
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    } 
}

测试线程池

创造一个情景，每个任务执行的时间都是1秒。
刚开始是间隔1秒钟向线程池中添加任务，
然后间隔时间越来越短，执行任务的线程还没有来得及结束，新的任务又来了。
就会观察到线程池里的其他线程被唤醒来执行这些任务。
TestThread.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
public class TestThread {
    public static void main(String[] args) {
        ThreadPool pool= new ThreadPool();
        int sleep=1000;
        while(true){
            pool.add(new Runnable(){
                @Override
                public void run() {
                    //System.out.println("执行任务");
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            });
            try {
                Thread.sleep(sleep);
                sleep = sleep>100?sleep-100:sleep;
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
              
        }
          
    }
}

使用Java自带的线程池

Java提供自带的线程池，而不需要自己去开发一个自定义线程池。
线程池类ThreadPoolExecutor在包java.util.concurrent下

1
ThreadPoolExecutor threadPool= new ThreadPoolExecutor(10, 15, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());

第一个参数10表示这个线程池初始化了10个线程在里面工作。核心线程数
第二个参数15表示如果10个线程不够用了，就会自动增加到最多15个线程。最大线程数
第三个参数60结合第四个参数TimeUnit.SECONDS，表示经过60秒,多出来的线程还没有接到活儿，就会回收，最后保持池子里就10个。闲置线程存活时间
第四个参数TimeUnit.SECONDS如上。时间单位
第五个参数new LinkedBlockingQueue()用来放任务的集合。线程队列
execute方法用于添加新的任务。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
   
public class TestThread {
   
    public static void main(String[] args) throws InterruptedException {
           
        ThreadPoolExecutor threadPool= new ThreadPoolExecutor(10, 15, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
           
        threadPool.execute(new Runnable(){
   
            @Override
            public void run() {
                // TODO Auto-generated method stub
                System.out.println("任务1");
            }
        });
    }
}

借助线程池同步查找文件内容

初始化一个大小是10的线程池
遍历所有文件，当遍历到文件是.java的时候，创建一个查找文件的任务，把这个任务扔进线程池去执行，继续遍历下一个文件。
SearchFileTask.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
 
public class SearchFileTask implements Runnable{
 
    private File file;
    private String search;
    public SearchFileTask(File file,String search) {
        this.file = file;
        this.search= search;
    }
     
    public void run(){
         
        String fileContent = readFileConent(file);
        if(fileContent.contains(search)){
            System.out.printf( "线程: %s 找到子目标字符串%s,在文件:%s%n",Thread.currentThread().getName(), search,file);
             
        }
    }
     
    public String readFileConent(File file){
        try (FileReader fr = new FileReader(file)) {
            char[] all = new char[(int) file.length()];
            fr.read(all);
            return new String(all);
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }  
}

ThreadPool.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
import java.util.LinkedList;
  
public class ThreadPool {
    // 线程池大小
    int threadPoolSize;
  
    // 任务容器
    LinkedList<Runnable> tasks = new LinkedList<Runnable>();
  
    // 试图消费任务的线程
  
    public ThreadPool() {
        threadPoolSize = 10;
  
        // 启动10个任务消费者线程
        synchronized (tasks) {
            for (int i = 0; i < threadPoolSize; i++) {
                new TaskConsumeThread("任务消费者线程 " + i).start();
            }
        }
    }
  
    public void add(Runnable r) {
        synchronized (tasks) {
            tasks.add(r);
            // 唤醒等待的任务消费者线程
            tasks.notifyAll();
        }
    }
  
    class TaskConsumeThread extends Thread {
        public TaskConsumeThread(String name) {
            super(name);
        }
  
        Runnable task;
  
        public void run() {
            while (true) {
                synchronized (tasks) {
                    while (tasks.isEmpty()) {
                        try {
                            tasks.wait();
                        } catch (InterruptedException e) {
                            // TODO Auto-generated catch block
                            e.printStackTrace();
                        }
                    }
                    task = tasks.removeLast();
                    // 允许添加任务的线程可以继续添加任务
                    tasks.notifyAll();
  
                }
                task.run();
            }
        }
    }
  
}

TestThread.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import java.io.File;
   
public class TestThread {
   
    static ThreadPool pool= new ThreadPool();
    public static void search(File file, String search) {
         
        if (file.isFile()) {
            if(file.getName().toLowerCase().endsWith(".java")){
                SearchFileTask task = new SearchFileTask(file, search);
                pool.add(task);
            }
        }
        if (file.isDirectory()) {
            File[] fs = file.listFiles();
            for (File f : fs) {
                search(f, search);
            }
        }
    }
       
    public static void main(String[] args) {
        File folder =new File("e:\\project");
        search(folder,"Magic");
    }
}

Your support will encourage me to continue to create!