Java多线程与线程池技术分享

1. 多线程

什么是线程？

线程是一个程序执行流的最小单元，一个程序至少有一个执行流，即主线程。主线程在JVM启动时就存在了。

创建线程的方式

继承Thread类

重写Thread类的run()方法。

public class MyThread extends Thread {
    @Override
    public void run() {
        // 线程体
    }
}

实现Runnable接口

实现Runnable接口中的run()方法。

public class MyRunnable implements Runnable {
    @Override
    public void run() {
        // 线程体
    }
}

使用匿名内部类

new Thread(new Runnable() {
    @Override
    public void run() {
        // 线程体
    }
}).start();

线程状态

• new：线程创建后尚未启动。
• runnable：可运行状态，可运行但可能暂停执行。
• blocked：阻塞状态，表示线程因为某些原因暂停执行。
• waiting：等待状态，线程进入该状态表示在等待某个条件发生，按需等待或按时间等待。
• time waiting：超时等待状态，等待某个条件的发生，超时后自动执行。
• terminated：终止状态。

线程同步

synchronized关键字

synchronized是Java提供的一种锁机制，保证同一时刻只有一个线程可以访问同步代码块。

synchronized (锁对象) {
    // 同步代码块
}

Lock接口

Lock接口提供比synchronized更细粒度的同步机制。在编程中使用ReentrantLock。

Lock lock = new ReentrantLock();
lock.lock();
try {
    // 同步代码块
} finally {
    lock.unlock(); // 必须释放锁，否则会导致死锁
}

线程间通信

wait、notify、notifyAll方法

wait方法释放对象锁并挂起当前线程，notify方法唤醒等待该对象锁的线程。

synchronized (锁对象) {
    while (条件不满足) {
        锁对象.wait();
    }
    // 条件满足执行操作
    // 唤醒其他线程
    锁对象.notify();
    // 或者唤醒所有等待线程
    锁对象.notifyAll();
}

Condition接口

Condition接口提供的await、signal、signalAll方法与wait、notify、notifyAll方法类似。在编程中使用ReentrantLock配合Condition。

Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();
lock.lock();
try {
    while (条件不满足) {
        condition.await();
    }
    // 条件满足执行操作
    condition.signal(); // 唤醒一个等待线程
    condition.signalAll(); // 唤醒所有等待线程
} finally {
    lock.unlock();
}

示例

多线程下载图片

public class DownloadImage {
    public static void main(String[] args) throws Exception {
        URL url = new URL("https://example.com/image.jpg");
        URLConnection connection = url.openConnection();
        int size = connection.getContentLength();
        if (size <= 0) {
            throw new Exception("文件大小为0或小于0");
        }
        int threadCount = 5;
        int[][] ranges = new int[threadCount][2];
        int blockSize = size / threadCount;
        for (int i = 0; i < threadCount; i++) {
            int startPos = i * blockSize;
            int endPos = (i + 1) * blockSize - 1;
            if (i == threadCount - 1) {
                endPos = size - 1;
            }
            ranges[i][0] = startPos;
            ranges[i][1] = endPos;
            System.out.printf("线程%d下载范围：%d-%d %n", i, startPos, endPos);
        }
        RandomAccessFile raf = new RandomAccessFile("image.jpg", "rw");
        raf.setLength(size);
        raf.close();
        ExecutorService service = Executors.newFixedThreadPool(threadCount);
        for (int i = 0; i < threadCount; i++) {
            service.execute(new DownloadThread(url, ranges[i][0], ranges[i][1]));
        }
        service.shutdown();
    }
}

class DownloadThread implements Runnable {
    private URL url;
    private int startPos;
    private int endPos;

    public DownloadThread(URL url, int startPos, int endPos) {
        this.url = url;
        this.startPos = startPos;
        this.endPos = endPos;
    }

    @Override
    public void run() {
        try {
            HttpURLConnection conn = (HttpURLConnection) url.openConnection();
            conn.setRequestMethod("GET");
            conn.setRequestProperty("Range", "bytes=" + startPos + "-" + endPos);
            InputStream in = conn.getInputStream();
            RandomAccessFile raf = new RandomAccessFile("image.jpg", "rw");
            raf.seek(startPos);
            byte[] buffer = new byte[1024];
            int len;
            while ((len = in.read(buffer)) != -1) {
                raf.write(buffer, 0, len);
            }
            in.close();
            raf.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

该示例中，使用多线程的方式下载大文件，将文件分为若干个范围，每个线程负责下载对应范围内的文件。同时，也要注意对同步代码块、文件写入等要素的安全管理。

2. 线程池

线程池的概念

线程池是一种特殊的线程组，包含若干个工作线程，它们可重复利用，执行一些预设好的任务，并且提供了管理、调度等机制。

线程池的好处

• 创建、销毁线程的开销比较大，线程池可以重复利用线程，节省开销。
• 线程数量受到限制，避免因过多的线程而导致CPU、内存的浪费或调度的问题。

Java线程池

Java提供了ThreadPoolExecutor类，可以通过该类创建线程池。

ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize,
                keepAliveTime, unit, workQueue, threadFactory, rejectedExecutionHandler);

ThreadPoolExecutor的构造函数有7个参数，具体含义如下：

• corePoolSize：核心线程池大小，当提交一个任务时，如果当前线程池大小小于corePoolSize，则新建线程执行任务。当提交一个任务时，如果当前线程池大小等于corePoolSize，且队列未满，则任务被存储在队列中等待执行。
• maximumPoolSize：线程池最大的大小，在队列满了之后，当新的任务再次到来时，如果已创建的线程数小于maximumPoolSize，则会新创建线程执行该任务。
• keepAliveTime：表示线程没有任务执行时最多保持多久时间会终止。默认情况下，corePoolSize中的线程不会退出，除非设置allowCoreThreadTimeOut（允许核心线程超时退出）为true。
• unit：keepAliveTime的时间单位。
• workQueue：线程池中的任务队列，常用的有以下四种：
  • SynchronousQueue：同步队列，任务直接提交给线程池的线程执行，不存储任务，提交任务时，线程池中没有可用线程则会新建一个线程。
  • LinkedBlockingQueue：无界阻塞队列，适用于源源不断的异步任务。
  • ArrayBlockingQueue：有界阻塞队列，适用于限制线程数量不太大并发度适中的情况。
  • PriorityBlockingQueue：具有优先级的任务队列，优先级由Comparable规定。
• threadFactory：线程工厂，用来创建线程类，默认的即可。
• rejectedExecutionHandler：饱和策略，当任务队列满了且当前线程数等于maximumPoolSize时，处理该如何了。常用的饱和策略有以下几种：
  • ThreadPoolExecutor.AbortPolicy：默认策略，丢弃任务，并抛出RejectedExecutionException异常。
  • ThreadPoolExecutor.CallerRunsPolicy：回退策略，由提交任务的线程来直接执行改任务。
  • ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面（最早提交的）任务，并重新提交当前任务。

线程池的状态

• running：线程池处于运行状态，可以接受新任务并处理已加入队列的任务。
• shutdown：线程池处于关闭状态，不再接受新任务，但会处理已加入队列的任务。
• stop：线程池处于永久终止状态，不会处理已加入队列的任务。
• tidying：任务队列为空，线程池中活跃线程数等于0，正在执行终止任务。
• terminated：执行完终止任务后变为该状态。

示例

线程池下载图片

public class DownloadImageWithThreadPool {

    public static void main(String[] args) throws Exception {
        URL url = new URL("https://www.example.com/image.jpg");
        HttpURLConnection connection = (HttpURLConnection) url.openConnection();
        int size = connection.getContentLength();
        if (size <= 0) {
            throw new Exception("文件大小为0或小于0");
        }
        int threadCount = 5;
        int[][] ranges = new int[threadCount][2];
        int blockSize = size / threadCount;
        for (int i = 0; i < threadCount; i++) {
            int startPos = i * blockSize;
            int endPos = (i + 1) * blockSize - 1;
            if (i == threadCount - 1) {
                endPos = size - 1;
            }
            ranges[i][0] = startPos;
            ranges[i][1] = endPos;
            System.out.printf("线程%d下载范围：%d-%d %n", i, startPos, endPos);
        }
        RandomAccessFile raf = new RandomAccessFile("image2.jpg", "rw");
        raf.setLength(size);
        raf.close();
        ExecutorService service = Executors.newFixedThreadPool(threadCount);
        CountDownLatch latch = new CountDownLatch(threadCount);
        for (int i = 0; i < threadCount; i++) {
            service.execute(new DownloadTask(url, ranges[i][0], ranges[i][1], latch));
        }
        latch.await();
        service.shutdown();
    }
}

class DownloadTask implements Runnable {
    private URL url;
    private int startPos;
    private int endPos;
    private CountDownLatch latch;

    public DownloadTask(URL url, int startPos, int endPos, CountDownLatch latch) {
        this.url = url;
        this.startPos = startPos;
        this.endPos = endPos;
        this.latch = latch;
    }

    @Override
    public void run() {
        try {
            HttpURLConnection conn = (HttpURLConnection) url.openConnection();
            conn.setRequestMethod("GET");
            conn.setRequestProperty("Range", "bytes=" + startPos + "-" + endPos);
            InputStream in = conn.getInputStream();
            RandomAccessFile raf = new RandomAccessFile("image2.jpg", "rw");
            raf.seek(startPos);
            byte[] buffer = new byte[1024];
            int len;
            while ((len = in.read(buffer)) != -1) {
                raf.write(buffer, 0, len);
            }
            in.close();
            raf.close();
            System.out.printf("Thread %d downloaded %d bytes from %d to %d %n", Thread.currentThread().getId(), endPos - startPos + 1, startPos, endPos);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            latch.countDown();
        }
    }
}

该示例使用线程池的方式下载大文件，这样不仅可以避免由于线程创建销毁带来的开销，还可以方便管理已在队列中排队等待处理的任务。使用CountDownLatch，等待全部线程执行结束之后，关闭线程池。