线程和进程

一个进程往往可以包含多个线程,至少包含一个!
Java默认有几个线程? 2 个: mian、GC
对于Java而言:Thread、Runnable、Callable 三种实现线程的方式。
JAVA不能开启线程,是调用本地方法,查看start方法可以看到底层是C++来开启线程的

并发和并行

  1. 并发(多线程操作同一个资源):CPU 一核 ,模拟出来多条线程,天下武功,唯快不破,快速交替。
  2. 并行(同时多条线程同时执行):CPU 多核 ,多个线程可以同时执行; 线程池。

并发编程的本质:充分利用CPU的资源

线程的6种状态

public enum State {    
    // 新生    
    NEW,
    // 运行    
  RUNNABLE,
    // 阻塞    
    BLOCKED,        
    // 等待,死死地等    
    WAITING,
    // 超时等待    
    TIMED_WAITING,
    // 终止    
    TERMINATED;
}

wait和sleep的区别

  1. 来自不同的类 wait => Object sleep => Thread
  2. 关于锁的释放 wait 会释放锁,sleep 睡觉了,抱着锁睡觉,不会释放!
  3. 范围不同:wait必须在同步代码块,sleep可以任意地方。

synchronized和lock

synchronized方式上锁

/**
 * 真正的多线程开发,公司中的开发,降低耦合性
 * 线程就是一个单独的资源类,没有任何附属的操作!
 * 1、 属性、方法
 */
public class SaleTicketDemo01 {
    public static void main(String[] args) {
        // 并发:多线程操作同一个资源类, 把资源类丢入线程
        Ticket ticket = new Ticket();

        // @FunctionalInterface 函数式接口,jdk1.8  lambda表达式 (参数)->{ 代码 }
        new Thread(()->{
            for (int i = 1; i < 40 ; i++) {
                ticket.sale();
            }
        },"A").start();

        new Thread(()->{
            for (int i = 1; i < 40 ; i++) {
                ticket.sale();
            }
        },"B").start();

        new Thread(()->{
            for (int i = 1; i < 40 ; i++) {
                ticket.sale();
            }
        },"C").start();


    }
}

// 资源类 OOP
class Ticket {
    // 属性、方法
    private int number = 30;

    // 卖票的方式
    // synchronized 本质: 队列,锁
    public synchronized void sale(){
        if (number>0){
            System.out.println(Thread.currentThread().getName()+"卖出了"+(number--)+"票,剩余:"+number);
        }
    }

}

lock方式上锁

public class SaleTicketDemo02  {
    public static void main(String[] args) {

        // 并发:多线程操作同一个资源类, 把资源类丢入线程
        Ticket2 ticket = new Ticket2();


        // @FunctionalInterface 函数式接口,jdk1.8  lambda表达式 (参数)->{ 代码 }
        new Thread(()->{for (int i = 1; i < 40 ; i++) ticket.sale();},"A").start();
        new Thread(()->{for (int i = 1; i < 40 ; i++) ticket.sale();},"B").start();
        new Thread(()->{for (int i = 1; i < 40 ; i++) ticket.sale();},"C").start();


    }
}

// Lock三部曲
// 1、 new ReentrantLock();
// 2、 lock.lock(); // 加锁
// 3、 finally=>  lock.unlock(); // 解锁
class Ticket2 {
    // 属性、方法
    private int number = 30;

    Lock lock = new ReentrantLock();

    public void sale(){

        lock.lock(); // 加锁

        try {
           // 业务代码

            if (number>0){
                System.out.println(Thread.currentThread().getName()+"卖出了"+(number--)+"票,剩余:"+number);
            }

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock(); // 解锁
        }
    }

}

两者的区别

  1. Synchronized 内置的Java关键字, Lock 是一个Java类
  2. Synchronized 无法判断获取锁的状态,Lock 可以判断是否获取到了锁
  3. Synchronized 会自动释放锁,lock 必须要手动释放锁!如果不释放锁,死锁
  4. Synchronized 线程 1(获得锁,阻塞)、线程2(等待,傻傻的等);Lock锁就不一定会等待下去;
  5. Synchronized 可重入锁,不可以中断的,非公平;Lock ,可重入锁,可以 判断锁,非公平(可以 自己设置);
  6. Synchronized 适合锁少量的代码同步问题,Lock 适合锁大量的同步代码

生产者消费者模式两种实现

await 和 notify的实现方式

public class B  {
    public static void main(String[] args) {
        Data2 data = new Data2();

        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                try {
                    data.increment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"A").start();

        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                try {
                    data.decrement();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"B").start();

        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                try {
                    data.increment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"C").start();

        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                try {
                    data.decrement();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"D").start();

    }
}

// 判断等待,业务,通知
class Data2{ // 数字 资源类

    private int number = 0;

    Lock lock = new ReentrantLock();
    Condition condition = lock.newCondition();

    //condition.await(); // 等待
    //condition.signalAll(); // 唤醒全部
    //+1
    public void increment() throws InterruptedException {
        lock.lock();
        try {
            // 业务代码
            while (number!=0){  //0
                // 等待
                condition.await();
            }
            number++;
            System.out.println(Thread.currentThread().getName()+"=>"+number);
            // 通知其他线程,我+1完毕了
            condition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }

    }

    //-1
    public synchronized void decrement() throws InterruptedException {
        lock.lock();
        try {
            while (number==0){ // 1
                // 等待
                condition.await();
            }
            number--;
            System.out.println(Thread.currentThread().getName()+"=>"+number);
            // 通知其他线程,我-1完毕了
            condition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

}

用JUC实现

public class C {

    public static void main(String[] args) {
        Data3 data = new Data3();

        new Thread(()->{
            for (int i = 0; i <10 ; i++) {
                data.printA();
            }
        },"A").start();

        new Thread(()->{
            for (int i = 0; i <10 ; i++) {
                data.printB();
            }
        },"B").start();

        new Thread(()->{
            for (int i = 0; i <10 ; i++) {
                data.printC();
            }
        },"C").start();
    }

}

class Data3{ // 资源类 Lock

    //几个线程就用几个Condition 精准的通知和唤醒线程
     private Lock lock = new ReentrantLock();
    private Condition condition1 = lock.newCondition();
    private Condition condition2 = lock.newCondition();
    private Condition condition3 = lock.newCondition();
    private int number = 1; // 1A  2B  3C

    public void printA(){
        lock.lock();
        try {
            // 业务,判断-> 执行-> 通知
            while (number!=1){
                // 等待
                condition1.await();
            }
            System.out.println(Thread.currentThread().getName()+"=>AAAAAAA");
            // 唤醒,唤醒指定的人,B
            number = 2;
            condition2.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    public void printB(){
        lock.lock();
        try {
            // 业务,判断-> 执行-> 通知
            while (number!=2){
                condition2.await();
            }
            System.out.println(Thread.currentThread().getName()+"=>BBBBBBBBB");
            // 唤醒,唤醒指定的人,c
            number = 3;
            condition3.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    public void printC(){
        lock.lock();
        try {
            // 业务,判断-> 执行-> 通知
            // 业务,判断-> 执行-> 通知
            while (number!=3){
                condition3.await();
            }
            System.out.println(Thread.currentThread().getName()+"=>BBBBBBBBB");
            // 唤醒,唤醒指定的人,c
            number = 1;
            condition1.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

}

使用JUC的集合

原生的List,Map,Set的实现类方法在多线程会出现不同步的问题,所以有以下解决的方案:

List

  1. List list = new Vector<>(); 其实Vector底层也就是在add方法上加了synchronized。 在这里插入图片描述
  2. List list = Collections.synchronizedList(new ArrayList<> ());

  3. List list = new CopyOnWriteArrayList<>(); (推荐使用这个,效率高):底层使用了可重入锁Lock,效率更高。 在这里插入图片描述

Set

  1. Set set = Collections.synchronizedSet(new HashSet<>()); 使用这个锁的就是整个对象。 在这里插入图片描述 在这里插入图片描述
  2. Set set = new CopyOnWriteArraySet<>();(推荐使用这个,效率高):底层使用了可重入锁Lock,效率更高。

Map

  1. Map map= Collections.synchronizedMap(new HashMap<>());
  2. Map map = new ConcurrentHashMap<>();(推荐使用这个~)

使用Callable

  • 执行相同的任务有存在缓存,效率高。
  • 结果可能需要等待,会阻塞!
public class CallableTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // new Thread(new Runnable()).start();
        // new Thread(new FutureTask<V>()).start();
        // new Thread(new FutureTask<V>( Callable )).start();
        new Thread().start(); // 怎么启动Callable

        MyThread thread = new MyThread();
        FutureTask futureTask = new FutureTask(thread); // 适配类

        new Thread(futureTask,"A").start();
        new Thread(futureTask,"B").start(); // 结果会被缓存,效率高

        Integer o = (Integer) futureTask.get(); //这个get 方法可能会产生阻塞!把他放到最后
        // 或者使用异步通信来处理!
        System.out.println(o);

    }
}

class MyThread implements Callable<Integer> {

    @Override
    public Integer call() {
        System.out.println("call()"); // 会打印几个call
        // 耗时的操作
        return 1024;
    }

}

GUC种的三大辅助类

CountDownLatch(减法计数器)

  • countDownLatch.countDown(); // 数量-1
  • countDownLatch.await(); // 等待计数器归零,然后再向下执行
  • 每次有线程调用 countDown() 数量-1,假设计数器变为0,countDownLatch.await()就会被唤醒,继续执行
// 计数器
public class CountDownLatchDemo {
    public static void main(String[] args) throws InterruptedException {
        // 总数是6,必须要执行任务的时候,再使用!
        CountDownLatch countDownLatch = new CountDownLatch(6);

        for (int i = 1; i <=6 ; i++) {
            new Thread(()->{
                System.out.println(Thread.currentThread().getName()+" Go out");
                countDownLatch.countDown(); // 数量-1
            },String.valueOf(i)).start();
        }

        countDownLatch.await(); // 等待计数器归零,然后再向下执行

        System.out.println("Close Door");

    }
}

CyclicBarrier (加法计数器)

只有等待线程数达到特定值才会结束。

public class CyclicBarrierDemo {
    public static void main(String[] args) {
        /**
         * 集齐7颗龙珠召唤神龙
         */
        // 召唤龙珠的线程
        CyclicBarrier cyclicBarrier = new CyclicBarrier(7,()->{
            System.out.println("召唤神龙成功!");
        });

        for (int i = 1; i <=7 ; i++) {
            final int temp = i;
            new Thread(()->{
                System.out.println(Thread.currentThread().getName()+"收集"+temp+"个龙珠");
                try {
                    cyclicBarrier.await(); // 等待
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

Semaphore(信号量)

规定有几个线程就最多只能有几个线程执行。 - semaphore.acquire(); 获得,假设如果已经满了,等待,等待被释放为止 - semaphore.release();释放,会将当前的信号量释放 + 1,然后唤醒等待的线程!

作用: 多个共享资源互斥的使用!并发限流,控制大的线程数!

public class SemaphoreDemo {
    public static void main(String[] args) {
        // 线程数量:停车位! 限流!
        Semaphore semaphore = new Semaphore(3);

        for (int i = 1; i <=6 ; i++) {
            new Thread(()->{
                // acquire() 得到
                try {
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName()+"抢到车位");
                    TimeUnit.SECONDS.sleep(2);
                    System.out.println(Thread.currentThread().getName()+"离开车位");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    semaphore.release(); // release() 释放
                }

            },String.valueOf(i)).start();
        }

    }
}

读写锁

ReadLock 读锁,多线程可同时占有执行。 writeLock() 写锁,多线程中只能一个线程占有执行。

/**
 * 独占锁(写锁) 一次只能被一个线程占有
 * 共享锁(读锁) 多个线程可以同时占有
 * ReadWriteLock
 * 读-读  可以共存!
 * 读-写  不能共存!
 * 写-写  不能共存!
 */
public class ReadWriteLockDemo {
    public static void main(String[] args) {
        MyCache myCache = new MyCache();

        // 写入
        for (int i = 1; i <= 5 ; i++) {
            final int temp = i;
            new Thread(()->{
                myCache.put(temp+"",temp+"");
            },String.valueOf(i)).start();
        }

        // 读取
        for (int i = 1; i <= 5 ; i++) {
            final int temp = i;
            new Thread(()->{
                myCache.get(temp+"");
            },String.valueOf(i)).start();
        }

    }
}

// 加锁的
class MyCacheLock{

    private volatile Map<String,Object> map = new HashMap<>();
    // 读写锁: 更加细粒度的控制
    private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
    private Lock lock = new ReentrantLock();

    // 存,写入的时候,只希望同时只有一个线程写
    public void put(String key,Object value){
        readWriteLock.writeLock().lock();
        try {
            System.out.println(Thread.currentThread().getName()+"写入"+key);
            map.put(key,value);
            System.out.println(Thread.currentThread().getName()+"写入OK");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            readWriteLock.writeLock().unlock();
        }
    }

    // 取,读,所有人都可以读!
    public void get(String key){
        readWriteLock.readLock().lock();
        try {
            System.out.println(Thread.currentThread().getName()+"读取"+key);
            Object o = map.get(key);
            System.out.println(Thread.currentThread().getName()+"读取OK");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            readWriteLock.readLock().unlock();
        }
    }

}

/**
 * 自定义缓存
 */
class MyCache{

    private volatile Map<String,Object> map = new HashMap<>();

    // 存,写
    public void put(String key,Object value){
        System.out.println(Thread.currentThread().getName()+"写入"+key);
        map.put(key,value);
        System.out.println(Thread.currentThread().getName()+"写入OK");
    }

    // 取,读
    public void get(String key){
        System.out.println(Thread.currentThread().getName()+"读取"+key);
        Object o = map.get(key);
        System.out.println(Thread.currentThread().getName()+"读取OK");
    }
}

阻塞队列

  1. 抛出异常,add==>remove
    public static void test1(){
            // 队列的大小
            ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);
    
            System.out.println(blockingQueue.add("a"));
            System.out.println(blockingQueue.add("b"));
            System.out.println(blockingQueue.add("c"));
            // IllegalStateException: Queue full 抛出异常!
            // System.out.println(blockingQueue.add("d"));
    
            System.out.println("=-===========");
    
            System.out.println(blockingQueue.element()); // 查看队首元素是谁
            System.out.println(blockingQueue.remove());
    
    
            System.out.println(blockingQueue.remove());
            System.out.println(blockingQueue.remove());
    
            // java.util.NoSuchElementException 抛出异常!
            // System.out.println(blockingQueue.remove());
        }
  2. 有返回值,无抛出异常 offer==>poll peel(获取队首元素)

 public static void test2(){
        // 队列的大小
        ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);

        System.out.println(blockingQueue.offer("a"));
        System.out.println(blockingQueue.offer("b"));
        System.out.println(blockingQueue.offer("c"));

        System.out.println(blockingQueue.peek());
        // System.out.println(blockingQueue.offer("d")); // false 不抛出异常!
        System.out.println("============================");
        System.out.println(blockingQueue.poll());
        System.out.println(blockingQueue.poll());
        System.out.println(blockingQueue.poll());
        System.out.println(blockingQueue.poll()); // null  不抛出异常!
    }
3. 死死的阻塞等待 put==>take

public static void test3() throws InterruptedException {
        // 队列的大小
        ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);

        // 一直阻塞
        blockingQueue.put("a");
        blockingQueue.put("b");
        blockingQueue.put("c");
        // blockingQueue.put("d"); // 队列没有位置了,一直阻塞
        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take()); // 没有这个元素,一直阻塞

    }
  1. 等待,阻塞(设置等待超时)依旧是offer==>poll
 public static void test4() throws InterruptedException {
        // 队列的大小
        ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);

        blockingQueue.offer("a");
        blockingQueue.offer("b");
        blockingQueue.offer("c");
        // blockingQueue.offer("d",2,TimeUnit.SECONDS); // 等待超过2秒就退出
        System.out.println("===============");
        System.out.println(blockingQueue.poll());
        System.out.println(blockingQueue.poll());
        System.out.println(blockingQueue.poll());
        blockingQueue.poll(2,TimeUnit.SECONDS); // 等待超过2秒就退出

    }

同步队列

  • 和其他的BlockingQueue 不一样, SynchronousQueue 不存储元素
  • put了一个元素,必须从里面先take取出来,否则不能在put进去值!意思就是队列只能存一个值,取一个值。
public static void main(String[] args) {
        BlockingQueue<String> blockingQueue = new SynchronousQueue<>(); // 同步队列

        new Thread(()->{
            try {
                System.out.println(Thread.currentThread().getName()+" put 1");
                blockingQueue.put("1");
                System.out.println(Thread.currentThread().getName()+" put 2");
                blockingQueue.put("2");
                System.out.println(Thread.currentThread().getName()+" put 3");
                blockingQueue.put("3");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"T1").start();


        new Thread(()->{
            try {
                TimeUnit.SECONDS.sleep(3);
                System.out.println(Thread.currentThread().getName()+"=>"+blockingQueue.take());
                TimeUnit.SECONDS.sleep(3);
                System.out.println(Thread.currentThread().getName()+"=>"+blockingQueue.take());
                TimeUnit.SECONDS.sleep(3);
                System.out.println(Thread.currentThread().getName()+"=>"+blockingQueue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"T2").start();
    }

线程池

线程池的好处: 1. 降低资源的消耗 2. 提高响应的速度 3. 方便管理。

线程复用、可以控制最大并发数、管理线程

三大创建方法

  • ExecutorService threadPool = Executors.newSingleThreadExecutor();// 单个线 程
  • ExecutorService threadPool = Executors.newFixedThreadPool(10);// 创建一 个固定的线程池的大小
  • ExecutorService threadPool = Executors.newCachedThreadPool(); // 由执行的线程数决定大小。

具体过程:创建线程池对象--->执行---->关闭线程池

 try {
        for (int i = 1; i <= 10; i++) {
                // 使用了线程池之后,使用线程池来创建线程
                threadPool.execute(()->{
                    System.out.println(Thread.currentThread().getName());
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 线程池用完,程序结束,关闭线程池
            threadPool.shutdown();
        }

七大参数

一般开发中使用线程池最好不使用Executors去创建对象。如阿里巴巴开发手册所说的:OOM(内存溢出) 在这里插入图片描述 所以应该使用ThreadPoolExecutor来创建线程池 查看源码得到以下: - 当执行线程数大于核心线程锁时,其它休眠的线程位置将会启用,最多不超过最大线程数。 - 当超过最大线程锁的线程任务会在阻塞队列中进行等待。 - 当此时并发执行的线程任务超过了最大线程数+阻塞队列长度时,拒绝策略起效。

拒绝策略有四种

  • new ThreadPoolExecutor.AbortPolicy() // 同时并发的线程数大于最大线程数+阻塞队列长度时,不处理超过部分的线程任务,直接抛出异常。
  • new ThreadPoolExecutor.DiscardPolicy() //同时并发的线程数大于最大线程数+阻塞队列长度时,直接丢掉超过部分的线程任务,不抛出异常。
  • new ThreadPoolExecutor.CallerRunsPolicy() // 同时并发的线程数大于最大线程数+阻塞队列长度时,超过部分的线程任务回归main主线程执行。
  • new ThreadPoolExecutor.DiscardOldestPolicy() //阻塞队列满了,尝试去和最先执行的线程发起竞争,也不会抛出异常!
public ThreadPoolExecutor(int corePoolSize, //核心线程数
                              int maximumPoolSize, //最大线程数。最多几个线程并发。
                              long keepAliveTime, //当线程无任务时,几秒后结束该线程
                              TimeUnit unit,//线程结束的时间单位
                              BlockingQueue<Runnable> workQueue,//阻塞队列,限制等候线程数
                              ThreadFactory threadFactory,//线程工厂
                              RejectedExecutionHandler handler)//拒绝策略

例子如下:

ExecutorService threadPool = new ThreadPoolExecutor(
                2,
                Runtime.getRuntime().availableProcessors(),
                3,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(3),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.DiscardOldestPolicy());  //队列满了,尝试去和最早的竞争,也不会抛出异常!

如何选择线程池的大小

  1. IO 密集型 :判断你程序中十分耗IO的线程,有n个,线程池就设置(2n)个

  2. CPU 密集型:一般电脑几核,线程池就设置几个线程,可以保持cpu的效率最高!可以利用Runtime.getRuntime().availableProcessors()获取当前主机的核数。

评论