六万字【Java高并发编程入门】第十九篇：并发编程入门总览包教包会【值得收藏】

public class VisibleProblemTest {
    private static Boolean aBoolean = false;
    public static void main(String[] args) throws InterruptedException {
        new Thread(() -> {
            int i = 0;
            System.out.println("VisibleProblemTest.aBoolean====" + VisibleProblemTest.aBoolean);
            while (VisibleProblemTest.aBoolean != true) {
                i++;
            }
            System.out.println("i====" + i);
        }).start();
        //主线程休息2秒再操作共享变量，便于人眼可以直接看到
        Thread.sleep(2000);
        //修改共享变量值
        VisibleProblemTest.aBoolean = true;
        System.out.println("变量值修改");
        System.out.println("程序结束");
    }
}

public class AtomicTest {
    private  int atest = 0;


    public void countTest() {
        for (int i = 0; i < 10000; i++) {
            atest = atest + 1;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        final AtomicTest atomicTest = new AtomicTest();
        Thread threadA = new Thread(() -> atomicTest.countTest());
        Thread threadB = new Thread(() -> atomicTest.countTest());
        threadA.start();
        threadB.start();
        threadA.join();
        threadB.join();
        System.out.println("*******获得到的atest值为=" + atomicTest.atest);
    }
}

public class SingletonDemo {
    private static  SingletonDemo instance = null;

    private SingletonDemo(){

    }

    static SingletonDemo getSingletonDemo() {
        if (instance == null) {
            synchronized (SingletonDemo.class) {
                if (instance == null) {
                    instance = new SingletonDemo();    //6
                }
            }
        }
        return instance;
    }
}

 public void writer() {
        a = 43;
        b = a;
    }

   private volatile  boolean flag = false;
    private   int a = 0;

    public void writer() {
        a = 43;
        flag = true;
    }
    public void read() {
        if (flag) {
            System.out.println("读取到的a值是="+a);
        }
    }

   synchronized int getValue() {
        return value;
    }

    public void transfer() {
        c = a + b;  //1
        a = 100;  //2
        c = a - b;//3
    }

  new Thread(new Runnable() {
            public void run() {
                longTest.countTest();
            }
        }).start();

 final LongTest longTest = new LongTest();
        Thread threadA = new Thread(new Runnable() {
            public void run() {
                longTest.countTest();
            }
        });
        Thread threadB = new Thread(new Runnable() {
            public void run() {
                longTest.countTest();
            }
        });
        threadA.start(); //1
        threadA.join();  //2
        threadB.start(); //3
        threadB.join(); //4

    public static void main(String[] args) throws InterruptedException {
        Thread threadA = new Thread("线程A");
        threadA.start();
        threadA.interrupt();
        System.out.println("线程A是否终止"+threadA.isInterrupted());
    }

   synchronized (this) {
            this.a = 100;
            System.out.println("*******test2执行");
        }

class SynchronizedTest {  
     public synchronized  void test1() {
        System.out.println("*******test1执行");
    }
    public void test2() {
        synchronized (this) {
            System.out.println("*******test2执行");
        }
    }
    public synchronized static void test3() {
        System.out.println("*******test3执行");
    }
}

public class SynchronizedTest3 {
    int a = 0;
    int b = 0;
    static int c = 0;

    /**
     * 锁定的是this对象，保护了,a,b两个资源
     */
    synchronized void setValue() {
        a = 100;
        b = 20;
    }

    /**
     * 锁定的是SynchronizedTest3的class对象，
     * 保护了资源c
     * @return
     */
    synchronized static int getValue() {
        return SynchronizedTest3.c;
    }

}

//能够响应中断
 void lockInterruptibly() throws InterruptedException;
 //能够响应超时
  boolean tryLock(long time, TimeUnit unit) throws InterruptedException
  //能够非阻塞的获取锁
  boolean tryLock()

public class LockTest2 {
    final Lock lock = new ReentrantLock();
    int value = 0;

    public void addOne() {
        lock.lock();
        try {
            value + = 1;
        } finally {
            lock.unlock();
        }

    }
}

public class LockTest {

    public static void main(String[] args) {
        MyThreadService myThreadService = new MyThreadService();
        for (int i = 0; i < 5; i++) {
            MyThread myThread = new MyThread("线程" + i, myThreadService);
            myThread.start();
        }
    }

    static class MyThread extends Thread {
        MyThreadService myThreadService = null;

        public MyThread(String name, MyThreadService myThreadService) {
            super(name);
            this.myThreadService = myThreadService;
        }

        @Override
        public void run() {
            myThreadService.printThread();
        }
    }
}


public class MyThreadService {
    final Lock lock = new ReentrantLock();

    public void printThread() {
        lock.lock();
        try {
            for (int i = 0; i < 5; i++) {
                System.out.println("*******" + Thread.currentThread().getName() + (" " + i));
            }
        } finally {
            lock.unlock();
        }
    }

}

public class Allocator {
    private List<Object> applyList = new ArrayList<Object>();

    private final static Allocator allocator = new Allocator();

    private Allocator() {

    }

    /**
     * 只能由一个人完成，所以是单例模式
     * @return
     */
    public static Allocator getAllocator() {
        return allocator;
    }

    /**
     * 申请资源
     */
    synchronized boolean applyResource(Object from, Object to) {
        if (applyList.contains(from) ||
                applyList.contains(to)) {
            return false;
        }
        applyList.add(from);
        applyList.add(to);
        return true;
    }

    /**
     * 释放资源
     */
    synchronized void free(Object from, Object to) {
        applyList.remove(from);
        applyList.remove(to);
    }

public void eat(Account2 target) {
        //没有申请到锁就一直循环下去,直到成功
        while (!Allocator.getAllocator().applyResource(this, target)) {
            return;
        }
        try {
            //左边
            synchronized (this) {
                //右边
                synchronized (target) {

                }
            }
        } finally {
            //释放已经申请的资源
            Allocator.getAllocator().free(this, target);
        }
    }

class Philosopher extends Thread{
    private ReentrantLock left,right;

    public Philosopher(ReentrantLock left, ReentrantLock right) {
        super();
        this.left = left;
        this.right = right;
    }
    public void run(){
        try {
            while(true){
                Thread.sleep(1000);//思考一段时间
                left.lock();
                try{
                    if(right.tryLock(1000,TimeUnit.MILLISECONDS)){
                        try{
                            Thread.sleep(1000);//进餐一段时间
                        }finally {
                            right.unlock();
                        }
                    }else{
                        //没有获取到右手的筷子，放弃并继续思考
                    }
                }finally {
                    left.unlock();
                }
            }
        } catch (InterruptedException e) {
        }
    }
}

   static class Chopsticks {
        private int id;

        public Chopsticks(int id) {
            this.id = id;
        }

        public int getId() {
            return id;
        }
    }

     public Philosopher(Chopsticks left, Chopsticks right) {
            if (left.getId() > right.getId()) {
                this.left = right;
                this.right = left;
            } else {
                this.left = left;
                this.right = right;
            }
        }

//获得在i位置上的Node节点
  static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
        return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
    }
    //利用CAS算法设置i位置上的Node节点，之所以能实现并发是因为他指定了原来这个节点的值是多少。
    //在CAS算法中，会比较内存中的值与指定的值是是否相等，如果相等则更新，并返回true，如果不相等则不更新，直接返回false。
    static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
                                        Node<K,V> c, Node<K,V> v) {
        return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
    }
    //设置节点位置的值，仅在上锁区被调用
    static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
        U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
    }

 private final Node<K,V>[] initTable() {
        Node<K,V>[] tab; int sc;
        while ((tab = table) == null || tab.length == 0) {
        //sizeCtl 表示有其他线程正在进行初始化操作，把线程挂起，对于table的初始化操作，只能有一个线程进行
            if ((sc = sizeCtl) < 0)
                Thread.yield(); // lost initialization race; just spin
            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {  //利用CAS方法把sizeCtl的值设置为-1，表示本线程正在进行初始化
                try {
                    if ((tab = table) == null || tab.length == 0) {
                        int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                        table = tab = nt;
                        sc = n - (n >>> 2);  //相当于0.75*n  设置一个扩容的阈值
                    }
                } finally {
                    //将sc赋值给sizeCtl
                    sizeCtl = sc;
                }
                break;
            }
        }
        return tab;
    }

    private transient volatile int sizeCtl; //控制标识符

new Thread(new Runnable() {
            public void run() {
                longTest.countTest();
            }
        }).start()

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

 public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>(),
                                      threadFactory);
    }

 ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("测试-%s").build();

 public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

   public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.acc = System.getSecurityManager() == null ?
                null :
                AccessController.getContext();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

      ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("测试-%d").build();
        ExecutorService executorService = new ThreadPoolExecutor(10, 15, 1, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(10), threadFactory, new ThreadPoolExecutor.CallerRunsPolicy());

public class ExectorTest {
    public static void main(String[] args) {
       //给线程设置一个自定义名称
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("测试线程-%d").build();
        ThreadPoolExecutor executorService = new ThreadPoolExecutor(
                3,
                6,
                10,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(5),
                threadFactory
//                , new ThreadPoolExecutor.CallerRunsPolicy()
        );
        for (int i=0;i<20;i++) {
            executorService.execute(()->{
               //模拟耗时的任务
                System.out.println(Thread.currentThread().getName()+" 开始执行任务");
                int j = 10000 * 10000;
                while (j >0) {
                    j--;
                }

                System.out.println(Thread.currentThread().getName()+" 执行结束");

            });
        }
    }
}

    public void execute(Runnable command) {
    //如果任务为空，则抛出NPE异常
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
         //获取线程池状态
        int c = ctl.get();
        //1.如果工作线程的数量小于核心线程数
        if (workerCountOf(c) < corePoolSize) {
            //调用addWorker增加一个新线程，并执行一个任务
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        //如果线程池的状态是运行状态，并且任务加入到了工作队列成功
        if (isRunning(c) && workQueue.offer(command)) {
            //双重检查，再次检查线程池的状态。
            int recheck = ctl.get();
            //如果线程池的状态不是运行状态并且移除任务成功则调用拒绝策略
            if (! isRunning(recheck) && remove(command))
                //调用RejectedExecutionHandler.rejectedExecution()方法。根据不同的拒绝策略去处理
                 reject(command);
            //如果工作线程的数量为0，说明工作队列中可能有任务没有线程执行，此时则新建一个线程来执行任务，由于执行的是队列中已经堆积的任务，所以没有传入具体的任务。
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        //如果前面的新增work，放入队列都失败，则会继续新增worker，此时线程池中的工作线程数达到corePoolSize，阻塞队列任务已满，只能基于maximumPoolSize来继续增加work，如果还是失败
        else if (!addWorker(command, false))
           //如果还是失败，则调用RejectedExecutionHandler.rejectedExecution()方法。根据不同的拒绝策略去处理
            reject(command);
    }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
           //1. 新建一个工作线程，Work后面会说
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                //加锁
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    //再次获取线程池的状态，在新建线程或者释放锁时，都会重新检查。
                    int rs = runStateOf(ctl.get());
                    //如果线程池的状态不是关闭状态，则进入下面的分支
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        //检查新建的线程是否是可运行状态
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        //将工作线程添加到HashSet类型的集合中
                        workers.add(w);
                        int s = workers.size();
                        //如果工作线程的集合数大于largestPoolSize
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        //新建工作线程成功之后，将操作标志workerAdded设为true，表示新增工作线程成功，后续流程用
                        workerAdded = true;
                    }
                } finally {
                    //释放锁
                    mainLock.unlock();
                }
                //如果新建工作线程成功，则调用start() 方法启动线程
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            //如果workerStarted为false，表示新建工作线程失败
            if (! workerStarted)
                //移除已经创建的工作线程
                addWorkerFailed(w);
        }
        return workerStarted;

public class ThreadLocalTest {
    public static void main(String[] args) {
        ThreadLocal<String> threadLocal = new ThreadLocal<>();
        threadLocal.set("张三");

        new Thread(()->{
            threadLocal.set("李四");
            System.out.println("*******"+Thread.currentThread().getName()+"获取到的数据"+threadLocal.get());
        },"线程1").start();
        new Thread(()->{
            threadLocal.set("王二");
            System.out.println("*******"+Thread.currentThread().getName()+"获取到的数据"+threadLocal.get());
        },"线程2").start();
        new Thread(()->{
            System.out.println("*******"+Thread.currentThread().getName()+"获取到的数据"+threadLocal.get());
        },"线程3").start();
        System.out.println("线程=" + Thread.currentThread().getName() + "获取到的数据=" + threadLocal.get());
    }
}

    public void set(T value) {
        Thread t = Thread.currentThread();
        ThreadLocalMap map = getMap(t);
        if (map != null)
            map.set(this, value);
        else
            createMap(t, value);
    }

public class Thread implements Runnable {

    //每个线程自己的ThreadLocalMap对象通过ThreadLocal保存下来
  ThreadLocal.ThreadLocalMap threadLocals = null;

  ThreadLocalMap getMap(Thread t) {
        return t.threadLocals;
    }
}

    public T get() {
        Thread t = Thread.currentThread();
        ThreadLocalMap map = getMap(t);
        if (map != null) {
            ThreadLocalMap.Entry e = map.getEntry(this);
            if (e != null) {
                @SuppressWarnings("unchecked")
                T result = (T)e.value;
                return result;
            }
        }
        return setInitialValue();
    }

public class ThreadLocal<T> {

    private static AtomicInteger nextHashCode =new AtomicInteger();

    //初始的Hash值是0x61c88647
     private static final int HASH_INCREMENT = 0x61c88647;

    //每次调用就原子性的将hash值增加HASH_INCREMENT
    private static int nextHashCode() {
        return nextHashCode.getAndAdd(HASH_INCREMENT);
    }

    static class ThreadLocalMap {
         //Entry继承WeakReference
          static class Entry extends WeakReference<ThreadLocal<?>> {
            /** The value associated with this ThreadLocal. */
            Object value;

            Entry(ThreadLocal<?> k, Object v) {
                super(k);
                value = v;
            }
        }

                private static final int INITIAL_CAPACITY = 16;


                 private void setThreshold(int len) {
                    threshold = len * 2 / 3;
                }


            void createMap(Thread t, T firstValue) {
                t.threadLocals = new ThreadLocalMap(this, firstValue);
            }

             ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) {
                    table = new Entry[INITIAL_CAPACITY];
                    int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1);
                    table[i] = new Entry(firstKey, firstValue);
                    size = 1;
                    setThreshold(INITIAL_CAPACITY);
                }
            }
    }

  private void set(ThreadLocal<?> key, Object value) {
            Entry[] tab = table;
            int len = tab.length;
            //1.根据ThreadLocal对象的hash值，定位到table中的位置i
            int i = key.threadLocalHashCode & (len-1);

            for (Entry e = tab[i];
                 e != null;
                 e = tab[i = nextIndex(i, len)]) {
                ThreadLocal<?> k = e.get();
                //判断Entry.key等于当前的ThreadLoacl对象key，则覆盖旧值，退出。
                if (k == key) {
                    e.value = value;
                    return;
                }

                if (k == null) {
                    replaceStaleEntry(key, value, i);
                    return;
                }
            }

            tab[i] = new Entry(key, value);
            int sz = ++size;
            if (!cleanSomeSlots(i, sz) && sz >= threshold)
                rehash();
        }

     private Entry getEntry(ThreadLocal<?> key) {
            int i = key.threadLocalHashCode & (table.length - 1);
            Entry e = table[i];
            if (e != null && e.get() == key)
                return e;
            else
                return getEntryAfterMiss(key, i, e);
        }

         private Entry getEntryAfterMiss(ThreadLocal<?> key, int i, Entry e) {
            Entry[] tab = table;
            int len = tab.length;

            while (e != null) {
                ThreadLocal<?> k = e.get();
                if (k == key)
                    return e;
                if (k == null)
                    expungeStaleEntry(i);
                else
                    i = nextIndex(i, len);
                e = tab[i];
            }
            return null;
        }

      static class Entry extends WeakReference<ThreadLocal<?>> {
            /** The value associated with this ThreadLocal. */
            Object value;

            Entry(ThreadLocal<?> k, Object v) {
                super(k);
                value = v;
            }
        }