xxl-rpc源码阅读笔记之RpcProvider

一直对RPC这个名词非常好奇,xxl里面为RPC提供了一个实现--xxl-rpc。轻量级、分布式,阅读源码以后觉得收获非常大。整理最近看的过程,形成笔记记录下来。关于RPC的介绍可以参考这篇博客:https://www.cnblogs.com/twinhead/p/9900605.html

xxl-rpc项目地址为:https://www.xuxueli.com/xxl-rpc/#%E3%80%8A%E5%88%86%E5%B8%83%E5%BC%8F%E6%9C%8D%E5%8A%A1%E6%A1%86%E6%9E%B6XXL-RPC%E3%80%8B

xxl-rpc可以分为服务提供者和服务消费者,这里咱们先讲RpcProvider。总的一个系统构成如下图所示:

xxl-rpc源码阅读笔记之RpcProvider

哈哈哈,画了一个很丑的图图。

 

 

 

 从应用开始,官方提供了一个无框架版的RPC服务端模板,主要代码主要是初始化RPC服务提供端:

// init
        XxlRpcProviderFactory providerFactory = new XxlRpcProviderFactory();
        providerFactory.setServer(NettyServer.class);
        providerFactory.setSerializer(HessianSerializer.class);
        providerFactory.setCorePoolSize(-1);
        providerFactory.setMaxPoolSize(-1);
        providerFactory.setIp(null);
        providerFactory.setPort(7080);
        providerFactory.setAccessToken(null);
        providerFactory.setServiceRegistry(null);
        providerFactory.setServiceRegistryParam(null);

        // add services
        providerFactory.addService(DemoService.class.getName(), null, new DemoServiceImpl());

        // start
        providerFactory.start();

        while (!Thread.currentThread().isInterrupted()) {
            TimeUnit.HOURS.sleep(1);
        }

        // stop
        providerFactory.stop();

  这里的DemoServiceImpl就是服务端提供的远端调用实现。里面只有一个方法就是返回一个UserDto对象。这里我们看到providerFactory的一些属性,包括底层的通信方案是使用的NettyServer、序列化方案是使用的Hessian。而设置的注册中心的方案是提供了三种方案的,包括本地注册、xxl-registry注册、Zookeeper三种。作者推荐的是xxl-registry方案,源码之前有看过,源码的笔记可以看我的另一篇博客。CorePoolSize、MaxPoolSize是NettyServer接收请求后处理请求的线程池。IP、Port是服务提供的地址。

这里咱们看看start()方法的代码,其实这里只实现了两个功能:向注册中心注册和启动NettyServer:

this.serializerInstance = serializer.newInstance();

		// start server
		serviceAddress = IpUtil.getIpPort(this.ip, port);
		serverInstance = server.newInstance();
		serverInstance.setStartedCallback(new BaseCallback() {		// serviceRegistry started
			@Override
			public void run() throws Exception {
				// start registry
				if (serviceRegistry != null) {
					serviceRegistryInstance = serviceRegistry.newInstance();
					serviceRegistryInstance.start(serviceRegistryParam);
					if (serviceData.size() > 0) {
						serviceRegistryInstance.registry(serviceData.keySet(), serviceAddress);
					}
				}
			}
		});
		serverInstance.setStopedCallback(new BaseCallback() {		// serviceRegistry stoped
			@Override
			public void run() {
				// stop registry
				if (serviceRegistryInstance != null) {
					if (serviceData.size() > 0) {
						serviceRegistryInstance.remove(serviceData.keySet(), serviceAddress);
					}
					serviceRegistryInstance.stop();
					serviceRegistryInstance = null;
				}
			}
		});
		serverInstance.start(this);

 注册中心因此保留了键值对:服务名称(接口名)与提供该服务的地址链表。

NettyServer这里有两种实现方案,socket与Http方案。Netty的配置可以去Netty的文档找,这里就不贴了。主要是xxl-rpc这里设置的序列化方案和Netty读取数据后的处理过程NettyServerHandler。

.addLast(new IdleStateHandler(0,0, Beat.BEAT_INTERVAL*3, TimeUnit.SECONDS))     // beat 3N, close if idle
                                            .addLast(new NettyDecoder(XxlRpcRequest.class, xxlRpcProviderFactory.getSerializerInstance()))
                                            .addLast(new NettyEncoder(XxlRpcResponse.class, xxlRpcProviderFactory.getSerializerInstance()))
                                            .addLast(new NettyServerHandler(xxlRpcProviderFactory, serverHandlerPool));

 IdleStateHandler是Netty的心跳。这里咱们需要重点看看Netty服务器接收到数据后是如何进行处理的:

@Override
    public void channelRead0(final ChannelHandlerContext ctx, final XxlRpcRequest xxlRpcRequest) throws Exception {

        // filter beat
        if (Beat.BEAT_ID.equalsIgnoreCase(xxlRpcRequest.getRequestId())){
            logger.debug(">>>>>>>>>>> xxl-rpc provider netty server read beat-ping.");
            return;
        }

        // do invoke
        try {
            serverHandlerPool.execute(new Runnable() {
                @Override
                public void run() {
                    // invoke + response
                    XxlRpcResponse xxlRpcResponse = xxlRpcProviderFactory.invokeService(xxlRpcRequest);

                    ctx.writeAndFlush(xxlRpcResponse);
                }
            });
        } catch (Exception e) {
            // catch error
            XxlRpcResponse xxlRpcResponse = new XxlRpcResponse();
            xxlRpcResponse.setRequestId(xxlRpcRequest.getRequestId());
            xxlRpcResponse.setErrorMsg(ThrowableUtil.toString(e));

            ctx.writeAndFlush(xxlRpcResponse);
        }

    }

 注意到这么一行

XxlRpcResponse xxlRpcResponse = xxlRpcProviderFactory.invokeService(xxlRpcRequest);

 xxlRpcRequest是服务端发送的请求,大体里面填充了要请求那个接口、具体哪个方法,其中会包含一个ID值,依据ID进行请求的返回。而xxlRpcResponse则是返回到客户端的内容,主要包括请求ID、执行结果,通过Netty回传到客户端。invokeService方法根据请求的接口、方法,从serviceData中取出对应的实现(如开头的new DemoServiceImpl()),通过反射得到运行结果,并将结果回传。

public XxlRpcResponse invokeService(XxlRpcRequest xxlRpcRequest) {

		//  make response
		XxlRpcResponse xxlRpcResponse = new XxlRpcResponse();
		xxlRpcResponse.setRequestId(xxlRpcRequest.getRequestId());

		// match service bean
		String serviceKey = makeServiceKey(xxlRpcRequest.getClassName(), xxlRpcRequest.getVersion());
		Object serviceBean = serviceData.get(serviceKey);

		// valid
		if (serviceBean == null) {
			xxlRpcResponse.setErrorMsg("The serviceKey["+ serviceKey +"] not found.");
			return xxlRpcResponse;
		}

		if (System.currentTimeMillis() - xxlRpcRequest.getCreateMillisTime() > 3*60*1000) {
			xxlRpcResponse.setErrorMsg("The timestamp difference between admin and executor exceeds the limit.");
			return xxlRpcResponse;
		}
		if (accessToken!=null && accessToken.trim().length()>0 && !accessToken.trim().equals(xxlRpcRequest.getAccessToken())) {
			xxlRpcResponse.setErrorMsg("The access token[" + xxlRpcRequest.getAccessToken() + "] is wrong.");
			return xxlRpcResponse;
		}

		try {
			// invoke
			Class<?> serviceClass = serviceBean.getClass();
			String methodName = xxlRpcRequest.getMethodName();
			Class<?>[] parameterTypes = xxlRpcRequest.getParameterTypes();
			Object[] parameters = xxlRpcRequest.getParameters();

            Method method = serviceClass.getMethod(methodName, parameterTypes);
            method.setAccessible(true);
			Object result = method.invoke(serviceBean, parameters);

			/*FastClass serviceFastClass = FastClass.create(serviceClass);
			FastMethod serviceFastMethod = serviceFastClass.getMethod(methodName, parameterTypes);
			Object result = serviceFastMethod.invoke(serviceBean, parameters);*/

			xxlRpcResponse.setResult(result);
		} catch (Throwable t) {
			// catch error
			logger.error("xxl-rpc provider invokeService error.", t);
			xxlRpcResponse.setErrorMsg(ThrowableUtil.toString(t));
		}

		return xxlRpcResponse;
	}

 至此服务端生成客户端请求的内容,并将内容封装在XxlRpcResponse中通过Netty回传。

 

上一篇:分布式任务调度平台XXL-JOB


下一篇:快速使用分布式定时任务 xxl-job