标签：JUC 周阳版编程 System 线程 println new public out

Java8特性

四大函数式接口 supplier, consumer, function , predicate + 流式计算 stream (map,filter,reduce) + lambda + 方法引用

1. 区分并发\并行线程\进程

2. Lambda 表达式的使用 - 线程操作资源类

之前的实现是：自己的多线程类，继承Thread或者实现Runable接口，重写run方法

企业级开发的思路是 高内聚低耦合

现在的实现应该是：线程操作资源类

资源类中有对应的功能逻辑

线程通过Lambda创建，操作资源类中的内容

public class SaleTicket {
    public static void main(String[] args) {
        Ticket ticket = new Ticket();
        for (int i = 0; i < 3; i++) {
            new Thread(() -> {
                for (int j = 0; j < 30; j++) {
                    ticket.saleTicket();
                }
            }, String.valueOf(i)).start();
        }
    }
}
class Ticket {
    private int number = 30;

    public void saleTicket() {
        if (number > 0) {
            System.out.println(Thread.currentThread().getName() + "\t卖出第:" + number-- + "\t还剩下:" + number);
        }
    }
}

3. ReentrantLock - 精准通知，精确访问

例子：使用 ReentrantLock 循环打印 A,B,C

思路：ReentrantLock 可以获取多个 Condition,

由 number 标志位来判断当前打印内容

A,B,C 分别为不同 Condition

A完成打印后, 转到 B的Condition, number此时由 1->2

B完成打印后, 转到 C的Condition, number此时由 2->3

C完成打印后, 转到 A的Condition, number此时由 3->1

构成循环打印 A B C

public class xunhuanABC {
    public static void main(String[] args) {
        biaozhi b = new biaozhi();
        new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                b.printA();
            }
        }, "A").start();
        new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                b.printB();
            }
        }, "B").start();
        new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                b.printC();
            }
        }, "C").start();
    }
}

class biaozhi {
    private int number = 1;
    Lock lock = new ReentrantLock();
    Condition condition1 = lock.newCondition();
    Condition condition2 = lock.newCondition();
    Condition condition3 = lock.newCondition();
    public void printA() {
        lock.lock();
        try {
            while (number != 1) {
                condition1.await();
            }
            System.out.println("A");
            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("B");
            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("C");
            number = 1;
            condition1.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
}

4. Lock锁的范围 - 对象? 类?

对象锁可以有多把，但是类锁只有一把

5. 集合安全

ArrayList

ArrayList 不安全，并发读写会造成 ConcurrentModificationException

解决方案1 使用 Vector ，本质上在add方法中加了synchronized，并发量低，效率低

解决方案2 使用 Collections.synchronizedList 方法，将线程不安全的List转化为线程安全的List，小数据量可用

解决方案3 使用 CopyOnWriteArrayList 写时复制，适合读多写少

CopyOnWriteArrayList - 安全

写操作时,不是对本身object数组添加, 而是复制到新的object容器 Object[] newElements,长度为原数组长度+1,再把新元素添加到末尾.

添加完成后, 将原容器的引用指向新的容器

读写分离,可以并发读

public boolean add(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            Object[] elements = getArray();
            int len = elements.length;
            Object[] newElements = Arrays.copyOf(elements, len + 1);
            newElements[len] = e;
            setArray(newElements);
            return true;
        } finally {
            lock.unlock();
        }
    }

HashSet

内部实现是一个HashMap, 添加一个元素e, 就把e 作为key 存入, 那 value 用什么? 一个空的 Object对象 PRESENT常量

private static final Object PRESENT = new Object();
public boolean add(E e) {
        return map.put(e, PRESENT)==null;
}

HashSet 不安全, ConcurrentModificationException

解决方案1 使用 Collections.synchronizedSet 方法，将线程不安全的Set转化为线程安全的Set，小数据量可用

解决方案2 使用 CopyOnWriteArraySet 写时复制，适合读多写少, 原理同上

HashMap

HashMap内部存放的不是 K , V ,而是存了 K , V 的Node节点数组

static class Node<K,V> implements Map.Entry<K,V> {
        final int hash; 
        final K key;
        V value;
        Node<K,V> next;

        Node(int hash, K key, V value, Node<K,V> next) {
            this.hash = hash;
            this.key = key;
            this.value = value;
            this.next = next;
        }   
    }

更具体的看另外一篇 HashMap 深入

CopyOnWriteArraySet - 安全

内部实现其实是一个CopyOnWriteArrayList

6. 第三种获得多线程的方式 - Callable 接口

如何使用 Callable 来 new 一个线程 Thread?

Callable 接口对象, 传入FutureTask, 作为构造参数

而FutureTask 本身实现了Runable 接口, 可以被传给Thread

面向接口编程

public class CallableDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        FutureTask futureTask = new FutureTask(new MyThread());
        new Thread(futureTask,"A").start();
        System.out.println(futureTask.get());
    }
}
class MyThread implements Callable<Integer>{
    @Override
    public Integer call() throws Exception {
        System.out.println(".....come in here");
        return 1024;
    }
}

7. JUC辅助类

CountDownLatch

构造时, 传入 count 数字,
使用 .countdown() 方法,将 count 数字 -1
当 count 归零
被 .await() 阻塞的语句将会执行

public class CountDownLatchDemo {
    public static void main(String[] args) throws InterruptedException {
        closeDoor();
    }
    public static void closeDoor() throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(6);
        for (int i = 1; i <=6; i++) {
            new Thread(()->{
                System.out.println(Thread.currentThread().getName()+"\t离开教室");
                countDownLatch.countDown();
            },String.valueOf(i)).start();
        }
        System.out.println("主线程阻塞等待");
        countDownLatch.await();
        System.out.println(Thread.currentThread().getName()+"\t关门走人");
    }
}

CyclicBarrier

类似 CountDownLatch

区别在于前者是减少

而这个 CyclicBarrier 是增加 ,逐步靠近最后的终点

在构造函数里, 要设置parties 数量,和一个Runable 接口的命令

通过 cyclicBa 到 0,

public class CyclicBarrierDemo {
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(7,()->{System.out.println("召唤神龙");});
        for (int i = 1; i <=7; i++) {
            final int tempInt=i;
            new Thread(()->{
                System.out.println(Thread.currentThread().getName()+"\t收集到的第:"+tempInt+"颗龙珠");
                try {
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                } catch (BrokenBarrierException e) {
                    throw new RuntimeException(e);
                }
            },String.valueOf(i)).start();
        }
    }
}

Semaphore

在构造时, 传入一个数字,用来表示标志量

semaphore.acquire() 获得一个标志量

semaphore.release() 释放一个标志量

获取和释放通常在 try 里获得在 finally 里释放

public class SemaphoreDemo {
    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(3);
        for (int i = 1; i <= 6; i++) {
            new Thread(()->{
                try {
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName()+"\t抢占到了车位");
                    try {
                        TimeUnit.SECONDS.sleep(3);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    System.out.println(Thread.currentThread().getName()+"\t正在离开车位");
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }finally {
                    semaphore.release();
                    System.out.println(Thread.currentThread().getName()+"\t离开车位成功");
                }
            },String.valueOf(i)).start();
        }
    }
}

8. ReentrantReadWriteLock - 写加锁，读不加锁

思路写不可以同时, 但是读可以同时

读操作共享，写操作独占

通过实现了 ReadWriteLock接口的 ReentrantReadWriteLock类获得锁对象

通过锁对象获得读锁或者写锁，再来使用

class MyCache{
    private volatile Map<String,Object> map=new HashMap<>();
    private ReadWriteLock readWriteLock=new ReentrantReadWriteLock();
    public void put(String key,Object value) throws InterruptedException {
        Lock lock = readWriteLock.writeLock();
        lock.lock();
        try {
            System.out.println(Thread.currentThread().getName()+"\t 正在写入 "+key);
            TimeUnit.MICROSECONDS.sleep(300);
            map.put(key, value);
            System.out.println(Thread.currentThread().getName()+"\t 写入成功");
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }
    public void get(String key){
        readWriteLock.readLock().lock();
        try {
            System.out.println(Thread.currentThread().getName()+"\t 正在读取"+key);
            Object result = map.get(key);
            System.out.println(Thread.currentThread().getName()+"\t 读取成功 "+key+"值为 "+result);
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            readWriteLock.readLock().unlock();
        }
    }
}

9. 阻塞队列 - 海底捞等位区

阻塞队列是是一种队列，某个线程往阻塞队列添加元素，另外一个线程往队列里移走元素

当队列是空的，从队列移走元素操作会被阻塞

当队列是满的，向队列添加元素操作会被阻塞

生产者消费者模式

但是手动写这个互相阻塞的逻辑很麻烦，所有我们使用阻塞队列，阻塞队列会出手

阻塞队列BlockingQueue是一个接口

实现类有：ArrayBlockingQueue 数组结构组成的有界阻塞队列

LinkedBlockingQueue 链表结构组成的有界（默认大小是integer.MAX_VALUE）阻塞队列

LinkedBlockingDeque 链表结构组成的双向阻塞队列

方法类型	抛出异常	特殊值	阻塞	超时
插入	add(e)	offer(e)	put(e)	offer(e,time,unit)
移除	remove()	poll()	take()	poll(time,unit)
检查	element()	peek()	不可用	不可用

情况	解释
抛出异常	会抛异常，队列满还加，IllegalStateException.Queue full; 队列空还取，NoSuchElementException
特殊值	不抛异常，插入成功true，失败false；移除成功返回元素，失败返回null
阻塞	队列满还加，队列会阻塞直到添加成功或者响应中断退出；队列空还取，队列会阻塞到有东西取
超时	阻塞一段时间，超时主动退出(过时不候)

9.5 值传递和引用传递

int a=30;
String b="xxx";
Person c=new Person();

a 传入方法都是值传递

b 传入方法是引用传递, 但是由于String的特殊性, 不可变, 就在局内范围内重新指向了一个新的,

c 传入方法是引用传递

10. 线程池 - 池化技术

池化技术, 数据库连接池 , 德鲁伊

优点 : 避免频繁创建-销毁 , 浪费资源 - 线程复用

控制最大并发数

帮助创建销毁 - 管理线程

//一池5个工作线程,类似一个银行有5个受理窗口
ExecutorService threadPool = Executors.newFixedThreadPool(5);
//一池1个工作线程,类似一个银行有1个受理窗口
ExecutorService threadPool = Executors.newSingleThreadExecutor();
//一池N个工作线程,类似一个银行有N个受理窗口(高峰需要几个就创建几个线程,容易引发OOM问题)
ExecutorService threadPool = Executors.newCachedThreadPool();

以上三个本质都是使用ThreadPoolExecutor

ThreadPoolExecutor 七大参数

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;
}

corePoolSize

线程池中常驻核心线程数

maximumPoolSize

线程池中能够容纳同时执行的最大线程数, 此值必须大于等于1

keepAliveTime

多余的空闲线程的存活时间:当前池中线程数量超过corePoolSize时, 当空闲时间达到keepAliveTime 时, 多余线程会被销毁直到剩下corePoolSize 个线程为止

unit

keepAliveTime 单位

BlockingQueue workQueue

任务队列, 提交但未被执行的程序

ThreadFactory threadFactory

表示生成线程池中工作线程的线程工厂, 用于创建线程, 一般默认 Executors.defaultThreadFactory()

RejectedExecutionHandler handler(四大拒绝策略)

拒绝策略, 队列满时,并且工作现场大于等于最大线程数 , 一般默认 defaultHandler = new AbortPolicy()

共有四个拒绝策略的实现类 :默认的是报错; 直接在 execute方法的调用线程中运行被拒绝的任务; 丢弃最旧任务; 静默丢弃

线程池底层工作原理

关键点: 常驻核心线程处理, 阻塞队列满后, 再增加临时线程处理, 直到到达最大线程数

实际怎么用 - 另辟蹊径

实际上, 阿里巴巴开发手册要求不允许使用Executors去创建线程池, 而是通过 ThreadPoolExecutor 手动创建

原因在于可以更加明确线程池的运行规则, 避免资源耗尽的风险

1.Executors 中的 FixedThreadPool和SingleThreadPool,允许的请求队列长度为Integer.MAX_VALUE, 会堆积大量请求造成OOM

2.Executors中的CachedThreadPool和ScheduledThreadPool,允许的创建线程数量为Integer.MAX_VALUE,会创建大量请求造成OOM

11. Stream流式计算

区分Stream文件流

四大函数式接口 supplier, consumer, function , predicate

位于 java.util包下

函数式接口	参数类型	返回类型	用途
Consumer 消费型接口	T	void	对类型为 T 的对象应用操作, 包含方法: void accept(T t)
Supplier 供给型接口	无	T	返回类型为 T 的对象, 包含方法: T get()
Function<T,R> 函数型接口	T	R	对类型为 T 的对象应用操作, 并返回结构. 结果是R类型的对象. 包含方法 R apply(T t)
Predicate断定型接口	T	boolean	确定类型为T的对象是否满足某约束, 并返回boolean值. 包含方法 boolean test(T t)

可以结合 Lambda 表达式

public class StreamDemo {
    public static void main(String[] args) {
        Function<String,Integer> function=new Function<String, Integer>() {
            @Override
            public Integer apply(String s) {
                return 1023;
            }
        };
        System.out.println(function.apply("abc"));
        Function<String,Integer> function1=s -> {
            return s.length();
        };
        System.out.println(function1.apply("sadsadfas"));
    }
}

流式计算 stream

Stream 特性

本身不存储结果

不会改变原有对象,而是返回一个持有新结果的新Stream

操作延迟执行,到需要结果时才开始计算

filter需要传入一个Predicate接口, 里面写的是过滤条件, 里面可以写自己的条件, 使用Lambda表达式

map需要传入一个Function接口, 里面写的是映射条件, 返回生成的新Stream

reduce需要传入一个Supplier接口

12. ForkJoinPool

ForkJoinPool类实现了 ExecutorService 接口

分治思想

RecursiveTask 继承自 ForkJoinTask

资源类继承自 RecursiveTask

在主线程中, 使用 ForkJoinPool 的对象来 .submit 资源类

使用 .get() 获取结果

使用 .shutdow() 关闭 ForkJoinPool

public class ForkJoinDemo {
    public static void main(String[] args) {
        MyTask myTask=new MyTask(0,100);
        ForkJoinPool threadPool=new ForkJoinPool();
        ForkJoinTask<Integer> forkJoinTask = threadPool.submit(myTask);
        System.out.println(forkJoinTask.get());
        threadPool.shutdown();
    }
}
class MyTask extends RecursiveTask<Integer>{
    public static final Integer ADJUST_VALUE=10;
    private int begin;
    private  int end;
    private int result;
    public MyTask(int begin, int end) {
        this.begin = begin;
        this.end = end;
    }
    @Override
    protected Integer compute() {
        if ((end-begin)<=ADJUST_VALUE){
            for (int i = begin; i <=end; i++) {
                result=result+i;
            }
        }else{
            int middle=(end+begin)/2;
            MyTask task1=new MyTask(begin,middle);
            MyTask task2=new MyTask(middle+1,end);
            task1.fork();
            task2.fork();
            result=task1.join()+task2.join();
        }
        return result;
    }
}

区别于手动分治算法, ForkJoinPool关键在于本质是使用不同的线程进行分治

资源类中使用 fork() 进行分治, 使用 join() 合并, 最后返回结果

13. 异步回调

CompletableFutureDemo

public class CompletableFutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        /**
         * 异步调用
         * 没有返回值
         *
         */
        CompletableFuture<Void> completableFuture=CompletableFuture.runAsync(()->{
            System.out.println(Thread.currentThread().getName()+"没有返回,update mysql ok");
        });
        completableFuture.get();

        /**
         * 异步调用
         * 有返回值
         * 内含 供给型接口
         * whenComplete 需要两个参数传入 BiConsumer
         * 正常返回t 1024
         * 异常返回u 4444
         */
        CompletableFuture<Integer> completableFuture2=CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread().getName()+"没有返回,update mysql ok");
            int age=10/0;
            return 1024;
        });
        System.out.println(completableFuture2.whenComplete((t, u) -> {
            System.out.println("******t:" + t);
            System.out.println("******u:" + u);
        }).exceptionally(f -> {
            System.out.println("******exception:" + f.getMessage());
            return 4444;
        }).get());
    }
}

标签：JUC,周阳版,编程,System,线程,println,new,public,out
From： https://www.cnblogs.com/yuhaozhe/p/17051928.html

JUC_并发编程(周阳版)

Java8特性

1. 区分并发\并行线程\进程

2. Lambda 表达式的使用 - 线程操作资源类

3. ReentrantLock - 精准通知，精确访问

4. Lock锁的范围 - 对象? 类?

5. 集合安全

ArrayList

CopyOnWriteArrayList - 安全

HashSet

HashMap

CopyOnWriteArraySet - 安全

6. 第三种获得多线程的方式 - Callable 接口

7. JUC辅助类

CountDownLatch

CyclicBarrier

Semaphore

8. ReentrantReadWriteLock - 写加锁，读不加锁

9. 阻塞队列 - 海底捞等位区

9.5 值传递和引用传递

10. 线程池 - 池化技术

ThreadPoolExecutor 七大参数

corePoolSize

maximumPoolSize

keepAliveTime

unit

BlockingQueue workQueue

ThreadFactory threadFactory

RejectedExecutionHandler handler(四大拒绝策略)

线程池底层工作原理

实际怎么用 - 另辟蹊径

11. Stream流式计算

四大函数式接口 supplier, consumer, function , predicate

流式计算 stream

12. ForkJoinPool

13. 异步回调

相关文章

赞助商

阅读排行

JUC_并发编程(周阳版)

Java8特性

1. 区分 并发\并行 线程\进程

2. Lambda 表达式的使用 - 线程操作资源类

3. ReentrantLock - 精准通知，精确访问

4. Lock锁的范围 - 对象? 类?

5. 集合安全

ArrayList

CopyOnWriteArrayList - 安全

HashSet

HashMap

CopyOnWriteArraySet - 安全

6. 第三种获得多线程的方式 - Callable 接口

7. JUC辅助类

CountDownLatch

CyclicBarrier

Semaphore

8. ReentrantReadWriteLock - 写加锁，读不加锁

9. 阻塞队列 - 海底捞等位区

9.5 值传递和引用传递

10. 线程池 - 池化技术

ThreadPoolExecutor 七大参数

corePoolSize

maximumPoolSize

keepAliveTime

unit

BlockingQueue workQueue

ThreadFactory threadFactory

RejectedExecutionHandler handler(四大拒绝策略)

线程池底层工作原理

实际怎么用 - 另辟蹊径

11. Stream流式计算

四大函数式接口 supplier, consumer, function , predicate

流式计算 stream

12. ForkJoinPool

13. 异步回调

相关文章

赞助商

阅读排行

1. 区分并发\并行线程\进程