Java并发编程

推荐一个好的网站：并发编程网 - ifeve.com，上面全是各种大牛原创或编译的并发编程文章。

Semaphore(信号量) 控制并发资源

例子：多个线程抢几个打印机的使用权

import org.junit.Test;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by fengzp on 16/6/30.
 */
public class SemaphoreDemo {

    @Test
    public void test() throws InterruptedException {
        PrintQueue printQueue = new PrintQueue();

        int threadCount = 10;

        Thread thread[] = new Thread[threadCount];
        for (int i = 0; i < threadCount; i++) {
            thread[i] = new Thread(new Job(printQueue), "Thread" + i);
        }

        for (int i = 0; i < threadCount; i++) {
            thread[i].start();
        }

        TimeUnit.SECONDS.sleep(10);
    }

    public class Job implements Runnable {

        private PrintQueue printQueue;

        public Job(PrintQueue printQueue) {
            this.printQueue = printQueue;
        }

        public void run() {
            System.out.printf("%s: Going to print a job
", Thread.currentThread().getName());
            printQueue.printJob(new Object());
            System.out.printf("%s: The document has been printed
", Thread.currentThread().getName());
        }
    }

    class PrintQueue {
        private boolean freePrinters[];//用来存放打印机的状态，true表示空闲，false表示正在打印

        private Lock lockPrinters;//增加了锁，保证多个线程，只能获取得锁，才能查询哪台打印机空闲的

        private final Semaphore semaphore;

        private final int printerNum = 3;//假设有3台打印机

        public PrintQueue() {
            semaphore = new Semaphore(printerNum);
            freePrinters = new boolean[printerNum];

            for (int i = 0; i < printerNum; i++) {
                freePrinters[i] = true;//初始化时，默认所有打印机都空闲
            }
            lockPrinters = new ReentrantLock();
        }


        private int getPrinter() {
            int ret = -1;
            try {
                lockPrinters.lock();//先加锁，保证1次只能有1个线程来获取空闲的打印机
                for (int i = 0; i < freePrinters.length; i++) {
                    //遍历所有打印机的状态，发现有第1个空闲的打印机后，领取号码，
                    // 并设置该打印机为繁忙状态（因为马上就要用它）
                    if (freePrinters[i]) {
                        ret = i;
                        freePrinters[i] = false;
                        break;
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
                System.out.println(Thread.currentThread().getName() + "error");
            } finally {
                //最后别忘记了解锁，这样后面的线程才能上来领号
                lockPrinters.unlock();
            }
            return ret;
        }

        public void printJob(Object document) {
            try {
                semaphore.acquire();//取得对共享资源的访问权（即拿到了钥匙））

                int assignedPrinter = getPrinter();//领号
                long duration = (long) (1 + Math.random() * 10);
                System.out.printf("%s: PrintQueue: Printing a Job in Printer%d during %d seconds
", Thread.currentThread().getName(),
                        assignedPrinter, duration);
                Thread.sleep(duration);
                freePrinters[assignedPrinter] = true;//打印完以后，将该打印机重新恢复为空闲状态

            } catch (InterruptedException e) {
                e.printStackTrace();
                System.out.println(Thread.currentThread().getName() + "error");
            } finally {
                semaphore.release();//钥匙用完了，要还回去，这样其它线程才能继续有序的拿到钥匙，访问资源
            }
        }
    }
}

View Code

CyclicBarrier(等待多个线程执行完成后再后续处理)

import org.junit.Test;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

/**
 * Created by fengzp on 16/6/30.
 */
public class CyclicBarrierDemo {

    @Test
    public void test() throws BrokenBarrierException, InterruptedException {

        final int threadNum = 10;
        CyclicBarrier cb = new CyclicBarrier(threadNum + 1);//注意：10个子线程 + 1个主线程

        for (int i = 0; i < threadNum; i++) {
            new Thread(new MyRunable(cb, i)).start();
        }

        cb.await();
        System.out.println("-----------
所有thread执行完成！");
    }

    static class MyRunable implements Runnable {
        CyclicBarrier _cb;
        int _i = 0;

        public MyRunable(CyclicBarrier cb, int i) {
            this._cb = cb;
            this._i = i;
        }

        @Override
        public void run() {
            try {
                Thread.sleep((long) (Math.random() * 100));
                System.out.println("thread " + _i + " done，正在等候其它线程完成...");
                _cb.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

View Code

Quere(队列，先进先出)

JDK7提供了以下7个阻塞队列：

ArrayBlockingQueue ：由数组结构组成的有界阻塞队列。
LinkedBlockingQueue ：由链表结构组成的有界阻塞队列。
PriorityBlockingQueue ：支持优先级排序的无界阻塞队列。
DelayQueue：使用优先级队列实现的无界阻塞队列。
SynchronousQueue：不存储元素的阻塞队列。
LinkedTransferQueue：链表结构组成的无界阻塞队列。
LinkedBlockingDeque：链表结构组成的双向阻塞队列。

阻塞队列提供了下列四种处理方法:

方法处理方式	抛出异常	返回true/false	一直阻塞	超时退出
插入方法	add(e)	offer(e)	put(e)	offer(e,time,unit)
移除方法	remove()	poll()	take()	poll(time,unit)
检查方法	element()	peek()

这4类方法中，在队列已满（或为空）的情况下，有些会抛出异常，有些则返回true/false，有些则一直阻塞，还有些则可以设置超时时间，时间到了后，自动退出阻塞状态，实际项目中可根据需要选取适合的方法。

例子：模拟生产者-消费者，1个生产者，3个消费者

import org.junit.Test;

import java.util.concurrent.ArrayBlockingQueue;

/**
 * Created by fengzp on 16/6/30.
 */
public class QueueDemo {

    private static final int queueSize = 3;
    private static final ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(queueSize);
    private static final int produceSpeed = 2000;//生产速度(越小越快)
    private static final int consumeSpeed = 10;//消费速度(越小越快)

    public static void main(String[] args) {
        Thread producer = new Producer();
        Thread consumer = new Consumer();
        producer.start();
        consumer.start();
    }

    static class Producer extends Thread {
        public void run() {
            while (true) {
                try {
                    System.out.println("老板准备炸油条了，架子上还能放：" + (queueSize - queue.size()) + "根油条");
                    queue.put("1根油条");
                    System.out.println("老板炸好了1根油条，架子上还能放：" + (queueSize - queue.size()) + "根油条");
                    Thread.sleep(produceSpeed);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    static class Consumer extends Thread {
        public void run() {
            while (true) {
                try {
                    System.out.println("A 准备买油条了，架子上还剩" + queue.size() + "根油条");
                    queue.take();
                    System.out.println("A 买到1根油条，架子上还剩" + queue.size() + "根油条");
                    Thread.sleep(consumeSpeed);

                    System.out.println("B 准备买油条了，架子上还剩" + queue.size() + "根油条");
                    queue.take();
                    System.out.println("B 买到1根油条，架子上还剩" + queue.size() + "根油条");
                    Thread.sleep(consumeSpeed);

                    System.out.println("C 准备买油条了，架子上还剩" + queue.size() + "根油条");
                    queue.take();
                    System.out.println("C 买到1根油条，架子上还剩" + queue.size() + "根油条");
                    Thread.sleep(consumeSpeed);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

View Code

ThreadLocal(隔离变量)

/**
 * Created by fengzp on 16/6/30.
 */
public class ThreadLocalDemo {

    public static class MyRunnable implements Runnable {

        private ThreadLocal<Integer> threadLocal = new ThreadLocal<Integer>();

        @Override
        public void run() {
            threadLocal.set((int) (Math.random() * 100D));
            System.out.println(Thread.currentThread().getName() + ":" + threadLocal.get());
        }
    }

    public static void main(String[] args) {
        Thread t1 = new Thread(new MyRunnable(), "A");
        Thread t2 = new Thread(new MyRunnable(), "B");
        t1.start();
        t2.start();
    }
}

View Code

线程A与线程B中ThreadLocal保存的整型变量是各自独立的，互不相干，只要在每个线程内部使用set方法赋值，然后在线程内部使用get就能取到对应的值。

ThreadLocal还有一个派生的子类：InheritableThreadLocal ，可以允许线程及该线程创建的子线程均可以访问同一个变量(有些OOP中的proteced的意味)

/**
 * Created by fengzp on 16/6/30.
 */
public class InheritableThreadLocalDemo {

    private static InheritableThreadLocal<Integer> threadLocal = new InheritableThreadLocal<Integer>();

    public static class MyRunnable implements Runnable {

        public MyRunnable(String name) {
            System.out.println(name + " => " + Thread.currentThread().getName() + ":" + threadLocal.get());
        }

        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName() + ":" + threadLocal.get());
        }
    }

    public static void main(String[] args) {
        threadLocal.set(1);

        System.out.println(Thread.currentThread().getName() + ":" + threadLocal.get());
        Thread t1 = new Thread(new MyRunnable("R-A"), "A");
        Thread t2 = new Thread(new MyRunnable("R-B"), "B");

        t1.start();
        t2.start();
    }
}

View Code

DaemonThread(守护线程)

在正式理解这个概念前，先把守护线程与守护进程这二个极其相似的说法区分开，守护进程通常是为了防止某些应用因各种意外原因退出，而在后台独立运行的系统服务或应用程序。比如：我们开发了一个邮件发送程序，一直不停的监视队列池，发现有待发送的邮件，就将其发送出去。如果这个程序挂了(或被人误操作关了)，邮件就不发出去了，为了防止这种情况，再开发一个类似windows 系统服务的应用，常驻后台，监制这个邮件发送程序是否在运行，如果没运行，则自动将其启动。

而我们今天说的java中的守护线程(Daemon Thread) 指的是一类特殊的Thread，其优先级特别低(低到甚至可以被JVM自动终止)，通常这类线程用于在空闲时做一些资源清理类的工作，比如GC线程，如果JVM中所有非守护线程（即：常规的用户线程）都结束了，守护线程会被JVM中止，想想其实也挺合理，没有任何用户线程了，自然也不会有垃圾对象产生，GC线程也没必要存在了。

public class Program {
 
    public static void main(String[] args) {
        TestThread t1 = new TestThread();
        t1.setDaemon(true);
        t1.start();
    }
 
    private static class TestThread extends Thread {
        public void run() {
            System.out.println("test");
        }
    }
}

View Code

由于t1设置成Daemon Thread了，运行后，main进程马上就结束，此时没有用户进程在运行，守护进程默认是不执行的，因此运行后，没有任何输出结果，符合我们刚才的解释。

然后再写一个模拟例子，一个常规的用户线程负责写入日志，一个守护线程负责清除日志

/**
 * Created by fengzp on 16/6/30.
 */
public class DamonThreadDemo {

    private static int queueCapacity = 10;
    private static BlockingQueue<String> logQueue = new ArrayBlockingQueue<String>(queueCapacity);

    public static void main(String[] args) throws IOException {

        LogWriter writer = new LogWriter();
        LogCleaner cleaner = new LogCleaner();
        cleaner.setDaemon(true);

        writer.start();
        cleaner.start();
    }

    /**
     * 模拟不停写日志（直到队列写满）
     */
    private static class LogWriter extends Thread {
        public void run() {
            for (int i = 0; i < queueCapacity; i++) {
                try {
                    logQueue.put("" + i);
                    System.out.println("日志已写入，当前日志内容：" + logQueue);
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    /**
     * 模拟在空闲时清理日志（仅保留5条日志）
     */
    private static class LogCleaner extends Thread {
        public void run() {
            while (true) {
                if (logQueue.size() > 5) {
                    try {
                        logQueue.take();
                        System.out.println("多余日志被清理，当前日志内容：" + logQueue);
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
}

View Code