并发编程20

其他 2021-02-11 10:19:56 阅读次数: 0

并发编程20

回顾

指令重排的原理

  • // 场景：在多核CPU下，a被缓存到core2下面
// core1首先把a=1执行完，然后放到store buffer中
// 

// 修改变量a，异步通知core2，
a=1;
// load变量a，此时可能是0，导致b是一个错误的值
b=a+1;
if(b==1){
      
      
	doSomething();
}

core1首先把a=1执行完，


core1                              core2
  |                                  |
  |                                  |
  |--> 先存到store buffer中       交互比较的频繁
  |                                  |
  V                                  V
cache-同步a失效的通知（a==1时执行）->   cache
  |                                  ^
  |                                  |
  ---->   b= a+1                  ----

  • 异步的store buffer，导致最终执行的效果看上去像进行了指令重排

  • 怎么解决？执行b=a+1时，从store buffer中拿，加1个内存屏障

多线程效率

  • 什么时候采用多线程？多线程效率问题？

  • 通过jmh测试

  • 怎么创建jmh？--通过下面的命令创建一个项目

mvn archetype.generate
 -DinteractiveMode=false -DarchetypeGroupId=org.openjdk.jmh
 -DarchetypeArtifactId=jmh-java-benchmark-archetype -DgroupId=enjoy -DartifactId=enjoy -Dversion=1.0
    
// 打jar包
mvn clean pakage
    
// 测试多线程速度
java -jar -Xms1g benchmark.jar

// 多核情况下
c方法：0.020
d方法：0.041
    
// 单核情况下
// 通过虚拟机配置实现单核效果
c方法和d方法实现效果差不多，实际单线程效果更好，因为多线程存在线程切换，而线程切换很消耗性能
    
多线程跟CPU硬件相关，CPU核数越多效率越高

  • import org.openjdk.jmh.annotations.*;

import java.util.Arrays;
import java.util.concurrent.FutureTask;

/**
 * 什么时候采用多线程
 *
 * 多线程的效率问题
 */
@Fork(1)
@BenchmarkMode(Mode.AverageTime)
@Warmup(iterations=3) // 预热3次
@Measurement(iterations=5) // 执行5次
public class MyBenchmark {
      
      
    // 创建了个一亿的数组
    static int[] ARRAY = new int[1000_000_00];
    static {
      
      
        // 往ARRAY这个数组里面填充1
        Arrays.fill(ARRAY, 1);
    }


    /**
     * 4个线程  对1亿数据累加
     *
     * 分而治之 普通线程池不会有任务密取----不能充分利用多核CPU的并行特点
     *
     *
     * @return
     * @throws Exception
     */
    @Benchmark
    public int c() throws Exception {
      
      
        int[] array = ARRAY;

        //1
        FutureTask<Integer> t1 = new FutureTask<>(()->{
      
      
            int sum = 0;
            for(int i = 0; i < 250_000_00;i++) {
      
      
                sum += array[0+i];
            }
            return sum;
        });


        //1s
        FutureTask<Integer> t2 = new FutureTask<>(()->{
      
      
            int sum = 0;
            for(int i = 0; i < 250_000_00;i++) {
      
      
                sum += array[250_000_00+i];
            }
            return sum;
        });



        FutureTask<Integer> t3 = new FutureTask<>(()->{
      
      
            int sum = 0;
            for(int i = 0; i < 250_000_00;i++) {
      
      
                sum += array[500_000_00+i];
            }
            return sum;
        });


        FutureTask<Integer> t4 = new FutureTask<>(()->{
      
      
            int sum = 0;
            for(int i = 0; i < 250_000_00;i++) {
      
      
                sum += array[750_000_00+i];
            }
            return sum;
        });
        new Thread(t1).start();
        new Thread(t2).start();
        new Thread(t3).start();
        new Thread(t4).start();
        return t1.get() + t2.get() + t3.get()+ t4.get();
    }


    /**
     * 单线程 对1一亿次累加
     * @return
     * @throws Exception
     */
    @Benchmark
    public int d() throws Exception {
      
      
        int[] array = ARRAY;
        FutureTask<Integer> t1 = new FutureTask<>(()->{
      
      
            int sum = 0;
            for(int i = 0; i < 1000_000_00;i++) {
      
      
                sum += array[0+i];
            }
            return sum;
        });
        new Thread(t1).start();
        return t1.get();
    }
}

  • forkjoin除了分而治之（任务拆分），还有任务窃取（有的线程空闲时可以去窃取任务）

forkjoin的api理解

  • package BingFaBianCheng.bingFaBianCheng20.xx;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 工作原理
 *
 *
 * v1  直接邀请v1用户10个或者团队人数达到30个  升级到v2
 * v2
 * v3
 * v4
 * v5
 * v6
 *
 * 现在有一个人 N 注册
 * 假设是v1 可能会触发很多升级 此时不需要返回结果 用RecuriveAction
 *
 */
public class TestForkJoin {
      
      

    public static void main(String[] args) {
      
      
        // AtomicBoolean
        // forkjoin 本身是一种线程池
        // 创建出的线程池的线程数是4个
        ForkJoinPool pool = new ForkJoinPool(4);
        // runable
        // 需要自己写一个类继承FutureTask
        // RecursiveTask类继承了FutureTask
        // RecursiveAction是没有返回值的
        
        // Task1方法里面实现了1-5的累加
        // 里面是递归实现的
        System.out.println(pool.invoke(new Task1(5)));

        //5+task1(4)  4+task(3)  3+taks(2) 2+(taks1)
       // System.out.println(pool.invoke(new Task3(1, 5)));
    }
}

@Slf4j(topic = "e.task1")
class Task1 extends RecursiveTask<Integer> {
      
      

    int n;

    public Task1(int n) {
      
      
        this.n = n;
    }

    @Override
    public String toString() {
      
      
        return "{" + n + '}';
    }

    @Override
    protected Integer compute() {
      
      
        if (n == 1) {
      
      
            log.debug("join() {}", n);
            return n;
        }
        Task1 t1 = new Task1(n - 1);
        //分叉
        t1.fork();
        log.debug("fork() {} + {}", n, t1);
        int result = n + t1.join();//update
        log.debug("join() {} + {} = {}", n, t1, result);
        return result;
    }

}



@Slf4j(topic = "e.AddTask")
class Task3 extends RecursiveTask<Integer> {
      
      

    int begin;
    int end;

    public Task3(int begin, int end) {
      
      
        this.begin = begin;
        this.end = end;
    }

    @Override
    public String toString() {
      
      
        return "{" + begin + "," + end + '}';
    }

    @Override
    protected Integer compute() {
      
      
        if (begin == end) {
      
      
            log.debug("join() {}", begin);
            return begin;
        }
        if (end - begin == 1) {
      
      
            log.debug("join() {} + {} = {}", begin, end, end + begin);
            return end + begin;
        }


        int mid = (end + begin) / 2;

        Task3 t1 = new Task3(begin, mid);
        t1.fork();



        Task3 t2 = new Task3(mid + 1, end);
        t2.fork();


        log.debug("fork() {} + {} = ?", t1, t2);

        int result = t1.join() + t2.join();
        log.debug("join() {} + {} = {}", t1, t2, result);
        return result;
    }
}

  • 比较low的一种任务拆分

  • 线程t1  5+task(4)  t1执行最后的join，并返回
			|   ^
           fork |
            |  join
            V   |
线程t2   4+task(3)  t2执行join，并返回
            |   ^
           fork |
            |  join
            V   |
线程t3   3+task(2)  t3执行join，并返回
            |   ^
           fork |
            |  join
            V   |
线程t0   2+task(1)   t0执行join，并返回
            | ^
            V |
        实际只有join(1)，由t3执行，并返回

  • Task3是一种二分的拆法，更加高效

  • @Slf4j(topic = "e.AddTask")
class Task3 extends RecursiveTask<Integer> {
      
      

    int begin;
    int end;

    public Task3(int begin, int end) {
      
      
        this.begin = begin;
        this.end = end;
    }

    @Override
    public String toString() {
      
      
        return "{" + begin + "," + end + '}';
    }

    @Override
    protected Integer compute() {
      
      
        if (begin == end) {
      
      
            log.debug("join() {}", begin);
            return begin;
        }
        if (end - begin == 1) {
      
      
            log.debug("join() {} + {} = {}", begin, end, end + begin);
            return end + begin;
        }


        int mid = (end + begin) / 2;

        Task3 t1 = new Task3(begin, mid);
        t1.fork();



        Task3 t2 = new Task3(mid + 1, end);
        t2.fork();


        log.debug("fork() {} + {} = ?", t1, t2);

        int result = t1.join() + t2.join();
        log.debug("join() {} + {} = {}", t1, t2, result);
        return result;
    }
}

  • 10亿的累计通过forkjoin实现

  • package BingFaBianCheng.bingFaBianCheng20.xx;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.LongStream;


@Slf4j(topic = "e")
public class ForkJoinTest2 extends RecursiveTask<Long> {
      
      


    private Long start;  // 1
    private Long end;    // 1990900000

    // 临界值
    // 每一个任务只执行10000个数的相加
    private Long temp = 10000L;

    public ForkJoinTest2(Long start, Long end) {
      
      
        this.start = start;
        this.end = end;
    }

    // 计算方法
    @Override
    protected Long compute() {
      
      
        ReentrantLock lock = new ReentrantLock();
        //tiao
        Condition condition = lock.newCondition();
        lock.newCondition();
		//小于一万了就不再拆了
        if ((end-start)<temp){
      
      
            Long sum = 0L;
            for (Long i = start; i <= end; i++) {
      
      
                sum += i;
            }
            return sum;
        }else {
      
       // forkjoin 递归
            long middle = (start + end) / 2; // 中间值
            ForkJoinTest2 task1 = new ForkJoinTest2(start, middle);
            task1.fork(); // 拆分任务，把任务压入线程队列

            ForkJoinTest2 task2 = new ForkJoinTest2(middle+1, end);
            task2.fork(); // 拆分任务，把任务压入线程队列

            return task1.join() + task2.join();
        }
    }


    public static void main(String[] args) throws ExecutionException, InterruptedException {
      
      
       //6596
      // test1();

       //4872
      // test2();
        //160
        test3();
    }


    /**
     * 单线程的直接10亿累加
     */
    public static void test1(){
      
      
        Long sum = 0L;
        long start = System.currentTimeMillis();
        for (Long i = 1L; i <= 10_0000_0000; i++) {
      
      
            sum += i;
        }
        long end = System.currentTimeMillis();
        System.out.println("sum="+sum+" 时间："+(end-start));
    }


    //最基本的forkjoin 二分查找拆分任务
    //拆分任务 关键在于算法 你自己写

    public static void test2(){
      
      

        long start = System.currentTimeMillis();

        ForkJoinPool forkJoinPool = new ForkJoinPool();
        ForkJoinTask<Long> task = new ForkJoinTest2(0L, 10_0000_0000L);
        Long sum = forkJoinPool.invoke(task);
//        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();
        // jdk8 stream流式计算
        // 内部实现的更加优雅的拆分方法
        long sum = LongStream.rangeClosed(0L, 10_0000_0000L).parallel().reduce(0, Long::sum);
        long end = System.currentTimeMillis();
        System.out.println("sum="+sum+"时间："+(end-start));
    }

}

  • forkjoin使用cpu密集型计算，需要通过算法设计更加优秀的拆分

gc实现

  • 内存分配运行时数据区—c语言的内存分配
  • c语言实现堆 c语言实现栈
  • 标记清除
  • 当内存不够的时候回收

高并发的问题

  • 分布式、redis(缓存)、mq、微服务、nginx(负载均衡)

  • 高并发与讲的课程并发编程是两回事

  • 并发编程解决的问题

    • 数据迁移，第三方对接，一亿个用户，表结构不同，采用多线程解决
    • excel 3g 商品数，解析excel

待解决内容

  • 无界队列—源码
  • threadLocal—源码
  • completableFuture
  • CopyOnWrite 并发容器—一种设计模式、以及结合mq或者redis怎么使用
    • mq的队列进行操作，队列长度达到一定长度后删除数据。。。
  • disruptor

一些重点

  • currentHashMap

  • 线程池如何保证核心线程不被销毁

    • 核心线程是阻塞创建
    • 空闲线程是超时机制创建

  • synchronized关键字底层原理

  • synchronized与lock的区别

  • 公平锁和非公平锁

    • 公平锁是一朝排队，永远排队
    • 非公平锁，永远不排队，直接拿锁，但是进入队列后就没有公平锁和非公平锁的区别了，都是根据顺序执行了

  • volatile

    • 为jmm服务的一个关键字，主要解决可见性和重排序，所有的重排序都是为了可见性服务
    • 重排序有三个级别，编译器级别、执行引擎级别、内存级别

  • aqs—抽象的同步队列

    • 说出几个关键结构，node内部类（head、tail双向队列），state记录锁的状态，判断是不是等于0，等于0是自由状态，直接获取锁，exclusiveThread是持有锁的线程，还有唤醒机制(park机制(parkEvent.park方法)—系统调用—重量锁(用户态到内核态的切换))

  • synchronized关键字

    • mutext_lock
    • spinlock(linux中的函数)，java实现的是通过死循环，自旋锁需要站在两个角度考虑，os级别，大部分系统都实现了自旋，语言级别，可以通过死循环实现，lock这把锁中的自旋是通过for(;;)实现

  • Condtion原理(lock.newCondition())

    • 获取锁拿不到锁的线程，会去entrylist中等待，拿到锁但是执行条件不满足会进入waiting，满足条件后去竞争队列排队，等待获取到资源就执行

  • 解决高并发—并发安全问题、雪崩问题

    • 集群，分布式、微服务(最大的优点是容错)
    • 架构设计—多维度解决问题—熔断其他业务
    • 业务代码，mq集群、redis集群

  • 4个微服务，买两台应用服务器(最节约最合理的情况，贵一点的1w多一个月)，一台mq，一台redsi，一台数据库，一台网管服务器(包含netty服务)

    • 高可用雪崩一般是tomcat崩了，所以两台服务器一般够了
    • mq—死信路由（死信交换机），死信队列不严谨

猜你喜欢

转载自blog.csdn.net/Markland_l/article/details/112915263
今日推荐
周排行
Leetcode简单题61~80
解决zookeeper磁盘IO高的问题
多线程相关方法详解
Maven-setting.xml文件详解
Maven 项目的 classpath 理解
渊亭科技大数据笔试题
配置JVM内存分配
计算机网络个人学习笔记 （三）网络层 ：第三部分 连载
js中两个等号(==)和三个等号(===)的区别
用C程序自动打开电脑上的程序
每日归档
更多
2024-09-18(0)
2024-09-17(0)
2024-09-16(0)
2024-09-15(0)
2024-09-14(0)
2024-09-13(0)
2024-09-12(0)
2024-09-11(0)
2024-09-10(0)
2024-09-09(0)

代码天地 © 2018 codetd.com

境外服务器   最新电视剧2022