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

9：四大函数式接口

9.1 Function 函数式接口

9.2 Predicate 断定型接口

9.3 Consumer 消费性接口

9.4 Supplier 供给型接口

10 Stream流式计算

11 ForkJoin（分支合并）

12 异步回调

13 Volatile(可见性)

不保证原子性

不加锁保证原子性

14 单例模式

饿汉式单例

懒汉式单例

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

静态内部类

15 CAS

ABA

16 可重入锁

17 自旋锁

热门文章

最新文章

相关课程

相关电子书

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

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

9：四大函数式接口

9.1 Function 函数式接口

9.2 Predicate 断定型接口

9.3 Consumer 消费性接口

9.4 Supplier 供给型接口

10 Stream流式计算

11 ForkJoin（分支合并）

12 异步回调

13 Volatile(可见性)

不保证原子性

不加锁保证原子性

14 单例模式

饿汉式单例

懒汉式单例

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

静态内部类

15 CAS

ABA

16 可重入锁

17 自旋锁

热门文章

最新文章

相关课程

相关电子书