[Java][华为云Java编程创造营][学习笔记][第三阶段][05_Java多线程实战][05_JUC并发包]

5.1，线程的ThreadLocal本地缓存对象

• ThreadLocal线程范围内的共享变量：

  • 线程范围内的共享变量，每个线程只能自己的数据，不能访问别的线程的数据。

• 每个线程调用全局ThreadLocal对象的set方法，就相当于往其内部的map中增加一条记录，key分别是各自的线程，value是各自的set方法传进去的值。

• ThreadLocal以内存换安全

5.2，线程的volatile关键字

• volatile关键字可以用来修饰字段（成员变量），就是告知程序任何对该变量的访问均需要从共享内存中获取，而对它的改变必须同步刷新回共享内存，它能保证所有线程对变量访问的可见性。

class UserThread extends Thread
{
    private volatile boolean flag = false;

    public boolean isFlag()
    {
        return flag;
    }

    public void setFlag(boolean flag)
    {
        this.flag = flag;
    }

    @Override
    public void run()
    {
        System.out.println(Thread.currentThread().getName() + "，线程正在运行");
        while (flag)
        {

        }
        System.out.println(Thread.currentThread().getName() + "，线程运行结束");
    }
}

public class Test
{
    public static void main(String[] args)
    {
        UserThread ut = new UserThread();
        ut.start();

        try
        {
            Thread.sleep(5000);
        } catch (InterruptedException e)
        {
            e.printStackTrace();
        }

        ut.setFlag(false);
        /*
        * 输出结果
        *   Thread-0，线程正在运行
            Thread-0，线程运行结束 
        * */
    }
}

• volatile的作用：使变量在多个线程之间可见，但是无法保证原子性。
• 需要注意的是一般volatile用于只针对多个线程可见的变量操作，并不能代替synchronized的同步功能。

public class ThreadVolatile extends Thread
{
    public static volatile int n = 0;

    @Override
    public void run()
    {
        for (int i = 0; i < 10; i++)
        {
            try
            {
                n += 1;
                //为了使运行结果更随机，延迟3毫秒
                sleep(3);
            } catch (InterruptedException e)
            {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws Exception
    {
        Thread threads[] = new Thread[100];
        for (int i = 0; i < threads.length; i++)
        {
            //建立100个线程
            threads[i] = new ThreadVolatile();
        }
        for (int i = 0; i < threads.length; i++)
        {
            //运行刚才建立的100个线程
            threads[i].start();
        }
        for (int i = 0; i < threads.length; i++)
        {
            //100个线程都执行完后继续
            threads[i].join();
        }
        System.out.println("n= " + ThreadVolatile.n);//n= 909
    }
}

5.3，线程池的作用和应用

• 线程池的作用：

  1. 降低资源消耗。
     • 通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
  2. 提高响应速度。
     • 当任务到达时，任务可以不需要等到线程创建就能立即执行。
  3. 提高线程的可管理性。
     • 线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。

• 线程池的应用

  • 场景：请求频繁，考虑到服务的并发问题，如果每个请求来到后，服务都为它启动一个线程，那么这对服务的资源可能会造成很大的浪费。

5.4，线程的同步工具类CountDownLatch

• CountDownLatch同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。

  1. CountDownLatch类是一个同步计数器，构造时传入int参数，该参数就是计数器的初始值，每调用countDown()方法，计数器减1，计数器大于0时，await()方法会阻塞程序继续执行。
  2. 由于调用了countDown()方法，所以在当前计数到达0之前，await()方法会一直受阻塞。之后，会释放所有等待的线程，await()的所有后续调用都将立即返回。这种现象只出现一次，计数无法被重置。一个线程或者多个，等待另外N个线程完成某个事情之后才能执行。

• CountDownLatch最重要的方法是countDown()和await()

import java.util.concurrent.CountDownLatch;

class UserThread1 extends Thread
{
    private CountDownLatch cd;
    private int sum1;

    public UserThread1(CountDownLatch cd)
    {
        this.cd = cd;
    }

    @Override
    public void run()
    {
        for (int i = 0; i <= 100; i++)
        {
            sum1 += i;
        }
        cd.countDown();
    }

    public int getSum1()
    {
        return sum1;
    }
}

class UserThread2 extends Thread
{
    private CountDownLatch cd;
    private int sum2;

    public UserThread2(CountDownLatch cd)
    {
        this.cd = cd;
    }

    @Override
    public void run()
    {
        for (int i = 101; i <= 200; i++)
        {
            sum2 += i;
        }
        cd.countDown();
    }

    public int getSum2()
    {
        return sum2;
    }
}

public class Test
{
    public static void main(String[] args)
    {
        CountDownLatch cd = new CountDownLatch(2);
        UserThread1 u1 = new UserThread1(cd);
        UserThread2 u2 = new UserThread2(cd);

        u1.start();
        u2.start();

        try
        {
            cd.await();
            int sum = u1.getSum1() + u2.getSum2();
            System.out.println("两个线程计算的和为：" + sum);//两个线程计算的和为：20100
        } catch (InterruptedException e)
        {
            e.printStackTrace();
        }
    }
}

5.5，线程的同步工具类CyclicBarrier

• CyclicBarrier是一个同步辅助类，允许一组线程互相等待，直到到达某个公共屏障点（common barrier point）。因为该barrier在释放等待线程后可以重用，所以称它为循环的barrier。

5.6，线程的同步工具类semaphore

• Semaphore是一个计数信号量，它的本质是一个共享锁，是基于AQS实现的，通过state变量来实现共享。通过调用acquire方法，对state值减一，当调用release对state值加一。当state变量小于0时，在AQS队列中阻塞等待。

import java.util.concurrent.Semaphore;

class Car extends Thread
{
    private Address address;

    public Car(Address address)
    {
        this.address = address;
    }

    @Override
    public void run()
    {
        this.address.autoCar();
    }
}

class Address
{
    //每次停车的数量
    private int num;
    //信号量
    private Semaphore sm;

    public Address(int num)
    {
        this.num = num;
        sm = new Semaphore(this.num);
    }

    public void autoCar()
    {
        try
        {
            //加锁
            sm.acquire();
            System.out.println(Thread.currentThread().getName() + "，进入停车场");
            //模拟停车时间
            Thread.sleep(3000);
            System.out.println(Thread.currentThread().getName() + "，离开停车场");
            //释放锁
            sm.release();
        } catch (InterruptedException e)
        {
            e.printStackTrace();
        }
    }
}

public class Test
{
    public static void main(String[] args)
    {
        Address address = new Address(2);
        for (int i = 0; i < 5; i++)
        {
            new Car(address).start();
        }
        /*
        * 输出结果
        *   Thread-0，进入停车场
            Thread-1，进入停车场
            Thread-1，离开停车场
            Thread-0，离开停车场
            Thread-2，进入停车场
            Thread-3，进入停车场
            Thread-3，离开停车场
            Thread-2，离开停车场
        * */
    }
}

5.7，线程的交换类Exchanger

• Exchanger（交换者）是一个用于线程间协作的工具类，Exchanger用于进行线程间的数据交换。
• 它提供一个同步点，在这个同步点，两个线程可以交换彼此的数据。
• 这两个线程通过exchanger()交换数据。
• 如果第一个线程先执行exchange()方法，它会一直等待第二个线程也执行exchange()，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。

import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Test
{
    private static final Exchanger<String> expr = new Exchanger<>();
    private static ExecutorService threadpool = Executors.newFixedThreadPool(2);

    public static void main(String[] args)
    {
        threadpool.execute(new Runnable()
        {
            @Override
            public void run()
            {
                //A录入银行流水数据
                try
                {
                    String A = "银行流水A";
                    String B = expr.exchange(A);
                    System.out.println("Thread A:" + B);
                } catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
            }
        });
        threadpool.execute(new Runnable()
        {
            @Override
            public void run()
            {
                //B录入银行流水数据
                try
                {
                    String B = "银行流水B";
                    String A = expr.exchange(B);
                    System.out.println("Thread B:" + A);
                } catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
            }
        });
        threadpool.shutdown();
        /*
        * 输出结果
        *   Thread B:银行流水A
            Thread A:银行流水B
        * */
    }
}

5.8，线程的Fork-Join机制

• Fork/Join框架是Java7提供一个用于并行执行任务的框架，是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。
• 分治法：把一个规模大的问题划分为规模较小的子问题，然后分而治之，最后合并子问题的解得到原问题的解。
• Fork/Join案例：

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;

class CountTask extends RecursiveTask<Integer>
{
    private int start;
    private int end;

    //计算任务量的值
    private static final int TASKSIZE = 2;
    private static int count = 0;

    public CountTask(int start, int end)
    {
        this.start = start;
        this.end = end;
    }

    @Override
    protected Integer compute()
    {
        int sum = 0;

        System.out.println("开启线程进行计算：" + count++);
        boolean flag = (this.end - this.start) <= TASKSIZE;
        //如果是小于等于任务的值
        if (flag)
        {
            //没有必要拆分任务计算
            for (int i = start; i <= end; i++)
            {
                sum += i;
            }
        } else
        {
            //要进行拆分任务计算
            System.out.println("这个任务需要进行拆分任务进行计算...." + Thread.currentThread().getName());
            //任务大于值，分裂为两个任务 10+1/2
            int middle = (start + end) / 2;
            CountTask countTask1 = new CountTask(start, middle);
            CountTask countTask2 = new CountTask(middle + 1, end);
            //开启线程计算分布式任务
            invokeAll(countTask1, countTask2);
            Integer taskSum1 = countTask1.join();
            Integer taskSum2 = countTask2.join();
            //结果合并
            sum = taskSum1 + taskSum2;
        }
        return sum;
    }
}

public class Test
{
    public static void main(String[] args)
    {
        //分布式计算的池子
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        //初始化设置任务
        CountTask countTask = new CountTask(1, 10);
        //分布式计算任务，提交任务
        ForkJoinTask forkJoinTask = forkJoinPool.submit(countTask);
        //最后得到计算结果
        try
        {
            System.out.println(forkJoinTask.get());
        } catch (InterruptedException e)
        {
            e.printStackTrace();
        } catch (ExecutionException e)
        {
            e.printStackTrace();
        }
    }
}

5.9，线程的组合案例购票

/*
 * 线程并发购票
 * */
class Tickets
{
    private int allowance;

    public Tickets(int allowance)
    {
        this.allowance = allowance;
    }

    public int getAllowance()
    {
        return allowance;
    }

    public void setAllowance(int allowance)
    {
        this.allowance = allowance;
    }

    //购票方法
    public void buyTickets(int num)
    {
        synchronized (this)
        {
            int before = this.allowance;
            int after = before - num;
            try
            {
                Thread.sleep(1000);
            } catch (InterruptedException e)
            {
                e.printStackTrace();
            }
            this.setAllowance(after);
        }
    }
}

class CustomerRunnable implements Runnable
{
    private Tickets tickets;

    public CustomerRunnable(Tickets tickets)
    {
        this.tickets = tickets;
    }

    @Override
    public void run()
    {
        tickets.buyTickets(1);
        System.out.println(Thread.currentThread().getName() + "购票成功，余票：" +
                tickets.getAllowance());
    }
}

public class Test
{
    public static void main(String[] args)
    {
        //创建tickets对象
        Tickets tickets = new Tickets(5);
        //创建CustomerRunnable数组
        CustomerRunnable[] customerRunnables = new CustomerRunnable[5];
        //创建Thread数组
        Thread[] threads = new Thread[5];
        //创建100个CustomerRunnable对象
        for (int i = 0; i < customerRunnables.length; i++)
        {
            customerRunnables[i] = new CustomerRunnable(tickets);
        }
        //创建100个线程
        for (int i = 0; i < threads.length; i++)
        {
            threads[i] = new Thread(customerRunnables[i]);
            threads[i].start();
        }
        /*
        * 输出结果
        *   Thread-1购票成功，余票：4
            Thread-4购票成功，余票：3
            Thread-2购票成功，余票：2
            Thread-0购票成功，余票：1
            Thread-3购票成功，余票：0
        * */
    }
}

5.10，线程的组合案例购物

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/*
 * 线程并发秒杀购物
 * */
class Shop
{
    //信号量
    private Semaphore semaphore;

    public Shop(int num)
    {
        //实例化信号量，每次可以有5个线程获得许可
        semaphore = new Semaphore(num);
    }

    //用户去抢购
    public void userShopping(String name)
    {
        boolean flag = false;
        try
        {
            //怎么限流抢购，什么时候算抢购成功，还需要入库
            flag = this.semaphore.tryAcquire(1, TimeUnit.SECONDS);
            if (flag)
            {
                System.out.println(name + "，抢购成功，可以下单了");
                TimeUnit.SECONDS.sleep(1);
            } else
            {
                System.out.println(name + "，对不起，抢购没有成功，请重试。。。");
            }
        } catch (InterruptedException e)
        {
            e.printStackTrace();
        } finally
        {
            if (flag)
            {
                //抢购成功，一定要注意释放
                this.semaphore.release();
            }
        }
    }
}

class User extends Thread
{
    private String sname;
    private Shop shop;

    public User(String sname, Shop shop)
    {
        super();
        this.sname = sname;
        this.shop = shop;
    }

    @Override
    public void run()
    {
        this.shop.userShopping(sname);
    }

    public String getSname()
    {
        return sname;
    }

    public void setSname(String sname)
    {
        this.sname = sname;
    }

    public Shop getShop()
    {
        return shop;
    }

    public void setShop(Shop shop)
    {
        this.shop = shop;
    }
}

public class Test
{
    public static void main(String[] args)
    {
        Shop shop = new Shop(5);
        for (int i = 0; i < 5; i++)
        {
            new User("张" + i, shop).start();
        }
    }
    /*
    * 输出结果
    *   张0，抢购成功，可以下单了
        张4，抢购成功，可以下单了
        张2，抢购成功，可以下单了
        张1，抢购成功，可以下单了
        张3，抢购成功，可以下单了
    * */
}

5.11，线程的锁的synchronized和Lock、volatile区别

• synchronized和volatile区别：

  1. volatile关键字解决的是变量在多个线程之间的可见性；而synchronized关键字解决的是多个线程之间访问共享资源的同步性。
  2. volatile只能用于修饰变量，而synchronized可以修饰方法，以及代码块。
  3. 多线程访问volatile不会发生阻塞，而synchronized会出现阻塞。
  4. volatile能保证变量在多个线程之间的可见性，但不能保证原子性；而synchronized可以保证原子性，也可以间接保证可见性，因为它会将私有内存和公有内存中的数据做同步。

• synchronized和Lock区别：

  1. Lock是一个接口，而synchronized是Java中的关键字，synchronized是内置的语言实现。
  2. synchronized在发生异常时，会自动释放线程占有的锁，因此不会导致死锁现象的出现；而Lock在发生异常时，如果没有主动通过unlock()去释放锁，则很可能造成死锁现象，因此使用Lock时需要在finally块中释放锁。
  3. Lock可以提高多个线程进行读操作的效率（读写锁）。

5.12，线程的读写分离机制

• ReadWriteLock顾名思义，是读写锁：它维护了一对相关的锁"读取锁"和"写入锁"，一个用于读取操作，一个用于写入操作。
• 读取锁用于只读操作，它是共享锁，能同时被多个线程获取。
• 写入锁用于写入操作，它是独占锁，写入锁只能被一个线程锁获取。
• 不能同时存在读取锁和写入锁！可以读/读，但不能读/写，写/写。