UC并发编程学习笔记（简单易懂）4-阿里云开发者社区

9：四大函数式接口

9.1 Function 函数式接口

注意：传入一个T类型的参数R 返回一个参数R

/*
 * 函数式接口:有一个输入参数,有一个输出参数
 * */
public class function {
   public static void main(String[] args) {
      Function<String, String> function = new Function<String, String>() {
         @Override
         public String apply(String o) {
            return o;
         }
      };
      Function<String, String> function2 = (str)->{return str;};
      System.out.println(function2.apply("adns"));
   }
}

9.2 Predicate 断定型接口

注意：参入一个参数，返回一个布尔类型的 true / false

package fourinterface;
import java.util.function.Predicate;
/*
* 断定型接口:有一个输入参数,返回值只能是 布尔值
* */
public class PredicateDemo {
    public static void main(String[] args) {
        //判断字符串是否为空
//        Predicate<String> predicate = new Predicate<String>() {
//            @Override
//            public boolean test(String s) {
//                return s.isEmpty();
//            }
//        };
        Predicate<String> predicate = (str)->{return  str.isEmpty();};
        System.out.println(predicate.test(" "));
    }
}

9.3 Consumer 消费性接口

package fourinterface;
import java.util.function.Consumer;
/*
* Consumer 消费型接口: 只有输入,没有返回值
* */
public class ConsumerDemo {
    public static void main(String[] args) {
//        Consumer<String> consumer = new Consumer<String>() {
//            @Override
//            public void accept(String s) {
//                System.out.println(s);
//            }
//        };
        Consumer<String> consumer = (str)->{
            System.out.println(str);
        };
        consumer.accept("hello");
    }
}

9.4 Supplier 供给型接口

package fourinterface;
import java.util.function.Supplier;
/*
* Supplier 供给型接口 :没有参数 , 只有返回值
* */
public class SupplierDemo {
    public static void main(String[] args) {
//        Supplier supplier =new Supplier() {
//            @Override
//            public Object get() {
//                return 1024;
//            }
//        };
        Supplier supplier = ()->{return 1024;};
        System.out.println(supplier.get());
    }

10 Stream流式计算

方法摘要
Stream< T > filter(Predicate< ? super T > predicate)	返回由与此给定谓词匹配的此流的元素组成的流

Stream< T > sorted(Comparator< ? super T > comparator)

返回由该流的元素组成的流，根据提供的 Comparator进行排序

< R > Stream< R > map(Function< ? super T,? extends R > mapper)

返回由给定函数应用于此流的元素的结果组成的流

void forEach(Consumer< ? super T > action)	对此流的每个元素执行操作
Stream< T > limit( long maxSize )	返回由此流的元素组成的流，截短长度不能超过 maxSize

package fourinterface;
import java.util.Arrays;
import java.util.List;
public class StreamDemo {
    public static void main(String[] args) {
        User u1 = new User(1,"a",21);
        User u2 = new User(2,"b",22);
        User u3 = new User(3,"c",23);
        User u4 = new User(4,"d",24);
        User u5 = new User(6,"e",25);
        List<User> list = Arrays.asList(u1,u2,u3,u4,u5);
        //计算交给Stream流
        //lambda表达式,链式编程,函数式接口,Stream流式计算
        list.stream()
                .filter(u->{return u.getId()%2==0;})
                .filter(u->{return u.getAge()>23;})
                .map(u->{return  u.getName().toUpperCase();})
                .sorted((uu1,uu2)->{return uu2.compareTo(uu1);})
                .limit(1)
                .forEach(System.out::println);
    }
}
@Data
@NoArgsConstructor
@AllArgsConstructor
public class User {
   private int id ;
   private String name;
   private int age ;
}

11 ForkJoin（分支合并）

特点：工作窃取

ForkJoinTask

方法摘要
ForkJoinTask < V > fork()	在当前任务正在运行的池中异步执行此任务（如果适用）
V get()	等待计算完成，然后检索其结果

ForkJoinPool

方法摘要
< T > ForkJoinTask< T > submit(ForkJoinTask< T > task)	提交一个ForkJoinTask来执行

LongStream

方法摘要
< T > ForkJoinTask< T > submit(ForkJoinTask< T > task)	提交一个ForkJoinTask来执行

LongStream parallel()

返回平行的等效流

static LongStream rangeClosed(long startInclusive, long endInclusive)

返回有序顺序 LongStream从 startInclusive （含）至 endInclusive通过的递增步长（含） 1

long reduce(long identity, LongBinaryOperator op)

使用提供的身份值和 associative累积功能对此流的元素执行 reduction ，并返回减小的值

< T > ForkJoinTask< T > submit(ForkJoinTask< T > task)  提交一个ForkJoinTask来执行
LongStream parallel() 返回平行的等效流
static LongStream rangeClosed(long startInclusive, long endInclusive) 返回有序顺序 LongStream从 startInclusive （含）至 endInclusive通过的递增步长（含） 1
long reduce(long identity, LongBinaryOperator op) 使用提供的身份值和 associative累积功能对此流的元素执行 reduction ，并返回减小的值
package forkjoin;
import java.util.concurrent.RecursiveTask;
public class ForkJoinDemo extends RecursiveTask<Long> {
    private Long start;
    private Long end;
    private Long temp = 10000L;
    public ForkJoinDemo(Long start, Long end) {
        this.start = start;
        this.end = end;
    }
    @Override
    protected Long compute() {
        if ((end-start) < temp ){
            Long sum = 0L;
            for (Long  i = start; i <= end; i++) {
                sum += i;
            }
            return sum;
        }else{
            long middle = (start + end)/2;//中间值
            ForkJoinDemo task1 = new ForkJoinDemo(start,middle);
            task1.fork();//拆分任务,把任务压入线程队列
            ForkJoinDemo task2 = new ForkJoinDemo(middle+1,end);
            task2.fork();//拆分任务,把任务压入线程队列
            return task1.join()+task2.join();
        }
    }
}

package forkjoin;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.stream.LongStream;
public class Test {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //test1();//5533
        //test2();//3709
        test3();//222
    }
    public static void test1(){
        Long sum = 0L;
        long start = System.currentTimeMillis();
        for (Long i = 1L; i <= 10_0000_0000L; i++) {
            sum += i;
        }
        long end = System.currentTimeMillis();
        System.out.println("sum = "+sum+" 时间 : "+(end-start));
    }
    public static void test2() throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        ForkJoinTask<Long> task = new ForkJoinDemo(0L,10_0000_0000L);
        ForkJoinTask<Long> submit = forkJoinPool.submit(task);
        Long sum = submit.get();
        long end = System.currentTimeMillis();
        System.out.println("sum = "+sum+" 时间 : "+(end-start));
    }
    public static void test3(){
        long start = System.currentTimeMillis();
        long sum = LongStream.rangeClosed(0L,10_0000_0000L).parallel().reduce(0,Long::sum);
        long end = System.currentTimeMillis();
        System.out.println("sum = "+sum+" 时间 : "+(end-start));
    }
}

12 异步回调

package main;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
public class Demo2 {
    public static void main(String[] args) throws Exception {
        //没有返回值的runAsync 异步回调
//        CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(()->{
//            try {
//                TimeUnit.SECONDS.sleep(2);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            System.out.println(Thread.currentThread().getName()+"runAsync-->Void");
//        });
//        System.out.println("1111");
//        completableFuture.get();//获得阻塞执行结果
        //有返回值的 supplyAsync 异步回调
        //分为成功和失败的回调
        //失败返回的是错误信息
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread().getName()+"supplyAsync-->Integer");
            int i = 10/0;
            return 1024;
        });
        System.out.println(completableFuture.whenComplete((t,u)->{
            System.out.println("t-->"+t);//正常的返回结果
            System.out.println("u-->"+u);//错误信息
        }).exceptionally((e)->{
            System.out.println(e.getMessage());//可获取到错误的返回结果
            return 2233;
        }).get());
    }
}

13 Volatile(可见性)

package main;
import java.util.concurrent.TimeUnit;
public class VolatileDemo {
    private static volatile int num = 0;
    public static void main(String[] args) throws Exception {
        new Thread(()->{
            while (num == 0) {//num不加volatile无法跳出循环
            }
        }).start();
        TimeUnit.SECONDS.sleep(1);
        num=1;
        System.out.println(num);
    }
}

不保证原子性

package main;
import java.io.PrintWriter;
//volatile不保证原子性
public class VolatileDemo2 {
    private volatile static int num = 0;
    public static void add(){
        num++;//不是原子性操作,底层分好几步
    }
    public static void main(String[] args) {
        //理论上num结果为2万
        for (int i = 0; i < 20; i++) {
            new Thread(()->{
                for (int j = 0; j < 1000; j++) {
                    add();
                }
            }).start();
        }
        while (Thread.activeCount() > 2) {//main gc
            Thread.yield();
        }
        System.out.println(Thread.currentThread().getName()+" "+num);
    }
}

不加锁保证原子性

package main;
import java.util.concurrent.atomic.AtomicInteger;
//volatile不保证原子性
public class VolatileDemo2 {
    //原子类的Integer
    private volatile static AtomicInteger num = new AtomicInteger();
    public static void add(){
        //num++;//不是原子性操作,底层分好几步
        num.getAndIncrement();//AtomicInteger + 1 方法 ,CAS
    }
    public static void main(String[] args) {
        //理论上num结果为2万
        for (int i = 0; i < 20; i++) {
            new Thread(()->{
                for (int j = 0; j < 1000; j++) {
                    add();
                }
            }).start();
        }
        while (Thread.activeCount() > 2) {
            Thread.yield();
        }
        System.out.println(Thread.currentThread().getName()+" "+num);
    }
}

14 单例模式

饿汉式单例

package single;
//饿汉式单例
public class Hungry {
    //可能会浪费空间
    private byte[] data1 = new byte[1024*1024];
    private byte[] data2 = new byte[1024*1024];
    private byte[] data3 = new byte[1024*1024];
    private Hungry(){
    }
    private final static Hungry HUNGRY = new Hungry();
    public static Hungry getInstance(){
        return HUNGRY;
    }
}

懒汉式单例

package single;
import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
//懒汉式单例
//道高一尺魔高一丈
public class LazyMan {
    private static boolean code = false;
    private LazyMan(){
        synchronized (LazyMan.class){
            if (code == false){
                code = true;
            }else{
                throw  new RuntimeException("不要试图使用反射破坏异常");
            }
        }
    }
    private volatile static LazyMan lazyMan;
    //双重检测锁模式的 懒汉式单例 DCL懒汉式
    public static LazyMan getInstance(){
        if (lazyMan == null) {
            synchronized (LazyMan.class){
                if (lazyMan == null) {
                    lazyMan = new  LazyMan();//不是一个原子操作
                }
            }
        }
        return lazyMan;
    };
    //多线程并发
    public static void main(String[] args) throws NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException, NoSuchFieldException {
//        for (int i = 0; i < 10; i++) {
//            new Thread(()->{
//                LazyMan.getInstance();
//            }).start();
//        }
        //LazyMan instance  = LazyMan.getInstance();
        Field code = LazyMan.class.getDeclaredField("code");
        code.setAccessible(true);
        Constructor<LazyMan> declaredConstructor = LazyMan.class.getDeclaredConstructor(null);
        declaredConstructor.setAccessible(true);
        LazyMan instance2 = declaredConstructor.newInstance();
        code.set(instance2,false);
        LazyMan instance3 = declaredConstructor.newInstance();
        System.out.println(instance2.hashCode());
        System.out.println(instance3.hashCode());
    }
}
/*
 * 1.分配内存空间
 * 2.执行构造方法,初始化对象
 * 3.把这个对象指向这个空间
 * 123
 * 132 A
 *     B //此时lazyman还没有完成构造
 * */

为什么要用两个if判断呢？用一个if加上synchronized不行吗？下面我们来分析为什么不能用一个if+synchronized。

public Singleton getSingleton(){
        if(singleton == null){
            synchronized(Singleton.class){
                singleton = new Singleton();
            }
        }
        return singleton;
    }

上面代码中，当多个线程在等待锁的时候，第一个线程抢到锁的线程先执行了 singleton = new Singleton();此时已经创建了一个实例,即singleton !=null。执行完后第一个线程释放了锁，后面的线程抢到了锁，又去执行 singleton = new Singleton(); 又创建了一个实例。这样就破坏了单例的性质，就不是单例模式了。所以抢到锁之后还要判断下singleton是否等于空，为空时才创建对象，不为空时就不创建对象。

所以，DCL懒汉式用了2个if+synchronized来保证线程安全。

静态内部类

package single;
//静态内部类
public class Holder {
    private Holder(){
    }
    public static Holder getInstance(){
        return InnerClass.HOLDER;
    };
    public static class InnerClass{
        private static final Holder HOLDER = new Holder();
    }
}

详情见：

单例模式的4种实现方式_聪颖之夏的博客-CSDN博客

保证一个类只有一个实例，并且提供一个全局访问点重量级的对象，不需要多个实例，如线程池，数据库连接池。

https://blog.csdn.net/XikYu/article/details/130790707?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522168802876916800225520150%2522%252C%2522scm%2522%253A%252220140713.130102334.pc%255Fblog.%2522%257D&request_id=168802876916800225520150&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~blog~first_rank_ecpm_v1~rank_v31_ecpm-1-130790707-null-null.268%5Ev1%5Ekoosearch&utm_term=%E5%8D%95%E4%BE%8B%E6%A8%A1%E5%BC%8F&spm=1018.2226.3001.4450

15 CAS

AS通俗的解释就是：

比较当前工作内存中的值和主内存中的值，如果相同则执行规定操作，否则继续比较直到主内存和工作内存中的值一致为止.

ABA

CAS存在的问题

CAS虽然很高效的解决原子操作，但是CAS仍然存在三大问题。ABA问题，循环时间长开销大和只能保证一个共享变量的原子操作

ABA问题。因为CAS需要在操作值的时候检查下值有没有发生变化，如果没有发生变化则更新，但是如果一个值原来是A，变成了B，又变成了A，那么使用CAS进行检查时会发现它的值没有发生变化，但是实际上却变化了。ABA问题的解决思路就是使用版本号。在变量前面追加上版本号，每次变量更新的时候把版本号加一，那么A－B－A 就会变成1A - 2B－3A。

public class CASDemo {
    //AtomicStampedReference 注意，如果泛型是一个包装类，注意对象的引用问题
    // 正常在业务操作，这里面比较的都是一个个对象
    static AtomicStampedReference<Integer> atomicStampedReference = new AtomicStampedReference<>(1,1);
    // CAS  compareAndSet : 比较并交换！
    public static void main(String[] args) {
        new Thread(()->{
            int stamp = atomicStampedReference.getStamp(); // 获得版本号
            System.out.println("a1=>"+stamp);
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            Lock lock = new ReentrantLock(true);
            atomicStampedReference.compareAndSet(1, 2,
                    atomicStampedReference.getStamp(), atomicStampedReference.getStamp() + 1);
            System.out.println("a2=>"+atomicStampedReference.getStamp());
            System.out.println(atomicStampedReference.compareAndSet(2, 1,
                    atomicStampedReference.getStamp(), atomicStampedReference.getStamp() + 1));
            System.out.println("a3=>"+atomicStampedReference.getStamp());
        },"a").start();
        // 乐观锁的原理相同！
        new Thread(()->{
            int stamp = atomicStampedReference.getStamp(); // 获得版本号
            System.out.println("b1=>"+stamp);
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(atomicStampedReference.compareAndSet(1, 6,
                    stamp, stamp + 1));
            System.out.println("b2=>"+atomicStampedReference.getStamp());
        },"b").start();
    }
}

UC并发编程学习笔记（简单易懂）4