Java 多线程（6）----线程池（上）

/**
 * 自定义阻塞队列实现生产者、消费者模式：
 */
public static class CustomBlockingQueueTest {
    // 自定义的阻塞队列
    public static class CustomBlockingQueue<T> {
        private Object[] elements = null;
        // 当前元素个数
        private int elementsCount = 0;
        // 阻塞队列的锁资源
        private ReentrantLock lock = new ReentrantLock();
        // 队列已满的线程阻塞控制器
        Condition fullLock = null;
        // 队列为空的线程阻塞控制
        Condition emptyLock = null;

        public CustomBlockingQueue(int elementsCount) {
            elements = new Object[elementsCount];
            fullLock = lock.newCondition();
            emptyLock = lock.newCondition();
        }

        /**
         * 向队列末尾插入新元素的方法，如果队列元素已满，那么阻塞当前线程
         */
        public void put(T ele) throws InterruptedException {
            if (ele == null) {
                throw new IllegalArgumentException("插入元素不能为空！");
            } else {
                // 构造同步块
                lock.lock();
                try {
                    // 如果当前队列已满，那么阻塞当前生产者线程
                    while (elementsCount == elements.length) {
                        System.out.println("队列已满，元素插入失败，线程被阻塞！");
                        fullLock.await();
                    }
                    // 将元素插入队列末尾
                    elements[elementsCount++] = ele;
                    // 队列中已经有元素，唤醒所有阻塞的消费者线程
                    emptyLock.signalAll();
                    System.out.println("插入元素，当前队列元素数量：" + elementsCount);
                } finally {
                    lock.unlock();
                }
            }
        }

        /**
         * 从阻塞队列中取出元素的方法，如果队列中没有元素，那么阻塞当前线程
         */
        public T take() throws InterruptedException {
            // 构造同步块
            lock.lock();
            try {
                // 如果当前队列已空，那么阻塞当前消费者线程
                while (elementsCount == 0) {
                    System.out.println("队列已空，元素取出失败，线程被阻塞！");
                    emptyLock.await();
                }
                // 队列中已经有剩余空间，唤醒所有阻塞的生产者线程
                fullLock.signalAll();
                T ele = (T) elements[--elementsCount];
                System.out.println("取出元素，当前队列元素数量：" + elementsCount);
                // 将队列中元素的引用置为 null，有利于 GC 回收这个对象
                elements[elementsCount] = null;
                return ele;
            } finally {
                lock.unlock();
            }
        }
    }

    // 模拟产品的类
    public static class Product {
        private String productName;

        public Product(String name) {
            this.productName = name;
        }
    }
    // 记录产品数量
    private static int productCount = 0;
    // 创建一个容量为 5 的自定义阻塞队列
    private static CustomBlockingQueue<Product> queue = 
            new CustomBlockingQueue<Product>(5);

    // 创建生产者线程
    private static Thread productThread = new Thread() {
        @Override
        public void run() {
            Product pro = null;
            while (true) {
                try {
                    pro = new Product("产品" + (++productCount));
                    queue.put(pro);
                    System.out.println(pro.productName + " 存入成功！");
                } catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
        }
    };
    // 创建消费者线程
    private static Thread customThread = new Thread() {
        @Override
        public void run() {
            Product pro = null;
            while (true) {
                try {
                    pro = queue.take();
                    if (pro != null) {
                        System.out.println(pro.productName + " 取出成功！");
                    }
                } catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
        };
    };

    public static void startTest() {
        productThread.start();
        customThread.start();
    }
}

ArrayBlockingQueue：数组构成的有界阻塞队列，即储存元素的数据结构是数组

LinkedBlockingQueue：链表构成的有界阻塞队列，即储存元素的数据结构是链表

PriorityBlockingQueue：支持按某个优先级对元素进行排序的无界阻塞队列

DelayQueue：使用优先队列实现的无界阻塞队列

SynchronousQueue：不储存元素的阻塞队列

LinkedTransferQueue：由链表组成的无界阻塞队列

LinkedBlockingDeque：由链表构成的双向阻塞队列

// 创建一个容量为 1000 的公平阻塞队列
ArrayBlockingQueue queue = new ArrayBlockingQueue(1000, true); 

/**
 * ArrayBlockingQueue 类的使用，使用 ArrayBlockingQueue 实现生产者、消费者问题：
 */
public static class ArrayBlockingQueueTest {
    static int productCount = 0;
    // 模拟产品的类
    public static class Product {
        private String productName;

        public Product(String name) {
            this.productName = name;
        }
    }
    // 创建一个容量为 5 的公平阻塞队列，
    // 公平即为先被阻塞的线程在被唤醒后可以先得到锁资源继续上次未完成的操作
    private static ArrayBlockingQueue<Product> queue = new ArrayBlockingQueue<Product>(5, true);
    // 创建生产者线程
    private static Thread productThread = new Thread() {
        @Override
        public void run() {
            Product pro = null;
            while (true) {
                try {
                    pro = new Product("产品" + (++productCount));
                    // 插入元素到队尾
                    queue.put(pro);
                    System.out.println(pro.productName + " 存入成功！");
                } catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
        }
    };
    // 创建消费者线程
    private static Thread customThread = new Thread() {
        @Override
        public void run() {
            Product pro = null;
            while (true) {
                try {
                    // 取出队头元素
                    pro = queue.take();
                    if (pro != null) {
                        System.out.println(pro.productName + " 取出成功！");
                    }
                } catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
        };
    };

    public static void startTest() {
        productThread.start();
        customThread.start();
    }
}

public static void main(String[] args) {
    CustomBlockingQueueTest.startTest();
}

public interface Future<V> {

    /**
     * 尝试取消执行提交的对应任务，如果对应任务已经执行完成或者已经被取消或者是其他原因不能被取消，
     * 那该操作将会失败，否则的话提交的对应任务将不会被线程池中的线程执行。
     * 如果尝试取消的任务正在执行，那么通过参数来确定是否应该中断正在执行该任务的线程，
     * 如果为 false，那么将不会中断正在执行该任务的线程。
     * 在该方法返回之后，调用 isDone 方法会返回 true，
     * 调用 isCancalled 方法的返回值和该方法的返回值相同。
     */
    boolean cancel(boolean mayInterruptIfRunning);

    /**
     * 如果提交的对应任务还没有完全执行完成之前就被取消了，那么方法会返回 true，否则返回 false
     */
    boolean isCancelled();

    /**
     * 返回任务是否完成：包括下面几种情况：
     * 1、任务正常执行完成
     * 2、执行过程发生异常
     * 3、任务被取消
     * 上面几种情况发生，方法均会返回 true，
     * 如果任务正在执行，或者还未执行，那么方法返回 false
     */
    boolean isDone();

    /**
     * 阻塞调用该方法的线程，直到提交的对应任务执行完成之后，
     * 方法会返回一个泛型结果对象表示任务执行的结果，
     * 如果提交的任务是一个 Callable 类型的对象，那么返回 Callable 对象的 call 方法的返回值，
     * 如果提交的任务是一个 Runnable 类型的对象，那么返回 Runnable 对象的 run 方法的返回值，即为 null。
     * 如果在调用线程阻塞的过程中发生了中断，那么方法抛出 InterruptedException 异常
     */
    V get() throws InterruptedException, ExecutionException;

    /**
     * 功能同上面的重载方法，但是添加了一个条件, 即阻塞时间，
     * 该方法使得调用线程的阻塞时间不会超过参数指定的时间，
     * 如果在规定时间内对应任务没有运行完成，方法抛出一个 TimeoutException 异常
     */
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

public interface Callable<V> {
    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
    V call() throws Exception;
}

/**
 * Future 接口测试
 */
public static class FutureTaskTest {
    public static void startTest() {
        // 创建一个新的线程池
        ExecutorService executor = Executors.newCachedThreadPool();
        // 向线程池中提交一个新的 Callable 类型的任务，提交之后会对任务对象进行包装
        Future<String> task = executor.submit(new Callable<String>() {
            // 复写 call() 方法，线程池中线程最终会调用 call() 方法
            @Override
            public String call() throws Exception {
                // 执行线程休眠 5 秒钟
                Thread.sleep(5000);
                return "hello world";
            }
        });
        long startTime = System.currentTimeMillis();
        try {
            // 调用 task.get() 获取执行结果，
            // 这里即为获取上述代码中 submit 方法提交的 Callable 中 call() 方法的返回值，
            // 该方法会阻塞当前调用线程，直到任务执行完成后返回
            System.out.println(task.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        long endTime = System.currentTimeMillis();
        System.out.println("方法获取任务结果所用时间：" + (endTime - startTime) + "ms");
    }
}

public static void main(String[] args) {
    FutureTaskTest.startTest();
}