RPC基本原理以及如何用Netty來實現RPC

時間  2019-11-06
標籤 rpc 基本原理 以及 如何 netty 實現 欄目 微服務 简体版
原文   原文鏈接

前言

在微服務大行其道的今天，分佈式系統愈來愈重要，實現服務化首先就要考慮服務之間的通訊問題。這裏面涉及序列化、反序列化、尋址、鏈接等等問題。。不過，有了RPC框架，咱們就無需苦惱。node

1、什麼是RPC？

RPC（Remote Procedure Call）— 遠程過程調用，是一個計算機通訊協議。該協議容許運行於一臺計算機的程序調用另外一臺計算機的子程序，而程序員無需額外地爲這個交互做用編程。git

值得注意是，兩個或多個應用程序都分佈在不一樣的服務器上，它們之間的調用都像是本地方法調用同樣。 程序員

RPC遠程過程調用

RPC框架有不少，比較知名的如阿里的Dubbo、google的gRPC、Go語言的rpcx、Apache的thrift。固然了，還有Spring Cloud，不過對於Spring Cloud來講，RPC只是它的一個功能模塊。github

複雜的先不講，若是要實現一個基本功能、簡單的RPC，要涉及哪些東西呢？數據庫

  • 動態代理
  • 反射
  • 序列化、反序列化
  • 網絡通訊
  • 編解碼
  • 服務發現和註冊
  • 心跳與鏈路檢測
  • ......

下面咱們一塊兒經過代碼來分析，怎麼把這些技術點串到一塊兒，實現咱們本身的RPC。編程

2、環境準備

在開始以前，筆者先介紹一下所用到的軟件環境。json

SpringBoot、Netty、zookeeper、zkclient、fastjsonbootstrap

  • SpringBoot 項目的基礎框架，方便打成JAR包，便於測試。
  • Netty 通訊服務器
  • zookeeper 服務的發現與註冊
  • zkclient zookeeper客戶端
  • fastjson 序列化、反序列化

3、RPC生產者

一、服務接口API

整個RPC，咱們分爲生產者和消費者。首先它們有一個共同的服務接口API。在這裏，咱們搞一個操做用戶信息的service接口。緩存

public interface InfoUserService {
    List<InfoUser> insertInfoUser(InfoUser infoUser);
    InfoUser getInfoUserById(String id);
    void deleteInfoUserById(String id);
    String getNameById(String id);
    Map<String,InfoUser> getAllUser();
}
複製代碼

二、服務類實現

做爲生產者，它固然要有實現類，咱們建立InfoUserServiceImpl實現類，並用註解把它標註爲RPC的服務，而後註冊到Spring的Bean容器中。在這裏，咱們把infoUserMap當作數據庫，存儲用戶信息。bash

package com.viewscenes.netsupervisor.service.impl;

@RpcService
public class InfoUserServiceImpl implements InfoUserService {

    Logger logger = LoggerFactory.getLogger(this.getClass());
	//當作數據庫，存儲用戶信息
    Map<String,InfoUser> infoUserMap = new HashMap<>();

    public List<InfoUser> insertInfoUser(InfoUser infoUser) {
        logger.info("新增用戶信息:{}", JSONObject.toJSONString(infoUser));
        infoUserMap.put(infoUser.getId(),infoUser);
        return getInfoUserList();
    }
    public InfoUser getInfoUserById(String id) {
        InfoUser infoUser = infoUserMap.get(id);
        logger.info("查詢用戶ID:{}",id);
        return infoUser;
    }

    public List<InfoUser> getInfoUserList() {
        List<InfoUser> userList = new ArrayList<>();
        Iterator<Map.Entry<String, InfoUser>> iterator = infoUserMap.entrySet().iterator();
        while (iterator.hasNext()){
            Map.Entry<String, InfoUser> next = iterator.next();
            userList.add(next.getValue());
        }
        logger.info("返回用戶信息記錄數:{}",userList.size());
        return userList;
    }
    public void deleteInfoUserById(String id) {
        logger.info("刪除用戶信息:{}",JSONObject.toJSONString(infoUserMap.remove(id)));
    }
    public String getNameById(String id){
        logger.info("根據ID查詢用戶名稱:{}",id);
        return infoUserMap.get(id).getName();
    }
    public Map<String,InfoUser> getAllUser(){
        logger.info("查詢全部用戶信息{}",infoUserMap.keySet().size());
        return infoUserMap;
    }
}
複製代碼

元註解定義以下：

package com.viewscenes.netsupervisor.annotation;

@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Component
public @interface RpcService {}
複製代碼

三、請求信息和返回信息

全部的請求信息和返回信息，咱們用兩個JavaBean來表示。其中的重點是，返回信息要帶有請求信息的ID。

package com.viewscenes.netsupervisor.entity;
public class Request {
	private String id;
	private String className;// 類名
	private String methodName;// 函數名稱
	private Class<?>[] parameterTypes;// 參數類型
	private Object[] parameters;// 參數列表
	get/set ...
}
複製代碼
package com.viewscenes.netsupervisor.entity;
public class Response {
	private String requestId;
	private int code;
	private String error_msg;
	private Object data;
	get/set ...
}
複製代碼

四、Netty服務端

Netty做爲高性能的NIO通訊框架，在不少RPC框架中都有它的身影。咱們也採用它當作通訊服務器。說到這，咱們先看個配置文件，重點有兩個，zookeeper的註冊地址和Netty通訊服務器的地址。

TOMCAT端口
server.port=8001
#zookeeper註冊地址
registry.address=192.168.245.131:2181,192.168.245.131:2182,192.168.245.131:2183
#RPC服務提供者地址
rpc.server.address=192.168.197.1:18868
複製代碼

爲了方便管理，咱們把它也註冊成Bean，同時實現ApplicationContextAware接口，把上面@RpcService註解的服務類撈出來，緩存起來，供消費者調用。同時，做爲服務器，還要對客戶端的鏈路進行心跳檢測，超過60秒未讀寫數據，關閉此鏈接。

package com.viewscenes.netsupervisor.netty.server;
@Component
public class NettyServer implements ApplicationContextAware,InitializingBean{

    private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);
    private static final EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    private static final EventLoopGroup workerGroup = new NioEventLoopGroup(4);

    private Map<String, Object> serviceMap = new HashMap<>();

    @Value("${rpc.server.address}")
    private String serverAddress;

    @Autowired
    ServiceRegistry registry;

    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(RpcService.class);
        for(Object serviceBean:beans.values()){
            Class<?> clazz = serviceBean.getClass();
            Class<?>[] interfaces = clazz.getInterfaces();
            for (Class<?> inter : interfaces){
                String interfaceName = inter.getName();
                logger.info("加載服務類: {}", interfaceName);
                serviceMap.put(interfaceName, serviceBean);
            }
        }
        logger.info("已加載所有服務接口:{}", serviceMap);
    }
    public void afterPropertiesSet() throws Exception {
        start();
    }
    public void start(){
        final NettyServerHandler handler = new NettyServerHandler(serviceMap);
        new Thread(() -> {
            try {
                ServerBootstrap bootstrap = new ServerBootstrap();
                bootstrap.group(bossGroup,workerGroup).
                        channel(NioServerSocketChannel.class).
                        option(ChannelOption.SO_BACKLOG,1024).
                        childOption(ChannelOption.SO_KEEPALIVE,true).
                        childOption(ChannelOption.TCP_NODELAY,true).
                        childHandler(new ChannelInitializer<SocketChannel>() {
                            //建立NIOSocketChannel成功後，在進行初始化時，將它的ChannelHandler設置到ChannelPipeline中，用於處理網絡IO事件
                            protected void initChannel(SocketChannel channel) throws Exception {
                                ChannelPipeline pipeline = channel.pipeline();
                                pipeline.addLast(new IdleStateHandler(0, 0, 60));
                                pipeline.addLast(new JSONEncoder());
                                pipeline.addLast(new JSONDecoder());
                                pipeline.addLast(handler);
                            }
                        });
                String[] array = serverAddress.split(":");
                String host = array[0];
                int port = Integer.parseInt(array[1]);
                ChannelFuture cf = bootstrap.bind(host,port).sync();
                logger.info("RPC 服務器啓動.監聽端口:"+port);
                registry.register(serverAddress);
                //等待服務端監聽端口關閉
                cf.channel().closeFuture().sync();
            } catch (Exception e) {
                e.printStackTrace();
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }).start();
    }
}
複製代碼

上面的代碼就把Netty服務器啓動了，在處理器中的構造函數中，咱們先把服務Bean的Map傳進來，全部的處理要基於這個Map才能找到對應的實現類。在channelRead中，獲取請求方法的信息，而後經過反射調用方法獲取返回值。

package com.viewscenes.netsupervisor.netty.server;
@ChannelHandler.Sharable
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    private final Logger logger = LoggerFactory.getLogger(NettyServerHandler.class);
    private final Map<String, Object> serviceMap;

    public NettyServerHandler(Map<String, Object> serviceMap) {
        this.serviceMap = serviceMap;
    }
    public void channelActive(ChannelHandlerContext ctx)   {
        logger.info("客戶端鏈接成功!"+ctx.channel().remoteAddress());
    }
    public void channelInactive(ChannelHandlerContext ctx)   {
        logger.info("客戶端斷開鏈接!{}",ctx.channel().remoteAddress());
        ctx.channel().close();
    }
    public void channelRead(ChannelHandlerContext ctx, Object msg)   {
        Request request = JSON.parseObject(msg.toString(),Request.class);

        if ("heartBeat".equals(request.getMethodName())) {
            logger.info("客戶端心跳信息..."+ctx.channel().remoteAddress());
        }else{
            logger.info("RPC客戶端請求接口:"+request.getClassName()+" 方法名:"+request.getMethodName());
            Response response = new Response();
            response.setRequestId(request.getId());
            try {
                Object result = this.handler(request);
                response.setData(result);
            } catch (Throwable e) {
                e.printStackTrace();
                response.setCode(1);
                response.setError_msg(e.toString());
                logger.error("RPC Server handle request error",e);
            }
            ctx.writeAndFlush(response);
        }
    }
    /**
     * 經過反射，執行本地方法
     * @param request
     * @return
     * @throws Throwable
     */
    private Object handler(Request request) throws Throwable{
        String className = request.getClassName();
        Object serviceBean = serviceMap.get(className);

        if (serviceBean!=null){
            Class<?> serviceClass = serviceBean.getClass();
            String methodName = request.getMethodName();
            Class<?>[] parameterTypes = request.getParameterTypes();
            Object[] parameters = request.getParameters();

            Method method = serviceClass.getMethod(methodName, parameterTypes);
            method.setAccessible(true);
            return method.invoke(serviceBean, getParameters(parameterTypes,parameters));
        }else{
            throw new Exception("未找到服務接口,請檢查配置!:"+className+"#"+request.getMethodName());
        }
    }
    /**
     * 獲取參數列表
     * @param parameterTypes
     * @param parameters
     * @return
     */
    private Object[] getParameters(Class<?>[] parameterTypes,Object[] parameters){
        if (parameters==null || parameters.length==0){
            return parameters;
        }else{
            Object[] new_parameters = new Object[parameters.length];
            for(int i=0;i<parameters.length;i++){
                new_parameters[i] = JSON.parseObject(parameters[i].toString(),parameterTypes[i]);
            }
            return new_parameters;
        }
    }
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt)throws Exception {
        if (evt instanceof IdleStateEvent){
            IdleStateEvent event = (IdleStateEvent)evt;
            if (event.state()== IdleState.ALL_IDLE){
                logger.info("客戶端已超過60秒未讀寫數據,關閉鏈接.{}",ctx.channel().remoteAddress());
                ctx.channel().close();
            }
        }else{
            super.userEventTriggered(ctx,evt);
        }
    }
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)   {
        logger.info(cause.getMessage());
        ctx.close();
    }
}
複製代碼

四、服務註冊

咱們啓動了Netty通訊服務器，而且把服務實現類加載到緩存，等待請求時調用。這一步，咱們要進行服務註冊。爲了簡單化處理，咱們只註冊通訊服務器的監聽地址便可。 在上面代碼中，bind以後咱們執行了registry.register(serverAddress); 它的做用就是，將Netty監聽的IP端口註冊到zookeeper。

package com.viewscenes.netsupervisor.registry;
@Component
public class ServiceRegistry {
    Logger logger = LoggerFactory.getLogger(this.getClass());
    @Value("${registry.address}")
    private String registryAddress;
    private static final String ZK_REGISTRY_PATH = "/rpc";

    public void register(String data) {
        if (data != null) {
            ZkClient client = connectServer();
            if (client != null) {
                AddRootNode(client);
                createNode(client, data);
            }
        }
    }
	//鏈接zookeeper
    private ZkClient connectServer() {
        ZkClient client = new ZkClient(registryAddress,20000,20000);
        return client;
    }
	//建立根目錄/rpc
    private void AddRootNode(ZkClient client){
        boolean exists = client.exists(ZK_REGISTRY_PATH);
        if (!exists){
            client.createPersistent(ZK_REGISTRY_PATH);
            logger.info("建立zookeeper主節點 {}",ZK_REGISTRY_PATH);
        }
    }
	//在/rpc根目錄下，建立臨時順序子節點
    private void createNode(ZkClient client, String data) {
        String path = client.create(ZK_REGISTRY_PATH + "/provider", data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        logger.info("建立zookeeper數據節點 ({} => {})", path, data);
    }
}
複製代碼

有一點須要注意，子節點必須是臨時節點。這樣，生產者端停掉以後，才能通知到消費者，把此服務從服務列表中剔除。到此爲止，生產者端已經完成。咱們看一下它的啓動日誌：

加載服務類: com.viewscenes.netsupervisor.service.InfoUserService
已加載所有服務接口:{com.viewscenes.netsupervisor.service.InfoUserService=com.viewscenes.netsupervisor.service.impl.InfoUserServiceImpl@46cc127b}
Initializing ExecutorService 'applicationTaskExecutor'
Tomcat started on port(s): 8001 (http) with context path ''
Started RpcProviderApplication in 2.003 seconds (JVM running for 3.1)
RPC 服務器啓動.監聽端口:18868
Starting ZkClient event thread.
Socket connection established to 192.168.245.131/192.168.245.131:2183, initiating session
Session establishment complete on server 192.168.245.131/192.168.245.131:2183, sessionid = 0x367835b48970010, negotiated timeout = 4000
zookeeper state changed (SyncConnected)
建立zookeeper主節點 /rpc
建立zookeeper數據節點 (/rpc/provider0000000000 => 192.168.197.1:28868)
複製代碼

4、RPC消費者

首先，咱們須要把生產者端的服務接口API，即InfoUserService。以相同的目錄放到消費者端。路徑不一樣，調用會找不到的哦。

一、代理

RPC的目標其中有一條，《程序員無需額外地爲這個交互做用編程。》因此，咱們在調用的時候，就像調用本地方法同樣。就像下面這樣：

@Controller
public class IndexController {	
	@Autowired
    InfoUserService userService;
	
	@RequestMapping("getById")
    @ResponseBody
    public InfoUser getById(String id){
        logger.info("根據ID查詢用戶信息:{}",id);
        return userService.getInfoUserById(id);
    }
}
複製代碼

那麼，問題來了。消費者端並無此接口的實現，怎麼調用到的呢？這裏，首先就是代理。筆者這裏用的是Spring的工廠Bean機制建立的代理對象，涉及的代碼較多，就不在文章中體現了，若是有不懂的同窗，請想象一下，MyBatis中的Mapper接口怎麼被調用的。能夠參考筆者文章：Mybatis源碼分析（四）mapper接口方法是怎樣被調用到的

總之，在調用userService方法的時候，會調用到代理對象的invoke方法。在這裏，封裝請求信息，而後調用Netty的客戶端方法發送消息。而後根據方法返回值類型，轉成相應的對象返回。

package com.viewscenes.netsupervisor.configurer.rpc;

@Component
public class RpcFactory<T> implements InvocationHandler {

    @Autowired
    NettyClient client;

    Logger logger = LoggerFactory.getLogger(this.getClass());
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        Request request = new Request();
        request.setClassName(method.getDeclaringClass().getName());
        request.setMethodName(method.getName());
        request.setParameters(args);
        request.setParameterTypes(method.getParameterTypes());
        request.setId(IdUtil.getId());

        Object result = client.send(request);
        Class<?> returnType = method.getReturnType();

        Response response = JSON.parseObject(result.toString(), Response.class);
        if (response.getCode()==1){
            throw new Exception(response.getError_msg());
        }
        if (returnType.isPrimitive() || String.class.isAssignableFrom(returnType)){
            return response.getData();
        }else if (Collection.class.isAssignableFrom(returnType)){
            return JSONArray.parseArray(response.getData().toString(),Object.class);
        }else if(Map.class.isAssignableFrom(returnType)){
            return JSON.parseObject(response.getData().toString(),Map.class);
        }else{
            Object data = response.getData();
            return JSONObject.parseObject(data.toString(), returnType);
        }
    }
}
複製代碼

二、服務發現

在生產者端，咱們把服務IP端口都註冊到zookeeper中，因此這裏，咱們要去拿到服務地址，而後經過Netty鏈接。重要的是，還要對根目錄進行監聽子節點變化，這樣隨着生產者的上線和下線，消費者端能夠及時感知。

package com.viewscenes.netsupervisor.connection;

@Component
public class ServiceDiscovery {

    @Value("${registry.address}")
    private String registryAddress;
    @Autowired
    ConnectManage connectManage;

    // 服務地址列表
    private volatile List<String> addressList = new ArrayList<>();
    private static final String ZK_REGISTRY_PATH = "/rpc";
    private ZkClient client;

    Logger logger = LoggerFactory.getLogger(this.getClass());

    @PostConstruct
    public void init(){
        client = connectServer();
        if (client != null) {
            watchNode(client);
        }
    }
	
	//鏈接zookeeper
    private ZkClient connectServer() {
        ZkClient client = new ZkClient(registryAddress,30000,30000);
        return client;
    }
	//監聽子節點數據變化
    private void watchNode(final ZkClient client) {
        List<String> nodeList = client.subscribeChildChanges(ZK_REGISTRY_PATH, (s, nodes) -> {
            logger.info("監聽到子節點數據變化{}",JSONObject.toJSONString(nodes));
            addressList.clear();
            getNodeData(nodes);
            updateConnectedServer();
        });
        getNodeData(nodeList);
        logger.info("已發現服務列表...{}", JSONObject.toJSONString(addressList));
        updateConnectedServer();
    }
	//鏈接生產者端服務
    private void updateConnectedServer(){
        connectManage.updateConnectServer(addressList);
    }

    private void getNodeData(List<String> nodes){
        logger.info("/rpc子節點數據爲:{}", JSONObject.toJSONString(nodes));
        for(String node:nodes){
            String address = client.readData(ZK_REGISTRY_PATH+"/"+node);
            addressList.add(address);
        }
    }
}
複製代碼

其中，connectManage.updateConnectServer(addressList);就是根據服務地址，去鏈接生產者端的Netty服務。而後建立一個Channel列表，在發送消息的時候，從中選取一個Channel和生產者端進行通訊。

三、Netty客戶端

Netty客戶端有兩個方法比較重要，一個是根據IP端口鏈接服務器，返回Channel，加入到鏈接管理器;一個是用Channel發送請求數據。同時，做爲客戶端，空閒的時候還要往服務端發送心跳信息。

package com.viewscenes.netsupervisor.netty.client;

@Component
public class NettyClient {
    Logger logger = LoggerFactory.getLogger(this.getClass());
    private EventLoopGroup group = new NioEventLoopGroup(1);
    private Bootstrap bootstrap = new Bootstrap();
    @Autowired
    NettyClientHandler clientHandler;
    @Autowired
    ConnectManage connectManage;
   
    public Object send(Request request) throws InterruptedException{

        Channel channel = connectManage.chooseChannel();
        if (channel!=null && channel.isActive()) {
            SynchronousQueue<Object> queue = clientHandler.sendRequest(request,channel);
            Object result = queue.take();
            return JSONArray.toJSONString(result);
        }else{
            Response res = new Response();
            res.setCode(1);
            res.setError_msg("未正確鏈接到服務器.請檢查相關配置信息!");
            return JSONArray.toJSONString(res);
        }
    }
    public Channel doConnect(SocketAddress address) throws InterruptedException {
        ChannelFuture future = bootstrap.connect(address);
        Channel channel = future.sync().channel();
        return channel;
    }
	....其餘方法略
}
複製代碼

咱們必須重點關注send方法，它是在代理對象invoke方法調用到的。首先從鏈接器中輪詢選擇一個Channel，而後發送數據。可是，Netty是異步操做，咱們還要轉爲同步，就是說要等待生產者端返回數據才往下執行。筆者在這裏用的是同步隊列SynchronousQueue，它的take方法會阻塞在這裏，直到裏面有數據可讀。而後在處理器中，拿到返回信息寫到隊列中，take方法返回。

package com.viewscenes.netsupervisor.netty.client;
@Component
@ChannelHandler.Sharable
public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    @Autowired
    NettyClient client;
    @Autowired
    ConnectManage connectManage;
    Logger logger = LoggerFactory.getLogger(this.getClass());
    private ConcurrentHashMap<String,SynchronousQueue<Object>> queueMap = new ConcurrentHashMap<>();

    public void channelActive(ChannelHandlerContext ctx)   {
        logger.info("已鏈接到RPC服務器.{}",ctx.channel().remoteAddress());
    }
    public void channelInactive(ChannelHandlerContext ctx)   {
        InetSocketAddress address =(InetSocketAddress) ctx.channel().remoteAddress();
        logger.info("與RPC服務器斷開鏈接."+address);
        ctx.channel().close();
        connectManage.removeChannel(ctx.channel());
    }
    public void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {
        Response response = JSON.parseObject(msg.toString(),Response.class);
        String requestId = response.getRequestId();
        SynchronousQueue<Object> queue = queueMap.get(requestId);
        queue.put(response);
        queueMap.remove(requestId);
    }
    public SynchronousQueue<Object> sendRequest(Request request,Channel channel) {
        SynchronousQueue<Object> queue = new SynchronousQueue<>();
        queueMap.put(request.getId(), queue);
        channel.writeAndFlush(request);
        return queue;
    }
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt)throws Exception {
        logger.info("已超過30秒未與RPC服務器進行讀寫操做!將發送心跳消息...");
        if (evt instanceof IdleStateEvent){
            IdleStateEvent event = (IdleStateEvent)evt;
            if (event.state()== IdleState.ALL_IDLE){
                Request request = new Request();
                request.setMethodName("heartBeat");
                ctx.channel().writeAndFlush(request);
            }
        }else{
            super.userEventTriggered(ctx,evt);
        }
    }
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause){
        logger.info("RPC通訊服務器發生異常.{}",cause);
        ctx.channel().close();
    }
}
複製代碼

至此，消費者端也基本完成。一樣的，咱們先看一下啓動日誌：

Waiting for keeper state SyncConnected
Opening socket connection to server 192.168.139.129/192.168.139.129:2181. Will not attempt to authenticate using SASL (unknown error)
Socket connection established to 192.168.139.129/192.168.139.129:2181, initiating session
Session establishment complete on server 192.168.139.129/192.168.139.129:2181, sessionid = 0x100000273ba002c, negotiated timeout = 20000
zookeeper state changed (SyncConnected)
/rpc子節點數據爲:["provider0000000015"]
已發現服務列表...["192.168.100.74:18868"]
加入Channel到鏈接管理器./192.168.100.74:18868
已鏈接到RPC服務器./192.168.100.74:18868
Initializing ExecutorService 'applicationTaskExecutor'
Tomcat started on port(s): 7002 (http) with context path ''
Started RpcConsumerApplication in 4.218 seconds (JVM running for 5.569)
複製代碼

5、測試

咱們以Controller裏面的兩個方法爲例，先開啓100個線程調用insertInfoUser方法，而後開啓1000個線程調用查詢方法getAllUser。

public class IndexController {

    Logger logger = LoggerFactory.getLogger(this.getClass());
    @Autowired
    InfoUserService userService;

    @RequestMapping("insert")
    @ResponseBody
    public List<InfoUser> getUserList() throws InterruptedException {
        long start = System.currentTimeMillis();
        int thread_count = 100;
        CountDownLatch countDownLatch = new CountDownLatch(thread_count);
        for (int i=0;i<thread_count;i++){
            new Thread(() -> {
                InfoUser infoUser = new InfoUser(IdUtil.getId(),"Jeen","BeiJing");
                List<InfoUser> users = userService.insertInfoUser(infoUser);
                logger.info("返回用戶信息記錄:{}", JSON.toJSONString(users));
                countDownLatch.countDown();
            }).start();
        }
        countDownLatch.await();
        long end = System.currentTimeMillis();
        logger.info("線程數：{},執行時間:{}",thread_count,(end-start));
        return null;
    }
	@RequestMapping("getAllUser")
    @ResponseBody
    public Map<String,InfoUser> getAllUser() throws InterruptedException {

        long start = System.currentTimeMillis();
        int thread_count = 1000;
        CountDownLatch countDownLatch = new CountDownLatch(thread_count);
        for (int i=0;i<thread_count;i++){
            new Thread(() -> {
                Map<String, InfoUser> allUser = userService.getAllUser();
                logger.info("查詢全部用戶信息：{}",JSONObject.toJSONString(allUser));
                countDownLatch.countDown();
            }).start();
        }
        countDownLatch.await();
        long end = System.currentTimeMillis();
        logger.info("線程數：{},執行時間:{}",thread_count,(end-start));

        return null;
    }
}
複製代碼

結果以下：

新增用戶信息
查詢用戶信息.png

6、總結

本文簡單介紹了RPC的整個流程，若是你正在學習RPC的相關知識，能夠根據文中的例子，本身實現一遍。相信寫完以後，你會對RPC會有更深一些的認識。

生產者端流程：

  • 加載服務，並緩存
  • 啓動通信服務器（Netty）
  • 服務註冊（把通信地址放入zookeeper，也能夠把加載到的服務也放進去）
  • 反射，本地調用

消費者端流程：

  • 代理服務接口
  • 服務發現（鏈接zookeeper，拿到服務地址列表）
  • 遠程調用（輪詢生產者服務列表，發送消息）

限於篇幅，本文代碼並不完整，若有須要，訪問：https://github.com/taoxun/simple_rpc 或者添加筆者微信公衆號:<清幽之地的博客>），獲取完整項目。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程