XXL-JOB原理--执行器注册（二）

Love The Way You Lie 2022-05-12 05:52 180阅读 0赞

1、xxl-job添加执行器到任务调度中心有两种方式

（1）客户端执行器自动将名称和机器地址注册到任务调度中心

（2）可以在任务调度中心手动录入执行器名称和相关的机器地址（多个机器地址用逗号隔开）

![70][]

2、自动注册流程

（1）在执行器客户端配置执行器名称和任务调度中心地址：

### xxl-job admin address list, such as "http://address" or "http://address01,http://address02"
    ##任务调度中心地址
    xxl.job.admin.addresses=http://127.0.0.1:8080
    
    ### xxl-job executor address
    ##任务调度器名称和机器信息
    xxl.job.executor.appname=xxl-job-executor-sample
    xxl.job.executor.ip=127.0.0.1
    xxl.job.executor.port=9998

（2）相关注册执行代码：

在执行器启动时会读取配置，当存在任务调度中心地址会依次向任务调度中心注册其地址

XxlJobExecutor类在进行初始化时会进行如下操作。

//当存在多个任务调度中心时，创建代理类并注册，在NetComClientProxy
    private static void initAdminBizList(String adminAddresses, String accessToken) throws Exception {
            if (adminAddresses!=null && adminAddresses.trim().length()>0) {
                for (String address: adminAddresses.trim().split(",")) {
                    if (address!=null && address.trim().length()>0) {
                        String addressUrl = address.concat(AdminBiz.MAPPING);
                        AdminBiz adminBiz = (AdminBiz) new NetComClientProxy(AdminBiz.class, addressUrl, accessToken).getObject();
                        if (adminBizList == null) {
                            adminBizList = new ArrayList<AdminBiz>();
                        }
                        adminBizList.add(adminBiz);
                    }
                }
            }
        }

在XxlJobExecutor被调用时执行getObject方法，完成向任务调度中心发送请求进行服务注册操作。

@Override
    	public Object getObject() throws Exception {
    		return Proxy.newProxyInstance(Thread.currentThread()
    				.getContextClassLoader(), new Class[] { iface },
    				new InvocationHandler() {
    					@Override
    					public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
    
    						// filter method like "Object.toString()"
    						if (Object.class.getName().equals(method.getDeclaringClass().getName())) {
    							logger.error(">>>>>>>>>>> xxl-rpc proxy class-method not support [{}.{}]", method.getDeclaringClass().getName(), method.getName());
    							throw new RuntimeException("xxl-rpc proxy class-method not support");
    						}
    						
    						// request
    						RpcRequest request = new RpcRequest();
    	                    request.setServerAddress(serverAddress);
    	                    request.setCreateMillisTime(System.currentTimeMillis());
    	                    request.setAccessToken(accessToken);
    	                    request.setClassName(method.getDeclaringClass().getName());
    	                    request.setMethodName(method.getName());
    	                    request.setParameterTypes(method.getParameterTypes());
    	                    request.setParameters(args);
    
    	                    // send
    						//向任务调度中心发送请求进行服务注册
    	                    RpcResponse response = client.send(request);
    	                    
    	                    // valid response
    						if (response == null) {
    							throw new Exception("Network request fail, response not found.");
    						}
    	                    if (response.isError()) {
    	                        throw new RuntimeException(response.getError());
    	                    } else {
    	                        return response.getResult();
    	                    }
    	                   
    					}
    				});

在JettyClient中调用send方法完成服务注册操作

public RpcResponse send(RpcRequest request) throws Exception {
    		try {
    			// serialize request
    			byte[] requestBytes = HessianSerializer.serialize(request);
    
    			// reqURL
    			String reqURL = request.getServerAddress();
    			if (reqURL!=null && reqURL.toLowerCase().indexOf("http")==-1) {
    				reqURL = "http://" + request.getServerAddress() + "/";	// IP:PORT, need parse to url
    			}
                //发送post请求进行服务注册，简单注册一下IP和端口信息等
    			// remote invoke
    			byte[] responseBytes = HttpClientUtil.postRequest(reqURL, requestBytes);
    			if (responseBytes == null || responseBytes.length==0) {
    				RpcResponse rpcResponse = new RpcResponse();
    				rpcResponse.setError("Network request fail, RpcResponse byte[] is null");
    				return rpcResponse;
                }
    
                // deserialize response
    			RpcResponse rpcResponse = (RpcResponse) HessianSerializer.deserialize(responseBytes, RpcResponse.class);
    			return rpcResponse;
    		} catch (Exception e) {
    			logger.error(e.getMessage(), e);
    
    			RpcResponse rpcResponse = new RpcResponse();
    			rpcResponse.setError("Network request error: " + e.getMessage());
    			return rpcResponse;
    		}
    	}

服务注册地址：[http://127.0.0.1:8080/api][http_127.0.0.1_8080_api]