2020.04.10 线上性能优化以及Linux下NIO/Epoll模型全解--实战

1.支付宝模拟线上优化实战

2.手写JUC工具与提升tomcat吞吐量

3.网络通信BIO设计与缺陷   -- accept()  和 read()阻塞     

4.单线程解决高并发NIO精髓解读

5.OS内核下Epoll与Selete源码解读

第一部分: 性能优化

问题：如何在高并发场景下实现支付宝用户登录页面的信息获取?如用户信息，金额，积分等

浏览器  ---- Spring MVC ---- controller ----- service ---- 用户模块、余额模块、积分模块等

-- 调用多个系统提供的远程接口，返回数据结果(json格式)

第一种做法 传统做法：

@Service
public class UserServiceImpl implements UserService {

    @Autowired
    RemoteService remoteService;

    //做一个调多系统提供的远程接口
    @Override
    public String getUserInfo(String userId) {

        String v1 = remoteService.getUserInfo(userId);
        JSONObject userInfo  = JSONObject.parseObject(v1); // {1, 2, 3}

        String v2 = remoteService.getUserMoney(userId);
        JSONObject moneyInfo  = JSONObject.parseObject(v2); // {4, 5, 6}

        //合并
        JSONObject result = new JSONObject();
        result.putAll(userInfo);
        result.putAll(moneyInfo);  // result {1, 2, 3, 4, 5, 6}

        return result.toString();
    }
}

 4秒时间，不可以接受，时间太长

第二种：是不是考虑线程呢？

thread实现runnable接口，但是没有返回结果，也行不通。

Callable接口有返回值，与 FutureTask连用

FutureTask -----extends------runnable----run(todo??)--->Callable.call()

注意：

FutureTask.get()  获取线程返回结果,阻塞的。

 具体代码实现如下：

Callable<JSONObject> userCall = new Callable<JSONObject>() {
            @Override
            public JSONObject call() throws Exception {
                String v1 = remoteService.getUserInfo(userId);  //httpClient URL http
                JSONObject userInfo  = JSONObject.parseObject(v1); // {1, 2, 3}
                return userInfo;
            }
        };

        Callable<JSONObject> moneyCall = new Callable<JSONObject>() {
            @Override
            public JSONObject call() throws Exception {
                String v1 = remoteService.getUserInfo(userId);  //httpClient URL http
                JSONObject userInfo  = JSONObject.parseObject(v1); // {1, 2, 3}
                return userInfo;
            }
        };

        FutureTask<JSONObject> userTask = new FutureTask(userCall);
        FutureTask<JSONObject> moneyTask = new FutureTask(moneyCall);
        new Thread(userTask).start();
        new Thread(moneyTask).start();


        //合并
        JSONObject result = new JSONObject();
        try {
            result.putAll(userTask.get());   //阻塞  --没有拿到值,就想当与暂时，拿到值后就放行，如 F6 F8等
            result.putAll(moneyTask.get());  // result {1, 2, 3, 4, 5, 6}
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        System.out.println("执行消耗时间:" + (System.currentTimeMillis() - start));

   时间花费从4秒瞬间将至2秒，也是不可以接受的。

第二部分：手写JUC工具与提升tomcat吞吐量

自己手写一个FutureTask实现类：QianBingFutureTask.java

package com.suning.aaa.impl.service.publicclass;

import java.util.concurrent.*;

public class QianBingFutureTask<V> implements Runnable, Future<V> {

    Callable<V> callable;

    V result = null;

    public QianBingFutureTask(Callable<V> userCall) {
        this.callable = callable;
    }

    @Override
    public void run() {
        try {
            result = callable.call(); //获取返回结果
            System.out.println(result);  // -----此值一定会打印
            if(result != null){
                synchronized (this){
                    this.notifyAll(); //唤醒
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        return false;
    }

    @Override
    public boolean isCancelled() {
        return false;
    }

    @Override
    public boolean isDone() {
        return false;
    }

    @Override
    public V get() throws InterruptedException, ExecutionException {
        if (result != null){
            return result;
        }
        System.out.println(result); // 此值不会打印
        //没有取到result值，此时会阻塞
        synchronized (this){
            this.wait(); //阻塞
        }
        return null;
    }

    @Override
    public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        return null;
    }
}

1. 第一步: 实现同FutureTask接口类  runnable、Future<V>
public class QianBingFutureTask<V> implements Runnable, Future<V>

第二步：构建构造函数，参数Callable

Callable<V> callable;
public QianBingFutureTask(Callable<V> userCall) {
    this.callable = callable;
}

第三步:实现run()方法，得到Callable.call()返回结果值并唤醒其他线程
 定义一个全局变量 V result= null;

第四步:实现get()方法，取到结果集则返回；拿不到值则阻塞

tomcat吞吐量提升

tomcat7以后    BIO   NIO

一个tomcat并发量 500---800个请求。------------ 是谁在执行? tomcat线程

          tomcat --- server.xml----port:8080----maxThread=150　　

即tomcat在启动时候, 初始化150个线程. 

controller返回值定义为Callable接口

package com.suning.fiaas.controller.publicClass;
import com.suning.fiaas.intf.service.publicClass.UserService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;
import java.util.concurrent.Callable;
@Controller
@RequestMapping("/user")
public class UserController {
    @Autowired
    private UserService userService;
    /**
     * 获取用户信息(用户接口+余额接口)
     * @param userId
     * @return
     */
    @RequestMapping(value = "/getUserInfo", method = RequestMethod.GET )
    @ResponseBody
    public Callable<String> getUserInfo(String userId) {
        System.out.println("tomcat主线程开始-----" + Thread.currentThread() + "-----" + System.currentTimeMillis());
        Callable<String> callable = new Callable<String>() {
            @Override
            public String call() throws Exception {
                System.out.println("子线程开始-----" + Thread.currentThread() + "-----" + System.currentTimeMillis());
                String result =  userService.getUserInfo(userId);
                System.out.println("子线程开始-----" + Thread.currentThread() + "-----" + System.currentTimeMillis());
                return result;
            }
        };
        System.out.println("tomcat主线程结束-----" + Thread.currentThread() + "-----" + System.currentTimeMillis());
        return callable;
    }
}

 

tomcat优化与第一部分优化后2s相比，吞吐量提升2000倍。150个tomcat呢，提升1500*2000倍。

第三部分:BIO和NIO

服务器端代码:

package com.suning.fiaas.controller.publicClass.socket;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;

public class SocketServerTest {
    static byte[] bs = new byte[1024];

    public static void main(String[] args) throws IOException {
        ServerSocket socketServer = new ServerSocket(6363);
        while (true){
            System.out.println("==========wait connection===========");
            Socket socketClinet = socketServer.accept();  //阻塞
            System.out.println("==========success connection===========");

            System.out.println("==========wait read===========");
            socketClinet.getInputStream().read(bs);       //阻塞
            System.out.println("==========success read===========");

            System.out.println(new String(bs));
        }
    }
}

客户端代码：

package com.suning.fiaas.controller.publicClass.socket;

import java.io.IOException;
import java.net.Socket;

public class SocketClient {
    public static void main(String[] args) throws IOException {
        Socket socket = new Socket("127.0.0.1", 6363);
        socket.getOutputStream().write("abc".getBytes());
        socket.close();
    }
}

-- 验证serverSocket.accept()是阻塞的场景

-- 验证socketClinet.getInputStream().read(bs);是阻塞的场景  --- 键盘输入

 客户端：

问题： 服务器端 socketServer.accept() 和 socketClinet.getInputStream().read(bs);这两方法是阻塞的  ---> BIO

      10000个连接，如果只有1000个活跃，剩下的9000不活跃，也是浪费资源。

 BIO： 对每一个请求都开启一个线程链接。线程越多，线程上下文切换越多，资源越浪费。

package com.suning.fiaas.controller.publicClass.socket;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.List;

public class BIOSocketServer {
    static byte[] bs = new byte[1024];
    static List<Socket> socketList = null;

    public static void main(String[] args) throws IOException {
        ServerSocket socketServer = new ServerSocket(6363);
        while (true){
            //遍历老的链接
            for (Socket socket : socketList){
                socket.setBlocking(false);
                int read = socket.getInputStream().read(bs);   // 非阻塞
                if (read > 0){
                    System.out.println("read ......" + read );
                }
            }

            //添加新的链接
            socketServer.setBlocking(false);
            Socket socketClinet = socketServer.accept();    // 非阻塞
            if (socketClinet != null){
                socketList.add(socketClinet);
            }
            System.out.println("socketList length当前链接数目:" + socketList.size());
        }
    }
}

对于上面大量的for循环，java交给jvm处理-----内核。

socketList委托给内核处理，只需要1ms，性能提到大大的提升。for循环不应该出现在java中。

 Selete():     java --- native c----select(socketList)----java----循环

第四部分: NIO精髓解读与OS内核下Epoll与Selete源码解读

NIO ---- 利用单线程来处理并发问题 --- epoll模型

NIOSocketServer.java 

package com.suning.fiaas.controller.publicClass.socket;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;

public class NIOSocketServer {
    static ByteBuffer byteBuffer = ByteBuffer.allocate(512);
    static List<SocketChannel> socketList = new ArrayList<>();

    //NIO设计就是利用单线程来解决并发的问题
    public static void main(String[] args) throws IOException {
        ServerSocketChannel serverSocket = ServerSocketChannel.open();

        SocketAddress socketAddress = new InetSocketAddress("127.0.0.1", 6363)
        serverSocket.bind(socketAddress);
        serverSocket.configureBlocking(false);  //去除阻塞

        while (true){
            //遍历老的链接
            for (SocketChannel socketChannel : socketList){
                int read = socketChannel.read(byteBuffer);
                if (read > 0){
                    System.out.println("read ......" + read );
                    byteBuffer.flip();

                    byte[] bs = new byte[read];
                    byteBuffer.get(bs);
                    String content = new String(bs);
                    System.out.println("content ......" + content );
                    byteBuffer.flip();
                }
            }

            //添加新的链接
            SocketChannel accept = serverSocket.accept();
            if (accept != null){
                System.out.println("conn succ");
                accept.configureBlocking(false);  //去除阻塞
                socketList.add(accept);
                System.out.println("socketList length:" + socketList.size());
            }
        }
    }
}

操作系统 

window     提供  Selete函数(socketList)   ----> 非开源

linux        epoll  >   Selete()                      ----> epoll效率要比Selete效率更高

 -- man 2 select

-- man epoll

-- nc -l localhost 8080 

比如 redis、nginx只提供了linux版本的。

linux ---> linux ---->epoll()    c语言{

　　prinf()

}

小工具: 网络调试助手