手记

RPC实现原理之核心技术-异步处理机制

  1. 为什么要采用异步?

    影响到性能和吞吐量的根本原因是什么呢? 其实就是RPC请求的整体耗时,如果采用同步调用, CPU 大部分的时间都在等待而没有去计算,从而导致 CPU 的利用率不够。这就好比工地里面搬砖,砌墙,捣水泥都由一个人干,其他人旁观, 那效率就十分低下。

    RPC 请求比较耗时的原因主要是在哪里?

    > 在大多数情况下,RPC 本身处理请求的效率是在毫秒级的。RPC 请求的耗时大部分都是业务耗时,比如业务逻辑中有访问数据库执行慢 SQL 的操作,核心是在I/O瓶颈。所以说,在大多数情况下,影响到 RPC 调用的吞吐量的原因也就是业务逻辑处理慢了,CPU 大部分时间都在等待资源。

    现在流行的响应式开发,实质上就是通过异步方式提升业务处理的吞吐量,所以, 要作为一个高性能的RPC 框架必须要做到异步化,这样可以极大的提升整体吞吐量。

  2. 调用端如何实现异步?

    常用的方式就是Future 方式,它是返回 Future 对象,通过GET方式获取结果;或者采用入参为 Callback 对象的回调方式,处理结果。

    基于RPC的DUBBO框架是如何实现异步调用呢?

    用户端发送的每条消息都一个唯一的消息标识,调用端向服务端发送请求消息之前会先创建一个 Future,并会存储这个消息标识与这个 Future 的映射,动态代理所获得的返回值最终就是从这个 Future 中获取的;当收到服务端响应的消息时,调用端会根据响应消息的唯一标识,通过之前存储的映射找到对应的 Future,将结果注入给那个 Future,再进行一系列的处理逻辑,最后动态代理从 Future 中通过GET方法获得到正确的返回值。

  3. 服务端如何实现异步?

    RPC 服务端接收到请求的二进制消息之后会根据协议进行拆包解包,之后将完整的消息进行解码并反序列化,获得到入参参数之后再通过反射执行业务逻辑。那这些操作都是由一个线程负责执行吗?

    为了提升性能,不会放在一个线程处理, 这个就是服务端的异步化。对二进制消息数据包拆解包的处理是放在处理网络 IO 的线程中,而解码与反序列化不涉及业务逻辑, 一般也是放在 IO 线程中处理,那服务端的业务逻辑呢?业务逻辑是应该交给专门的业务线程池处理,以防止由于业务逻辑处理得过慢而影响到网络 IO 的处理。

    但大家思考下, 如果滥用多线程, 业务线程是很容易就被打满了,吞吐量很不理想,并且这时 CPU 的利用率也很低,那这个时候,服务端业务处理逻辑加入异步处理机制, 解耦处理不同逻辑, 将最终的结果以回调的方式响应给调用端, 是个很好的方法。

  4. RPC框架的异步实现

    RPC 框架的异步策略主要是调用端异步与服务端异步。调用端的异步就是通过 Future 方式实现异步,调用端发起一次异步请求并且从请求上下文中拿到一个 Future,之后通过 Future 的 get 方法获取结果,如果业务逻辑中同时调用多个其它的服务,则可以通过 Future 的方式减少业务逻辑的耗时,提升吞吐量。

    服务端异步则需要一种回调方式,让业务逻辑可以异步处理,之后调用 RPC 框架提供的回调接口,将最终结果异步通知给调用端。这样就实现了RPC调用的全异步。

    Dubbo源码:

    异步调用: AsyncToSyncInvoker.invoke方法

    @Override
        public Result invoke(Invocation invocation) throws RpcException {
            // 默认采用异步方式调用
            Result asyncResult = invoker.invoke(invocation);
    
            try {
                // 如果是同步方式调用, 将阻塞时间调整为最大
                if (InvokeMode.SYNC == ((RpcInvocation) invocation).getInvokeMode()) {
                    asyncResult.get(Integer.MAX_VALUE, TimeUnit.MILLISECONDS);
                }
            } catch (InterruptedException e) {
                throw new RpcException("Interrupted unexpectedly while waiting for remoting result to return!  method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
            } catch (ExecutionException e) {
                Throwable t = e.getCause();
                if (t instanceof TimeoutException) {
                    throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
                } else if (t instanceof RemotingException) {
                    throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
                }
            } catch (Throwable e) {
                throw new RpcException(e.getMessage(), e);
            }
            return asyncResult;
        }
    

    获取结果:ChannelWrappedInvoker.doInvoke方法

    @Override
        protected Result doInvoke(Invocation invocation) throws Throwable {
            RpcInvocation inv = (RpcInvocation) invocation;
            // use interface's name as service path to export if it's not found on client side
            inv.setAttachment(PATH_KEY, getInterface().getName());
            inv.setAttachment(CALLBACK_SERVICE_KEY, serviceKey);
    
            try {
                if (RpcUtils.isOneway(getUrl(), inv)) { // may have concurrency issue
                    currentClient.send(inv, getUrl().getMethodParameter(invocation.getMethodName(), SENT_KEY, false));
                    return AsyncRpcResult.newDefaultAsyncResult(invocation);
                } else {
                    // 获取响应future对象
                    CompletableFuture responseFuture = currentClient.request(inv);
                    AsyncRpcResult asyncRpcResult = new AsyncRpcResult(inv);
                    responseFuture.whenComplete((appResponse, t) -> {
                        if (t != null) {
                            // 异常处理
                            asyncRpcResult.completeExceptionally(t);
                        } else {
                            // 正常获取结果,触发响应回调
                            asyncRpcResult.complete((AppResponse) appResponse);
                        }
                    });
                    return asyncRpcResult;
                }
            } catch (RpcException e) {
                throw e;
            } catch (TimeoutException e) {
                throw new RpcException(RpcException.TIMEOUT_EXCEPTION, e.getMessage(), e);
            } catch (RemotingException e) {
                throw new RpcException(RpcException.NETWORK_EXCEPTION, e.getMessage(), e);
            } catch (Throwable e) { // here is non-biz exception, wrap it.
                throw new RpcException(e.getMessage(), e);
            }
        }
    

0人推荐
随时随地看视频
慕课网APP