高并发篇_1 多线程和高并发基础_高并发下多线程的使用

高并发篇_1 多线程和高并发基础

田超凡

原创博文，仿冒必究，部分素材转载自每特教育蚂蚁课堂

1 进程和线程

进程是系统运行程序的基本单位，一个进程至少有一个组成，一个进程中只有一条主线程，其他线程是子线程。

CPU通过划分时间片从而实现多线程并发交替执行(CPU上下文切换执行)，并不是每一时刻多线程同时执行的

线程工作内容的不同导致对CPU的消耗程度是不一样的，常见的线程工作模式分为CPU密集型线程和IO密集型线程两类：

CPU密集型线程指的是线程运行过程中需要频繁消耗CPU进行运算，IO密集型线程指的是线程运行过程中需要频繁进行IO操作(频繁进行用户态和内核态的切换)，但是对CPU计算资源消耗不大。

 

创建线程的方式：

1. 继承Thread类
2. 实现Runnable接口
3. 基于Callable和FutureTask创建线程异步获取县城执行结果
4. Jdk8+ lambda表达式创建线程
5. SpringMVC @Async创建线程
6. 使用匿名内部类创建线程
7. 基于线程池Executor创建线程

 

线程的状态

创建：实例化线程

就绪：start()

运行:  执行run()

阻塞：sleep()，join()

死亡：执行完run()

2 多线程常用API

Thread.currentThread().getName()获取当前线程名

T. setName(“设置线程名”);

t.start();启动

获取和设置线程优先级(1-10)

setPriority()/getPriority()

Thread.MAX_PRIORITY最高优先级10

Thread.MIN_PRIORITY最低优先级1

Thread.NORM_PRIORITY默认优先级5

注意：优先级高的线程分到cpu资源的概率高

线程的休眠 Thread.sleep(long ms);//参数为休眠时长，单位毫秒ms,休眠的线程处于阻塞状态

线程的强制执行t.join();//强制执行，该线程执行完后再执行其他线程

线程的礼让t.yield();//不确定是否一定会执行，而是把执行机会让给其他线程，该线程仍可能继续分配到CPU资源

LockSupport.park() 阻塞当前线程

LockSupport.unpark(Thread t) 唤醒阻塞的线程

 

3 线程同步和线程安全

当需要处理多线程并发共享数据引发的数据不安全问题时，可以使用同步方法或同步代码块实现，保护类型安全。可以理解为上锁，同一时刻只允许一个线程访问同步方法或同步代码块，其他线程可以访问非同步方法和非同步代码块。

同步方法：

访问修饰符 synchronized 返回值 方法名(参数列表)

{

//方法体

}

或

synchronized 访问修饰符 返回值 方法名(参数列表)

{

//方法体

}

 

同步代码块：

synchronized(this)

{

//同步代码块主体

}

 

4 并发和线程安全

什么是并发？

并发指的是多个线程共享同一变量的情况下，多线程同时运行操作该变量，由于多线程交替执行，CPU不停地进行上下文切换，导致该变量的值总是不符合正常理解逻辑的预期。

 

 

例如如下代码，创建2个线程并行执行，每个线程把共享全局静态变量sum从1加到10000，求最后sum的结果

如果从代码表面上理解，结果肯定是20000，但是实际运行结果会发现最终sum<20000的概率占99.9%，基本总是达不到20000，这是为什么呢？

这就是出现了并发导致的线程不安全问题

那么从代码层面如何发现问题呢？

使用javap -p -v Thread01.class 查看CPU具体执行的指令

这4行指令的意思是：

#2 表示全局静态变量sum

getstatic 获取全局静态变量的值 0

iconst_1 声明每次循环增长的步长为1

iadd 执行sum+1运算

putstatic 将sum+1后的结果重新赋值给全局静态变量sum，此时sum=1

 

这是一个线程一次循环执行的4行指令，那么两个线程一次循环执行后的结果是多少呢？很容易想到是2，但是其实不是，如图所示，这是因为在多线程并行运行的情况下产生了并发问题，CPU上下文切换导致出现了线程不安全问题。

由此可见，两个线程都只执行一次循环后，最终结果仍然是1，那么以此类推可以知道两个线程各执行10000次循环后，结果肯定是小于20000的，这就是从字节码+CPU上下文切换角度来分析出来的线程安全问题。

 

1. CPU调度算法

1. CPU的中文名称是中央处理器，是进行逻辑运算用的，主要由运算器、控制器、寄存器三部分组成，从字面意思看就是运算就是起着运算的作用，控制器就是负责发出cpu每条指令所需要的信息，寄存器就是保存运算或者指令的一些临时文件，这样可以保证更高的速度，也就是我们的线程运行在cpu之上。

 

1. CPU上下文切换又叫做CPU调度，常见的CPU调度算法有三种：

先来先服务、最短作业法、优先级调度法

先来先服务：哪个线程先启动，CPU先调度哪个线程

 

最短作业法：哪个线程消耗CPU时间短，CPU先调度哪个线程

 

优先级调度法：哪个线程设置的优先级高，CPU先调度哪个线程

 

1. Callable和FutureTask获取线程执行结果实现原理

Callable和FutureTask基本使用：

package com.tcf.juc.exercise.day01;

 

import java.util.concurrent.Callable;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.FutureTask;

 

/***

 * TODO TCF 3 基于Callable和FutureTask创建线程获取异步线程执行结果

 */

public class CallableAndFutureTaskTest {

 

    public static void main(String[] args)

    {

        Callable<Integer> callable = new Callable<Integer>() {

 

            @Override

            public Integer call() throws Exception

            {

                System.out.println("开始执行耗时业务代码......");

                try

                {

                    Thread.sleep(3000);

                }

                catch (InterruptedException e)

                {

                    e.printStackTrace();

                }

 

                return 1;

            }

        };

 

        FutureTask<Integer> futureTask = new FutureTask<Integer>(callable);

        new Thread(futureTask).start();

        try

        {

            Integer result = futureTask.get();

            System.out.println(result);

        }

        catch (Exception e)

        {

            e.printStackTrace();

        }

    }

}

 

 

自定义Callable及其实现：

package com.tcf.juc.exercise.day01;

 

@FunctionalInterface

public interface KidCallable<V> {

 

    V call();

 

}

 

package com.tcf.juc.exercise.day01;

public class KidCallableImpl<Integer> implements KidCallable<java.lang.Integer> {

 

    @Override

    public java.lang.Integer call()

    {

        System.out.println("BEGIN INVOKE LONGTIME BUSINESS CODES......");

        try

        {

            Thread.sleep(3000);

        }

        catch (InterruptedException e)

        {

            e.printStackTrace();

        }

 

        return 1;

    }

 

}

 

 

自定义FutureTask

package com.tcf.juc.exercise.day01;

import java.util.concurrent.locks.LockSupport;

public class KidFutureTask<V> implements Runnable {

 

    private KidCallable<V> kidCallable;

    // private Object lock = new Object();

    private Thread mainThread;

    private V result;

 

    // TODO TCF 构造注入

    public KidFutureTask(KidCallable<V> kidCallable)

    {

        this.kidCallable = kidCallable;

    }

 

    @Override

    public void run()

    {

        this.result = kidCallable.call();

        /*synchronized (lock)

        {

            lock.notify();

        }*/

        if(mainThread!=null)

        {

            LockSupport.unpark(mainThread);

        }

    }

 

    public V get()

    {

        /*synchronized (lock)

        {

            try

            {

                lock.wait();

            }

            catch (InterruptedException e)

            {

                e.printStackTrace();

            }

        }*/

        this.mainThread = Thread.currentThread();

        LockSupport.park();

        return result;

    }

 

}

 

整合自定义Callable和FutureTask

public static void main(String[] args)

{

        KidCallable<Integer> kidCallable = new KidCallableImpl<Integer>();

        KidFutureTask<Integer> kidFutureTask = new KidFutureTask<Integer>(kidCallable);

        new Thread(kidFutureTask).start();

        Integer result = kidFutureTask.get();

        System.out.println(result);

}

 

1. 基于无界阻塞队列手写异步日志采集框架

日志工具类Logger

package com.tcf.juc.exercise.day01;

 

import org.apache.commons.lang3.StringUtils;

import org.springframework.stereotype.Component;

import java.util.concurrent.LinkedBlockingQueue;

 

/***

 * TODO TCF 4 基于无界队列和多线程实现异步日志采集

 */

@Component

public class Logger {

 

    private LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<String>();

    private WriteLogThread writeLogThread;

 

    public Logger()

    {

        this.writeLogThread = new WriteLogThread();

        this.writeLogThread.start();

    }

 

    // TODO TCF 写入日志

    public void info(String text)

    {

        if(StringUtils.isNotEmpty(text))

        {

            queue.offer(text);

        }

    }

 

    class WriteLogThread extends Thread

    {

        @Override

        public void run()

        {

            while(true)

            {

                String text = queue.poll();

                StringBuilder stringBuilder = new StringBuilder();

                for(int i=1; i<=6; i++)

                {

                    if(i==1 && StringUtils.isNotEmpty(text))

                    {

                        stringBuilder.append(text);

                        continue;

                    }

 

                    text = queue.poll();

                    if(StringUtils.isNotEmpty(text))

                    {

                        stringBuilder.append(text);

                    }

                }

 

                if(StringUtils.isNotEmpty(stringBuilder.toString()))

                {

                    FileUtils.writeText("D:\\test\\controllerAspect.txt", stringBuilder.toString(), true);

                }

            }

        }

    }

 

}

 

Spring AOP切面类

package com.tcf.juc.exercise.day01;

 

import com.alibaba.fastjson.JSONObject;

import lombok.extern.slf4j.Slf4j;

import org.aspectj.lang.JoinPoint;

import org.aspectj.lang.annotation.Aspect;

import org.aspectj.lang.annotation.Before;

import org.aspectj.lang.annotation.Pointcut;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

import org.springframework.web.context.request.RequestContextHolder;

import org.springframework.web.context.request.ServletRequestAttributes;

 

import javax.servlet.http.HttpServletRequest;

import java.text.SimpleDateFormat;

import java.util.Date;

import java.util.Objects;

 

/***

 * TODO TCF Controller AOP切面

 */

@Component

@Aspect

public class ControllerAspect {

 

    @Autowired

    private Logger logger;

 

    @Pointcut("execution(public * com.tcf.juc.exercise.day01.*Controller.*(..))")

    public void pointcut()

    {

 

    }

 

    @Before("pointcut()")

    public void before(JoinPoint joinPoint)

    {

        ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();

        HttpServletRequest request = Objects.requireNonNull(attributes).getRequest();

 

        // TODO TCF 打印日志

        logger.info("【请求 时间】：" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));

        logger.info("【请求 URL】：" + request.getRequestURL());

        logger.info("【请求 IP】：" + request.getRemoteAddr());

        logger.info("【类名 Class】：" + joinPoint.getSignature().getDeclaringTypeName());

        logger.info("【方法名 Method】：" + joinPoint.getSignature().getName());

        logger.info("【请求参数 Args】：" +

                JSONObject.toJSONString(joinPoint.getArgs())

        );

    }

 

}