多线程学习—5-线程池的其他组件

发表于 2016-12-05 | 分类于 Java多线程 | | 阅读次数

前言

多线程的基础功能已经总结完了，还剩几个jdk为我们提供的组件来配合不同的使用场景。今天总结下使用方法，源码等有时间再深究吧。

Timer

Timer的作用

执行定时任务。
- 计划时间晚于当前时间：未来执行
- 计划时间早于当前时间：立即执行
- 允许多个任务同时执行
cancle
- timer的cancle，清除所有的定时任务
- timeTask的cancle，清除某一个任务

CountDownLatch

用”中式英语翻译”就是当”计数未完成时一直锁住”，即规定的线程个数达到预期的状态时触发事件，让其他线程得以继续工作。我觉得还是比较实用的，在”汇总”工作时经常用到。

package MyThread;
import java.util.concurrent.CountDownLatch;
/**
 * Created by yuhang on 2016/12/5.
 */
public class MyThread11 extends Thread {
    private CountDownLatch cdl;
    private int sleepSecond;
    public MyThread11(String name, CountDownLatch cdl, int sleepSecond) {
        super(name);
        this.cdl = cdl;
        this.sleepSecond = sleepSecond;
    }
    public void run() {
        try {
            System.out.println(this.getName() + "启动了 !");
            long time = System.currentTimeMillis();
            Thread.sleep(sleepSecond * 1000);
            cdl.countDown();
            System.out.println(this.getName() + "执行完了，时间为" + (System.currentTimeMillis() - time) / 1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    private static class DoneThread extends Thread {
        private CountDownLatch cdl;
        public DoneThread(String name, CountDownLatch cdl) {
            super(name);
            this.cdl = cdl;
        }
        public void run() {
            try {
                System.out.println(this.getName() + "要等待了！");
                long time = System.currentTimeMillis();
                cdl.await();//等待countDown()
                System.out.println(this.getName() + "等待完了, 等待时间为：" + (System.currentTimeMillis() - time) / 1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args) throws Exception {
        //改变参数来体验被唤醒的情况
        CountDownLatch cdl = new CountDownLatch(3);
        DoneThread dt0 = new DoneThread("DoneThread1", cdl);
        DoneThread dt1 = new DoneThread("DoneThread2", cdl);
        dt0.start();
        dt1.start();
        MyThread11 wt0 = new MyThread11("WorkThread1", cdl, 2);
        MyThread11 wt1 = new MyThread11("WorkThread2", cdl, 3);
        MyThread11 wt2 = new MyThread11("WorkThread3", cdl, 4);
        wt0.start();
        wt1.start();
        wt2.start();
    }
}

SemaPhore

SemaPhore有两个主要的方法：acquire和release。分别是获得和释放信号，在创建SemaPhore指定一个int型的参数，当acquire等于这个数时，后来的acquire方法会被阻塞住，直到release信号时。注意SemaPhore是可以指定公平锁的。

package MyThread;
import java.util.concurrent.Semaphore;
/**
 * Created by yuhang on 2016/12/5.
 */
public class MyThread12 {
    public static void main(String[] args) {
        //可以改变参数来控制允许并发的数量
        final Semaphore semaphore = new Semaphore(4);
        Runnable runnable = () -> {
            try {
                semaphore.acquire();
                System.out.println(Thread.currentThread().getName() + "获得了信号量 ！");
                long time = System.currentTimeMillis();
                Thread.sleep(2000);
                System.out.println(Thread.currentThread().getName() + "释放了信号量,时间为" + (System.currentTimeMillis() - time) / 1000 + " 秒");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                semaphore.release();
            }
        };
        Thread[] threads = new Thread[10];
        for (int i = 0; i < threads.length; i++)
            threads[i] = new Thread(runnable);
        for (int i = 0; i < threads.length; i++) {
            threads[i].start();
        }
    }
}

CyclicBarrier

再次用”用中式英语”理解下字面意思：将给定数量的成员集中在一个屏障点上，然后在统一执行后面的操作。意思即是若干个线程一起执行，当数量达到某个条件时，线程阻塞，等某个任务执行完时，再继续执行await()以后的代码。和前面的CountDownLatch有点像，但是是完全不同的，先运行下代码，再来总结不同点。

package MyThread;
import java.util.concurrent.CyclicBarrier;
/**
 * Created by yuhang on 2016/12/5.
 */
public class MyThread14 {
    public static class CyclicBarrierThread extends Thread {
        private CyclicBarrier cb;
        private int sleepSecond;
        public CyclicBarrierThread(CyclicBarrier cb, int sleepSecond) {
            this.cb = cb;
            this.sleepSecond = sleepSecond;
        }
        public void run() {
            try {
                System.out.println(this.getName() + "运行了");
                Thread.sleep(sleepSecond * 1000);
                System.out.println(this.getName() + "准备等待了, 时间为" + System.currentTimeMillis());
                cb.await();
                System.out.println(this.getName() + "结束等待了, 时间为" + System.currentTimeMillis());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args) {
        Runnable runnable = () -> System.out.println("CyclicBarrier的所有线程await()结束了，我运行了, 时间为" + System.currentTimeMillis());
        //三个屏障
        CyclicBarrier cb = new CyclicBarrier(3, runnable);
        CyclicBarrierThread cbt0 = new CyclicBarrierThread(cb, 3);
        CyclicBarrierThread cbt1 = new CyclicBarrierThread(cb, 6);
        CyclicBarrierThread cbt2 = new CyclicBarrierThread(cb, 9);
        cbt0.start();
        cbt1.start();
        cbt2.start();
    }
}

CyclicBarrier和CountDownLatch的区别

根据上面两个demo其实可以理解：

countDown()方法执行仅仅是给”计数器减一”，然后当前线程不阻塞，继续运行下面的代码，当”计数器为0”时，会触发多个任务；而CyclicBarrier是”计数器”运行到”屏障”前必须阻塞，知道屏障内的代码执行完，前面的线程才会继续执行，并且仅仅只唤醒了一个”屏障内的线程”。
CyclicBarrier可重用，CountDownLatch不可重用，计数值为0该CountDownLatch就不可再用了。

理解其原理，区别自然是很简单就能说出来了。

Future和Callable

比较简单。跑一遍代码，使用它们的优点是，如果先执行A方法再执行B方法，但是B方法需要10S，A方法需要5S，传统方法等A执行完，再执行B，一共需要15s,造成了时间上的浪费，使用Future和Callable能够并行执行，自然不会存在这样的问题了,传统多线程也能做到，只是不好取的返回值。

package MyThread;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
/**
 * Created by yuhang on 2016/12/5.
 */
public class MyThread15 {
    public static class CallableThread implements Callable<String> {
        public String call() throws Exception {
            System.out.println("进入CallableThread的call()方法, 开始睡觉, 睡觉时间为" + System.currentTimeMillis());
            Thread.sleep(10000);
            return "123";
        }
    }
    public static void main(String[] args) throws Exception {
        ExecutorService es = Executors.newCachedThreadPool();
        CallableThread ct = new CallableThread();
        Future<String> f = es.submit(ct);
        es.shutdown();
        Thread.sleep(5000);
        System.out.println("主线程等待5秒, 当前时间为" + System.currentTimeMillis());
        String str = f.get();
        System.out.println("Future已拿到数据, str = " + str + ", 当前时间为" + System.currentTimeMillis());
    }
}

小结

多线程常用知识终于总结完了，其实也没那么复杂，对吧？还是那句话，坚持就是胜利！

多线程学习—4-线程池的使用

发表于 2016-12-04 | 分类于 Java多线程 | | 阅读次数

前言

写到线程池这块先回忆一段小插曲。13年12月的时候，差不多也是现在这个季节，刚实习一个月似懂非懂拿着活就干的我第一次接触到线程池，当时是做一个上报到的任务，差不多每天需要上传70万数据到一个接口。沈阳的大背景技术不是很强，工作了三四年的一些老程序员也很少接触到多线程。当时没用多线程之前差不多是

700000/(60*60*24*10)接近一天才能上报完，并且是程序不出错的情况下。于是乎白哥（我的第一个项目经理,这里要感谢白哥在我工作之初的时候谆谆教导！和第一份工作远在东北的同事们！）想到了用多线程来处理，也许是没时间研究底层，出来的效果也不尽人意，总是会丢数据，虽然时间加快到两个小时以内，不知天高地厚的我，当然也是不懂把失败策略，改了下在类似以下的代码中把原来的代码（原来代码不记得了）改为：

//线程池对拒绝任务的处理策略
  final RejectedExecutionHandler Errorhandler = new RejectedExecutionHandler() {
     public void rejectedExecution( Runnable r, ThreadPoolExecutor executor ) {
        if (!executor.isShutdown()) {
           r.run();
        }
        }
  };

其实就是把if语句内替换成r.run()这一句，居然奇迹般的不丢数据了，当时我第一次感到程序和多线程的魅力，后来浑浑噩噩的这么久整整三年竟然还是似懂非懂，说来惭愧！这几天继续学习多线程，刚好看到线程池这边，回忆起往事真是感慨颇多，多年前定的计划能完成的真是少之又少，希望以后能分而划之的坚持下去！

线程池的作用

线程池的作用

线程池的作用就2个：

减少了创建和销毁线程的次数，每个工作线程都可以被重复利用，可执行多个任务
可以根据系统的承受能力，调整线程池中工作线程的数据，防止因为消耗过多的内存导致服务器崩溃

使用线程池，要根据系统的环境情况，手动或自动设置线程数目。少了系统运行效率不高，多了系统拥挤、占用内存多。用线程池控制数量，其他线程排队等候。一个任务执行完毕，再从队列中取最前面的任务开始执行。若任务中没有等待任务，线程池这一资源处于等待。当一个新任务需要运行，如果线程池中有等待的工作线程，就可以开始运行了，否则进入等待队列。

可以适当的修改文末代码中的参数来体验一下速度或者在excute里加上sleep方法，感受下失败策略（即丢数据的情况），减小线程池数量大小可以加快速度，因为减少了线程池创建的时间，但是相应的降低了系统的处理能力。

参数总结

池中线程数小于corePoolSize，新任务都不排队而是直接添加新线程
池中线程数大于等于corePoolSize，workQueue未满，首选将新任务假如workQueue而不是添加新线程
池中线程数大于等于corePoolSize，workQueue已满，但是线程数小于maximumPoolSize，添加新的线程来处理被添加的任务
池中线程数大于大于corePoolSize，workQueue已满，并且线程数大于等于maximumPoolSize，新任务被拒绝，使用handler处理被拒绝的任务

上面的四个总结是前辈所写，由于这里主要是记录本人学习过程，就不添加链接了。

Executors

JDK推荐的三个Executors：

单线程线程池

那么线程池中运行的线程数肯定是1。workQueue选择了无界的LinkedBlockingQueue，那么不管来多少任务都排队，前面一个任务执行完毕，再执行队列中的线程。从这个角度讲，第二个参数maximumPoolSize是没有意义的，因为maximumPoolSize描述的是排队的任务多过workQueue的容量，线程池中最多只能容纳maximumPoolSize个任务，现在workQueue是无界的，也就是说排队的任务永远不会多过workQueue的容量，那maximum其实设置多少都无所谓了。
newFixedThreadPool(int nThreads) 固定大小线程池

固定大小的线程池和单线程的线程池异曲同工，无非是让线程池中能运行的线程编程了手动指定的nThreads罢了。同样，由于是选择了LinkedBlockingQueue，因此其实第二个参数maximumPoolSize同样也是无意义的
newCachedThreadPool() 无界线程池

无界线程池，意思是不管多少任务提交进来，都直接运行。无界线程池采用了SynchronousQueue，采用这个线程池就没有workQueue容量一说了，只要添加进去的线程就会被拿去用。既然是无界线程池，那线程数肯定没上限，所以以maximumPoolSize为主了，设置为一个近似的无限大Integer.MAX_VALUE。另外注意一下，单线程线程池和固定大小线程池线程都不会进行自动回收的，也即是说保证提交进来的任务最终都会被处理，但至于什么时候处理，就要看处理能力了。但是无界线程池是设置了回收时间的，由于corePoolSize为0，所以只要60秒没有被用到的线程都会被直接移除。

代码

package MyThread;
import java.util.List;
import java.util.Vector;
import java.util.concurrent.*;
/**
 * Created by yuhang on 2016/12/4.
 */
public class MyThread9 {
    //维护线程池最小的数量
    private static final int CORE_POOL_SIZE = 100;//改完1速度更快了，因为开启线程数量变少，但是相应的系统处理能力就变弱了
    //线程池维护的最大数量
    private static final int MAX_POOL_SIZE = 100;
    //线程池维护线程所允许的空余时间
    private static final int KEEP_ALIVE_TIME = 0;
    //缓冲队列的大小
    private static final int WORK_QUEUE_SIZE = 2000;
    static final List<Integer> l = new Vector<>();
    //线程池对拒绝任务的处理策略
    static final RejectedExecutionHandler Errorhandler = new RejectedExecutionHandler() {
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            if (!executor.isShutdown()) {
                //r.run();
                System.out.println("注释掉 r.run() WORK_QUEUE_SIZE 为100时，会出现丢数据的情况!");
            }
        }
    };
    //JDK提倡用Executors三个工厂方法
    //单线程线程池
//    public static ExecutorService newSingleThreadExecutor() {
//        return new FinalizableDelegatedExecutorService
//                (new ThreadPoolExecutor(1, 1,
//                        0L, TimeUnit.MILLISECONDS,
//                        new LinkedBlockingQueue<>()));
//    }
    //无界限
    public static ExecutorService newCachedThreadPool(int corePoolSize) {
        return new ThreadPoolExecutor(corePoolSize, Integer.MAX_VALUE,
                60L, TimeUnit.SECONDS,
                new SynchronousQueue<>());
    }
    //固定大小线程池
    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>());
    }
    static final ThreadPoolExecutor tp =
            new ThreadPoolExecutor(CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(WORK_QUEUE_SIZE), Errorhandler);
//            (ThreadPoolExecutor) MyThread9.newFixedThreadPool(1);
//            (ThreadPoolExecutor) MyThread9.newCachedThreadPool(5);
    public static void main(String[] args) {
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < 2000; i++) {
            tp.execute(
                    new Runnable() {
                        @Override
                        public void run() {
                            try {
                                Thread.sleep(10);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            l.add(1);
                        }
                    }
            );
            System.out.println(i);
        }
        tp.shutdown();
        try {
            tp.awaitTermination(1, TimeUnit.DAYS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(System.currentTimeMillis() - startTime);
        System.out.println(l.size());
    }
}

多线程学习—3-ThreadLocal的使用

发表于 2016-12-02 | 分类于 Java多线程 | | 阅读次数

前言

ThreadLocal使用比较简单，重点是要理解其源码，佩服写源码的大神，为了解决散列表的冲突而引入的神奇的hash code: 0x61c88647设计太巧妙了，数学白痴就不妄言分析了直接享用大神的成果吧，反正就是为了散列均匀。直接上使用代码。

ThreadLocal不是集合，它不存储任何内容，真正存储数据的集合在Thread中。ThreadLocal只是一个工具，一个往各个线程的ThreadLocal.ThreadLocalMap中table的某一位置set一个值的工具而已

每个线程都有一个自己的ThreadLocal.ThreadLocalMap对象
每一个ThreadLocal对象都有一个循环计数器
ThreadLocal.get()取值，就是根据当前的线程，获取线程中自己的ThreadLocal.ThreadLocalMap，然后在这个Map中根据第二点中循环计数器取得一个特定value值

与同步对比

同步与ThreadLocal是解决多线程中数据访问问题的两种思路，前者是数据共享的思路，后者是数据隔离的思路
同步是一种以时间换空间的思想，ThreadLocal是一种空间换时间的思想

代码

package MyThread;
/**
 * Created by yuhang on 2016/12/2.
 */
public class MyThread4 {
    private static ThreadLocal<Integer> seqNum
            = new ThreadLocal<Integer>() {
        public Integer initialValue() {
            return 0;
        }
    };
    public int getNextNum() {
        seqNum.set(seqNum.get() + 1);
        return seqNum.get();
    }
    public static void main(String[] args) {
        MyThread4 sn = new MyThread4();
        TestClient t1 = new TestClient(sn);
        TestClient t2 = new TestClient(sn);
        TestClient t3 = new TestClient(sn);
        t1.start();
        t2.start();
        t3.start();
    }
    private static class TestClient extends Thread {
        private MyThread4 sn;
        public TestClient(MyThread4 sn) {
            this.sn = sn;
        }
        public void run() {
            for (int i = 0; i < 3; i++) {
                System.out.println("thread[" + Thread.currentThread().getName()
                        + "]sn[" + sn.getNextNum() + "]");
            }
        }
    }
}

多线程学习—2-volatile的使用

发表于 2016-12-01 | 分类于 Java多线程 | | 阅读次数

前言

今天继续学习多线程，本来volatile关键字应该很好理解，不需要单独写博客记录。但是今天碰到些我暂时不能解释的问题，固贴出来交流一下。

volatile

根据JMM，Java中有一块主内存，不同的线程有自己的工作内存，同一个变量值在主内存中有一份，如果线程用到了这个变量的话，自己的工作内存中有一份一模一样的拷贝。每次进入线程从主内存中拿到变量值，每次执行完线程将变量从工作内存同步回主内存中。

被volatile修饰的变量，保证了每次读取到的都是最新的那个值。线程安全围绕的是可见性和原子性这两个特性展开的，volatile解决的是变量在多个线程之间的可见性，但是无法保证原子性。如下面代码所示，进入while循环时，如果不加volatile关键字，注释掉System.out.println(“running”);这行代码，程序是跳不出循环的，但是加上volatile关键字就能跳出了，这证明了volatile解决的是变量在多个线程之间的可见性。可是奇怪的是我将System.out.println(“running”);取消注释，即使不加上volatile也能跳出循环，还望明白的交流解释。我的测试环境是jdk 1.8

代码

package MyThread;
/**
 * Created by yuhang on 2016/12/1.
 */
public class MyThread3 extends Thread {
    //    volatile
    int a = 1;
    @Override
    public void run() {
        System.out.println("a: " + a);
        while (a == 1) {
            //  System.out.println("running");
        }
        System.out.println("while loop end");
    }
    public static void main(String[] args) {
        MyThread3 myThread3 = new MyThread3();
        myThread3.start();
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        myThread3.a = 2;
    }
}

多线程学习—1-wait(),notify(),notifyAll()

发表于 2016-11-30 | 分类于 Java多线程 | | 阅读次数

前言

最近准备把多线程的知识重新过一遍，决定从基础写起。这篇主要讲的是多线程的wait(),notify(),notifyAll()方法。

这三个方法都属于类方法，Object类都会有这三个方法，wait()是暂时让出锁给其他线程，而当前线程会一直阻塞直到等到notify(All)通知或者中断，而notify()和notifyAll()则是通知持有该锁的对象”等待获取该对象的对象锁”，因此必须要写在同步代码块内，至于为什么notify()和notifyAll()是通知”等待”，下面会详细解释。

wait

wait()的作用是使当前执行代码的线程进行等待，将当前线程置入”预执行队列”中，并且wait()所在的代码处停止执行，直到接到通知或被中断。在调用wait()之前，线程必须获得该对象的锁，因此只能在同步方法/同步代码块中调用wait()方法。

notify(All)

notify()的作用是，如果有多个线程等待，那么线程规划器随机挑选出一个wait的线程，对其发出通知notify()，并使它等待获取该对象的对象锁。注意”等待获取该对象的对象锁”，这意味着，即使收到了通知，wait的线程也不会马上获取对象锁，必须等待notify()方法的线程释放锁才可以。和wait()一样，notify()也要在同步方法/同步代码块中调用。

总结起来就是，wait()使线程停止运行，notify()使停止运行的线程继续运行。

代码

上面关于wait和notify的总结是前辈写的，下面给出我自己的示例。请大家根据注释，适当的去掉注释看看结果。

package MyThread;
/**
 * Created by yuhang on 2016/11/30.
 */
public class MyThread1 extends Thread {
    Object lock = new Object();
    public MyThread1(Object lock) {
        System.out.println("Constructor method->>CurrentThreadName:  " + Thread.currentThread().getName());
        System.out.println("Constructor method->>this.threadName:   " + this.getName());
        System.out.println("**********************************************");
        this.lock = lock;
    }
    @Override
    public void run() {
        synchronized (this.lock) {
            System.out.println("Run method->>CurrentThreadName:   " + Thread.currentThread().getName());
            System.out.println("Run method->>this.threadName:     " + this.getName());
            try {
                System.out.println("MyThread1:" + Thread.currentThread().getName() + " get the lock,and i will wait for some notify!");
                long statrTime = System.currentTimeMillis();
                //wait是让使用wait方法的对象等待，暂时先把对象锁给让出来，给其它持有该锁的对象用，其它对象用完后再告知（notify）等待的那个对象可以继续执行了
                this.lock.wait();
                System.out.println("MyThread1" + Thread.currentThread().getName() + " wait end! lock time : " + (System.currentTimeMillis() - statrTime) / 1000);
            } catch (InterruptedException e) {
                System.out.println("wait be interrupted by interrupt method just now!");
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args) {
        Object lock = new Object();
        MyThread1 myThread1 = new MyThread1(lock);
        MyThread1 myThread2 = new MyThread1(lock);
        myThread1.start();
        myThread2.start();
        //wait和notify必须要写在同步代码块内部 否则会报java.lang.IllegalMonitorStateException异常
        synchronized (lock) {
            try {
                Thread.sleep(3000);
                lock.notifyAll();
                //lock.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

package MyThread;
/**
 * Created by yuhang on 2016/11/30.
 */
public class MyThread2 extends Thread {
    Object lock = new Object();
    public MyThread2(Object lock) {
        this.lock = lock;
    }
    @Override
    public void run() {
        synchronized (lock) {
            System.out.println(Thread.currentThread().getName() + "will notify!");
            //notify不会马上释放锁，会等到当前同步代码块执行完，才会释放锁
            lock.notify();
            System.out.println(Thread.currentThread().getName() + " notified ,but didn't release!");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + " after sleep,will release the lock!");
        }
    }
    public static void main(String[] args) {
        Object lock = new Object();
        MyThread1 myThread1 = new MyThread1(lock);
        myThread1.start();
        //测试Notify不释放锁
        MyThread2 myThread2 = new MyThread2(lock);
        //myThread2.start();
        //测试interrupt方法
//        try {
//            Thread.sleep(10000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
//        System.out.println(myThread1.getName() + "will execute interrupt method!");
//        myThread1.interrupt();
        try {
            //join方法会释放锁,但是不加时间会一直阻塞当前执行的线程直到myThead1的run方法执行完成
            myThread1.join();//.join(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //如果没有Join方法myThead2线程是不会启动的
        myThread2.start();
        System.out.println("i will print after myThead1 complete!");
    }
}

测试下分类和标签

发表于 2016-10-16 | 分类于测试 | | 阅读次数

测试分类和标签

代码块

public static void main(String args[]){
    System.out.printl("Hello World!");
}

Yuhang

我不曾有一刻敢去懈怠。

GitHub