java多线程的简单应用

概述

1、什么是线程

操作系统运行一个程序时，会为其创建一个进程，线程是系统工作调度的最小单位，一个进程运行时可能会有一个或者多个线程在工作。线程拥有各自的计数器、堆栈和局部变量等属性，且能够访问共享的内存变量。

2、线程的优先级

操作系统基本采用十分的形式调度运行的线程，操作系统会分出一个个时间片，线程会分配到若干时间片，时间片的多少决定了线程能使用处理器资源的多少。线程的优先级可以决定多或少的分配处理器资源。
构建线程的时候可以设置线程的priority属性控制优先级。优先级范围1~10，线程迷人优先级是5，优先级越高分配的时间片越多。

3、线程的状态

创建线程的方式

1、继承Thread类

public class MyThread extends Thread{
    
    // 重写run方法
    @Override
    public void run() {
        System.out.println("MyThread 子线程！");
    }
}
1
2
3
4
5
6
7
8

2、实现Runnable类

public class MyRunnable implements Runnable{
    
    // 重写run方法
    @Override
    public void run() {
        System.out.println("MyRunnable 子线程！");
    }
}
1
2
3
4
5
6
7
8

3、实现Callable类

public class CallThread implements Callable {
    
    // 重写call方法，有返回值。
    @Override
    public Object call() throws Exception {
        return "CallThread 子线程！";
    }
}
1
2
3
4
5
6
7
8

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CallThread callThread = new CallThread();
        // FutureTask用于获取线程处理结果
        FutureTask<String> futureTask = new FutureTask(callThread);
        Thread myCallThread = new Thread(futureTask);
        myCallThread.start();
        // futureTask.get() 会阻塞线程 知道子线程返回结果
        System.out.println(futureTask.get());

        System.out.println("这是主线程！");
    }
1
2
3
4
5
6
7
8
9
10
11

4、线程池

1、为何需要线程池？

线程的创建需要开辟虚拟机栈、本地方法栈、程序计数器等线程私有的内存空间。在线程销毁时需要回收这些系统资源。频繁的创建和销毁线程会浪费大量的系统资源，增加并发编程的风险。
线程池的作用：

• 利用线程池管理并服用线程，控制最大并发数等
• 实现任务线程队列缓存策略和拒绝机制
• 实现某些与时间相关的功能，如定时执行、周期执行等
• 隔离线程环境

2、ThreadPoolExecutor

核心参数：

• corePoolSize：保持线程池活动的最小线程数
• maximumPoolSize：线程池能够容纳同时执行的最大线程数
• keepAliveTime：线程池中线程的空闲时间，当空闲时间达到keepAliveTime时，线程会被销毁，直到只剩下corePoolSize个线程为止。线程数大于corePoolSize时keepAliveTime才会生效
• TimeUnit：keepAliveTime的时间单位，通常是TimeUnit.SECONDS
• workQueue：缓存队列。请求线程数大于corePoolSize时，线程会进入BlockingQueue阻塞队列，当队列缓存上限时，线程池会创建新的线程，最大线程数为maximumPoolSize
• threadFactory：线程工厂，用来生产一组相同任务的线程。
• handler：执行拒绝策略的对象

示例：

• ThreadFactory工厂

public class MyThreadFactory implements ThreadFactory {

    private final String namePrefix;

    private final AtomicInteger nextId = new AtomicInteger(1);


    MyThreadFactory(String threadPoolGroup) {
        namePrefix = "MyThreadFactory-" +  threadPoolGroup + "-Worker-";
    }

    @Override
    public Thread newThread(Runnable task) {
        String name = namePrefix + nextId.getAndIncrement();
        Thread thread = new Thread(task, name);
        System.out.println(thread.getName());
        return thread;
    }
}
class Task implements Runnable {

    private AtomicLong count = new AtomicLong(0L);

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + "- running_" + count.getAndIncrement());
    }
}
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

• RejectHandler拒绝策略

public class MyPoolRejectHandler implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        System.out.println("task reject --> " + executor.toString() );
    }
}
1
2
3
4
5
6

• 创建线程池

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 阻塞队列
        BlockingDeque blockingDeque = new LinkedBlockingDeque(2);
        // 拒绝策略
        MyPoolRejectHandler rejectHandler = new MyPoolRejectHandler();
        // 线程工厂
        MyThreadFactory f1 = new MyThreadFactory("threadGroup");
        // 初始化线程池
        ThreadPoolExecutor t1 = new ThreadPoolExecutor(1, 2, 60,
                TimeUnit.SECONDS, blockingDeque, f1, rejectHandler);
        Task task = new Task();
        for (int i = 0; i < 20; i++) {
            t1.execute(task);
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

线程工具类

1、CountDownLatch

CountDownLatch允许一个或多个线程等待其他线程完成操作。CountDownLatch的构造函数接收一个int类型的参数作为计数器，想等待N个点完成，就传入N。调用countDown方法时，N就会减一，await方法会阻塞当前线程，直至N变为0。

public class CountDownLatchTest {

    static CountDownLatch countDownLatch = new CountDownLatch(2);

    public static void main(String[] args) throws InterruptedException {
        Thread t1 = new Thread(() ->{
            System.out.println("t1完成");
            countDownLatch.countDown();
        }, "t1");
        Thread t2 = new Thread(() ->{
            System.out.println("t2完成");
            countDownLatch.countDown();
        }, "t2");
        t1.start();
        t2.start();

        countDownLatch.await();

        System.out.println("所有线程完成！");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

2、CyclicBarrier

可以让一组线程到达一个屏障（同步点）时被阻塞，知道最后一个线程到达屏障时，屏障才会开门，所有被拦截的线程才会继续运行。

public class CyclicBarrierTest {

    static CyclicBarrier cyclicBarrier = new CyclicBarrier(3);

    public static void main(String[] args) throws BrokenBarrierException, InterruptedException {
        Thread t1 = new Thread(() ->{
            try {
                cyclicBarrier.await();
            } catch (InterruptedException | BrokenBarrierException e) {
                throw new RuntimeException(e);
            }
            System.out.println("t1完成");
        }, "t1");
        Thread t2 = new Thread(() ->{
            try {
                cyclicBarrier.await();
            } catch (InterruptedException | BrokenBarrierException e) {
                throw new RuntimeException(e);
            }
            System.out.println("t2完成");
        }, "t2");
        t1.start();
        t2.start();

        cyclicBarrier.await();

        System.out.println("所有线程完成！");
    }
}
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

CyclicBarrier提供了更高级的构造函数CyclicBarrier(int parties, Runnable barrierAction)，用于线程到达屏障时，优先执行barrierAction，方便处理更复杂的业务场景。

public class CyclicBarrierTest implements Runnable{

    static CyclicBarrier cyclicBarrier = new CyclicBarrier(3, new CyclicBarrierTest());

    public static void main(String[] args) throws BrokenBarrierException, InterruptedException {
        Thread t1 = new Thread(() ->{
            try {
                cyclicBarrier.await();
            } catch (InterruptedException | BrokenBarrierException e) {
                throw new RuntimeException(e);
            }
            System.out.println("t1完成");
        }, "t1");
        Thread t2 = new Thread(() ->{
            try {
                cyclicBarrier.await();
            } catch (InterruptedException | BrokenBarrierException e) {
                throw new RuntimeException(e);
            }
            System.out.println("t2完成");
        }, "t2");
        t1.start();
        t2.start();

        cyclicBarrier.await();

        System.out.println("所有线程完成！");
    }

    @Override
    public void run() {
        System.out.println("优先执行的业务逻辑...");
    }
}
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