ソースを参照

调度中心回调API服务改为restful方式;

xuxueli 5 年 前
コミット
2ced601035

+ 6 - 4
doc/XXL-JOB官方文档.md ファイルの表示

1588
 - 17、DB脚本默认编码改为utf8mb4,修复字符乱码问题(建议Mysql版本5.7+);
1588
 - 17、DB脚本默认编码改为utf8mb4,修复字符乱码问题(建议Mysql版本5.7+);
1589
 - 18、调度中心任务平均分配,触发组件每次获取与线程池数量相关数量的任务,避免大量任务集中在单个调度中心集群节点;
1589
 - 18、调度中心任务平均分配,触发组件每次获取与线程池数量相关数量的任务,避免大量任务集中在单个调度中心集群节点;
1590
 - 19、调度中心移除SQL中的 "now()" 函数;集群部署时不再依赖DB时钟,仅需要保证调度中心应用节点时钟一致即可;
1590
 - 19、调度中心移除SQL中的 "now()" 函数;集群部署时不再依赖DB时钟,仅需要保证调度中心应用节点时钟一致即可;
1591
-- 20、[ING]xxl-rpc服务端线程优化,降低线程内存开销;
1592
-- 21、[ING]调度日志优化:支持设置日志保留天数,过期日志天维度记录报表,并清理;调度报表汇总实时数据和报表;
1593
-- 22、[ING]调度中心日志删除,改为分页获取ID,根据ID删除的方式;
1594
-- 23、[ING]任务回调改为restful方式;
1591
+- 20、xxl-rpc服务端线程优化,降低线程内存开销;
1592
+- 21、调度中心回调API服务改为restful方式;
1593
+- 22、[ING]调度日志优化:支持设置日志保留天数,过期日志天维度记录报表,并清理;调度报表汇总实时数据和报表;
1594
+- 23、[ING]调度中心日志删除,改为分页获取ID,根据ID删除的方式;
1595
+
1595
 
1596
 
1596
 
1597
 
1597
 ### TODO LIST
1598
 ### TODO LIST
1630
 - 33、任务执行一次的时候指定IP;
1631
 - 33、任务执行一次的时候指定IP;
1631
 - 34、通讯调整;双向HTTP,回调和其他API自定义AccessToken,Restful,执行器复用容器端口;
1632
 - 34、通讯调整;双向HTTP,回调和其他API自定义AccessToken,Restful,执行器复用容器端口;
1632
 - 35、父子任务参数传递;流程任务等,透传动态参数;
1633
 - 35、父子任务参数传递;流程任务等,透传动态参数;
1634
+- 36、任务操作API服务调整为和回调服务一致,降低接入成本;
1633
 
1635
 
1634
 
1636
 
1635
 ## 七、其他
1637
 ## 七、其他

+ 1 - 1
pom.xml ファイルの表示

24
 		<maven.compiler.target>1.7</maven.compiler.target>
24
 		<maven.compiler.target>1.7</maven.compiler.target>
25
 		<maven.test.skip>true</maven.test.skip>
25
 		<maven.test.skip>true</maven.test.skip>
26
 
26
 
27
-		<xxl-rpc.version>1.4.1</xxl-rpc.version>
27
+		<xxl-rpc.version>1.4.2</xxl-rpc.version>
28
 
28
 
29
 		<spring.version>4.3.25.RELEASE</spring.version>
29
 		<spring.version>4.3.25.RELEASE</spring.version>
30
 		<spring-boot.version>1.5.22.RELEASE</spring-boot.version>
30
 		<spring-boot.version>1.5.22.RELEASE</spring-boot.version>

+ 74 - 11
xxl-job-admin/src/main/java/com/xxl/job/admin/controller/JobApiController.java ファイルの表示

1
 package com.xxl.job.admin.controller;
1
 package com.xxl.job.admin.controller;
2
 
2
 
3
 import com.xxl.job.admin.controller.annotation.PermissionLimit;
3
 import com.xxl.job.admin.controller.annotation.PermissionLimit;
4
-import com.xxl.job.admin.core.scheduler.XxlJobScheduler;
4
+import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
5
 import com.xxl.job.core.biz.AdminBiz;
5
 import com.xxl.job.core.biz.AdminBiz;
6
-import org.springframework.beans.factory.InitializingBean;
6
+import com.xxl.job.core.biz.model.HandleCallbackParam;
7
+import com.xxl.job.core.biz.model.RegistryParam;
8
+import com.xxl.job.core.biz.model.ReturnT;
9
+import com.xxl.job.core.util.XxlJobRemotingUtil;
7
 import org.springframework.stereotype.Controller;
10
 import org.springframework.stereotype.Controller;
11
+import org.springframework.web.bind.annotation.RequestBody;
8
 import org.springframework.web.bind.annotation.RequestMapping;
12
 import org.springframework.web.bind.annotation.RequestMapping;
13
+import org.springframework.web.bind.annotation.ResponseBody;
9
 
14
 
10
-import javax.servlet.ServletException;
15
+import javax.annotation.Resource;
11
 import javax.servlet.http.HttpServletRequest;
16
 import javax.servlet.http.HttpServletRequest;
12
-import javax.servlet.http.HttpServletResponse;
13
-import java.io.IOException;
17
+import java.util.List;
14
 
18
 
15
 /**
19
 /**
16
  * Created by xuxueli on 17/5/10.
20
  * Created by xuxueli on 17/5/10.
17
  */
21
  */
18
 @Controller
22
 @Controller
19
-public class JobApiController implements InitializingBean {
23
+@RequestMapping("/api")
24
+public class JobApiController {
20
 
25
 
26
+    @Resource
27
+    private AdminBiz adminBiz;
21
 
28
 
22
-    @Override
23
-    public void afterPropertiesSet() throws Exception {
24
 
29
 
30
+    // ---------------------- admin biz ----------------------
31
+
32
+    /**
33
+     * callback
34
+     *
35
+     * @param callbackParamList
36
+     * @return
37
+     */
38
+    @RequestMapping("/callback")
39
+    @ResponseBody
40
+    @PermissionLimit(limit=false)
41
+    public ReturnT<String> callback(HttpServletRequest request, @RequestBody List<HandleCallbackParam> callbackParamList) {
42
+
43
+        if (XxlJobAdminConfig.getAdminConfig().getAccessToken()!=null
44
+                && XxlJobAdminConfig.getAdminConfig().getAccessToken().trim().length()>0
45
+                && !XxlJobAdminConfig.getAdminConfig().getAccessToken().equals(request.getHeader(XxlJobRemotingUtil.XXL_RPC_ACCESS_TOKEN))) {
46
+            return new ReturnT<String>(ReturnT.FAIL_CODE, "The access token is wrong.");
47
+        }
48
+
49
+        return adminBiz.callback(callbackParamList);
50
+    }
51
+
52
+
53
+
54
+    /**
55
+     * registry
56
+     *
57
+     * @param registryParam
58
+     * @return
59
+     */
60
+    @RequestMapping("/registry")
61
+    @ResponseBody
62
+    @PermissionLimit(limit=false)
63
+    public ReturnT<String> registry(HttpServletRequest request, @RequestBody RegistryParam registryParam) {
64
+
65
+        if (XxlJobAdminConfig.getAdminConfig().getAccessToken()!=null
66
+                && XxlJobAdminConfig.getAdminConfig().getAccessToken().trim().length()>0
67
+                && !XxlJobAdminConfig.getAdminConfig().getAccessToken().equals(request.getHeader(XxlJobRemotingUtil.XXL_RPC_ACCESS_TOKEN))) {
68
+            return new ReturnT<String>(ReturnT.FAIL_CODE, "The access token is wrong.");
69
+        }
70
+
71
+        return adminBiz.registry(registryParam);
25
     }
72
     }
26
 
73
 
27
-    @RequestMapping(AdminBiz.MAPPING)
74
+    /**
75
+     * registry remove
76
+     *
77
+     * @param registryParam
78
+     * @return
79
+     */
80
+    @RequestMapping("/registryRemove")
81
+    @ResponseBody
28
     @PermissionLimit(limit=false)
82
     @PermissionLimit(limit=false)
29
-    public void api(HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException {
30
-        XxlJobScheduler.invokeAdminService(request, response);
83
+    public ReturnT<String> registryRemove(HttpServletRequest request, @RequestBody RegistryParam registryParam) {
84
+
85
+        if (XxlJobAdminConfig.getAdminConfig().getAccessToken()!=null
86
+                && XxlJobAdminConfig.getAdminConfig().getAccessToken().trim().length()>0
87
+                && !XxlJobAdminConfig.getAdminConfig().getAccessToken().equals(request.getHeader(XxlJobRemotingUtil.XXL_RPC_ACCESS_TOKEN))) {
88
+            return new ReturnT<String>(ReturnT.FAIL_CODE, "The access token is wrong.");
89
+        }
90
+
91
+        return adminBiz.registryRemove(registryParam);
31
     }
92
     }
32
 
93
 
33
 
94
 
95
+    // ---------------------- job biz ----------------------
96
+
34
 }
97
 }

+ 0 - 7
xxl-job-admin/src/main/java/com/xxl/job/admin/core/conf/XxlJobAdminConfig.java ファイルの表示

5
 import com.xxl.job.admin.dao.XxlJobInfoDao;
5
 import com.xxl.job.admin.dao.XxlJobInfoDao;
6
 import com.xxl.job.admin.dao.XxlJobLogDao;
6
 import com.xxl.job.admin.dao.XxlJobLogDao;
7
 import com.xxl.job.admin.dao.XxlJobRegistryDao;
7
 import com.xxl.job.admin.dao.XxlJobRegistryDao;
8
-import com.xxl.job.core.biz.AdminBiz;
9
 import org.springframework.beans.factory.DisposableBean;
8
 import org.springframework.beans.factory.DisposableBean;
10
 import org.springframework.beans.factory.InitializingBean;
9
 import org.springframework.beans.factory.InitializingBean;
11
 import org.springframework.beans.factory.annotation.Value;
10
 import org.springframework.beans.factory.annotation.Value;
77
     @Resource
76
     @Resource
78
     private XxlJobGroupDao xxlJobGroupDao;
77
     private XxlJobGroupDao xxlJobGroupDao;
79
     @Resource
78
     @Resource
80
-    private AdminBiz adminBiz;
81
-    @Resource
82
     private JavaMailSender mailSender;
79
     private JavaMailSender mailSender;
83
     @Resource
80
     @Resource
84
     private DataSource dataSource;
81
     private DataSource dataSource;
126
         return xxlJobGroupDao;
123
         return xxlJobGroupDao;
127
     }
124
     }
128
 
125
 
129
-    public AdminBiz getAdminBiz() {
130
-        return adminBiz;
131
-    }
132
-
133
     public JavaMailSender getMailSender() {
126
     public JavaMailSender getMailSender() {
134
         return mailSender;
127
         return mailSender;
135
     }
128
     }

+ 16 - 55
xxl-job-admin/src/main/java/com/xxl/job/admin/core/scheduler/XxlJobScheduler.java ファイルの表示

6
 import com.xxl.job.admin.core.thread.JobScheduleHelper;
6
 import com.xxl.job.admin.core.thread.JobScheduleHelper;
7
 import com.xxl.job.admin.core.thread.JobTriggerPoolHelper;
7
 import com.xxl.job.admin.core.thread.JobTriggerPoolHelper;
8
 import com.xxl.job.admin.core.util.I18nUtil;
8
 import com.xxl.job.admin.core.util.I18nUtil;
9
-import com.xxl.job.core.biz.AdminBiz;
10
 import com.xxl.job.core.biz.ExecutorBiz;
9
 import com.xxl.job.core.biz.ExecutorBiz;
11
 import com.xxl.job.core.enums.ExecutorBlockStrategyEnum;
10
 import com.xxl.job.core.enums.ExecutorBlockStrategyEnum;
12
-import com.xxl.rpc.remoting.invoker.XxlRpcInvokerFactory;
13
 import com.xxl.rpc.remoting.invoker.call.CallType;
11
 import com.xxl.rpc.remoting.invoker.call.CallType;
14
 import com.xxl.rpc.remoting.invoker.reference.XxlRpcReferenceBean;
12
 import com.xxl.rpc.remoting.invoker.reference.XxlRpcReferenceBean;
15
 import com.xxl.rpc.remoting.invoker.route.LoadBalance;
13
 import com.xxl.rpc.remoting.invoker.route.LoadBalance;
16
-import com.xxl.rpc.remoting.net.NetEnum;
17
-import com.xxl.rpc.remoting.net.impl.servlet.server.ServletServerHandler;
18
-import com.xxl.rpc.remoting.provider.XxlRpcProviderFactory;
19
-import com.xxl.rpc.serialize.Serializer;
14
+import com.xxl.rpc.remoting.net.impl.netty_http.client.NettyHttpClient;
15
+import com.xxl.rpc.serialize.impl.HessianSerializer;
20
 import org.slf4j.Logger;
16
 import org.slf4j.Logger;
21
 import org.slf4j.LoggerFactory;
17
 import org.slf4j.LoggerFactory;
22
 
18
 
23
-import javax.servlet.ServletException;
24
-import javax.servlet.http.HttpServletRequest;
25
-import javax.servlet.http.HttpServletResponse;
26
-import java.io.IOException;
27
 import java.util.concurrent.ConcurrentHashMap;
19
 import java.util.concurrent.ConcurrentHashMap;
28
 import java.util.concurrent.ConcurrentMap;
20
 import java.util.concurrent.ConcurrentMap;
29
 
21
 
39
         // init i18n
31
         // init i18n
40
         initI18n();
32
         initI18n();
41
 
33
 
42
-        // admin-server
43
-        initRpcProvider();
44
-
45
         // admin registry monitor run
34
         // admin registry monitor run
46
         JobRegistryMonitorHelper.getInstance().start();
35
         JobRegistryMonitorHelper.getInstance().start();
47
 
36
 
72
         // admin registry stop
61
         // admin registry stop
73
         JobRegistryMonitorHelper.getInstance().toStop();
62
         JobRegistryMonitorHelper.getInstance().toStop();
74
 
63
 
75
-        // admin-server
76
-        stopRpcProvider();
77
     }
64
     }
78
 
65
 
79
     // ---------------------- I18n ----------------------
66
     // ---------------------- I18n ----------------------
84
         }
71
         }
85
     }
72
     }
86
 
73
 
87
-    // ---------------------- admin rpc provider (no server version) ----------------------
88
-    private static ServletServerHandler servletServerHandler;
89
-    private void initRpcProvider(){
90
-        // init
91
-        XxlRpcProviderFactory xxlRpcProviderFactory = new XxlRpcProviderFactory();
92
-        xxlRpcProviderFactory.initConfig(
93
-                NetEnum.NETTY_HTTP,
94
-                Serializer.SerializeEnum.HESSIAN.getSerializer(),
95
-                null,
96
-                0,
97
-                XxlJobAdminConfig.getAdminConfig().getAccessToken(),
98
-                null,
99
-                null);
100
-
101
-        // add services
102
-        xxlRpcProviderFactory.addService(AdminBiz.class.getName(), null, XxlJobAdminConfig.getAdminConfig().getAdminBiz());
103
-
104
-        // servlet handler
105
-        servletServerHandler = new ServletServerHandler(xxlRpcProviderFactory);
106
-    }
107
-    private void stopRpcProvider() throws Exception {
108
-        XxlRpcInvokerFactory.getInstance().stop();
109
-    }
110
-    public static void invokeAdminService(HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException {
111
-        servletServerHandler.handle(null, request, response);
112
-    }
113
-
114
-
115
     // ---------------------- executor-client ----------------------
74
     // ---------------------- executor-client ----------------------
116
     private static ConcurrentMap<String, ExecutorBiz> executorBizRepository = new ConcurrentHashMap<String, ExecutorBiz>();
75
     private static ConcurrentMap<String, ExecutorBiz> executorBizRepository = new ConcurrentHashMap<String, ExecutorBiz>();
117
     public static ExecutorBiz getExecutorBiz(String address) throws Exception {
76
     public static ExecutorBiz getExecutorBiz(String address) throws Exception {
128
         }
87
         }
129
 
88
 
130
         // set-cache
89
         // set-cache
131
-        executorBiz = (ExecutorBiz) new XxlRpcReferenceBean(
132
-                NetEnum.NETTY_HTTP,
133
-                Serializer.SerializeEnum.HESSIAN.getSerializer(),
134
-                CallType.SYNC,
135
-                LoadBalance.ROUND,
136
-                ExecutorBiz.class,
137
-                null,
138
-                3000,
139
-                address,
140
-                XxlJobAdminConfig.getAdminConfig().getAccessToken(),
141
-                null,
142
-                null).getObject();
90
+        XxlRpcReferenceBean referenceBean = new XxlRpcReferenceBean();
91
+        referenceBean.setClient(NettyHttpClient.class);
92
+        referenceBean.setSerializer(HessianSerializer.class);
93
+        referenceBean.setCallType(CallType.SYNC);
94
+        referenceBean.setLoadBalance(LoadBalance.ROUND);
95
+        referenceBean.setIface(ExecutorBiz.class);
96
+        referenceBean.setVersion(null);
97
+        referenceBean.setTimeout(3000);
98
+        referenceBean.setAddress(address);
99
+        referenceBean.setAccessToken(XxlJobAdminConfig.getAdminConfig().getAccessToken());
100
+        referenceBean.setInvokeCallback(null);
101
+        referenceBean.setInvokerFactory(null);
102
+
103
+        executorBiz = (ExecutorBiz) referenceBean.getObject();
143
 
104
 
144
         executorBizRepository.put(address, executorBiz);
105
         executorBizRepository.put(address, executorBiz);
145
         return executorBiz;
106
         return executorBiz;

+ 17 - 0
xxl-job-admin/src/main/java/com/xxl/job/admin/service/impl/AdminBizImpl.java ファイルの表示

17
 import org.slf4j.Logger;
17
 import org.slf4j.Logger;
18
 import org.slf4j.LoggerFactory;
18
 import org.slf4j.LoggerFactory;
19
 import org.springframework.stereotype.Service;
19
 import org.springframework.stereotype.Service;
20
+import org.springframework.util.StringUtils;
20
 
21
 
21
 import javax.annotation.Resource;
22
 import javax.annotation.Resource;
22
 import java.text.MessageFormat;
23
 import java.text.MessageFormat;
126
 
127
 
127
     @Override
128
     @Override
128
     public ReturnT<String> registry(RegistryParam registryParam) {
129
     public ReturnT<String> registry(RegistryParam registryParam) {
130
+
131
+        // valid
132
+        if (!StringUtils.hasText(registryParam.getRegistGroup())
133
+                || !StringUtils.hasText(registryParam.getRegistryKey())
134
+                || !StringUtils.hasText(registryParam.getRegistryValue())) {
135
+            return new ReturnT<String>(ReturnT.FAIL_CODE, "Illegal Argument.");
136
+        }
137
+
129
         int ret = xxlJobRegistryDao.registryUpdate(registryParam.getRegistGroup(), registryParam.getRegistryKey(), registryParam.getRegistryValue(), new Date());
138
         int ret = xxlJobRegistryDao.registryUpdate(registryParam.getRegistGroup(), registryParam.getRegistryKey(), registryParam.getRegistryValue(), new Date());
130
         if (ret < 1) {
139
         if (ret < 1) {
131
             xxlJobRegistryDao.registrySave(registryParam.getRegistGroup(), registryParam.getRegistryKey(), registryParam.getRegistryValue(), new Date());
140
             xxlJobRegistryDao.registrySave(registryParam.getRegistGroup(), registryParam.getRegistryKey(), registryParam.getRegistryValue(), new Date());
138
 
147
 
139
     @Override
148
     @Override
140
     public ReturnT<String> registryRemove(RegistryParam registryParam) {
149
     public ReturnT<String> registryRemove(RegistryParam registryParam) {
150
+
151
+        // valid
152
+        if (!StringUtils.hasText(registryParam.getRegistGroup())
153
+                || !StringUtils.hasText(registryParam.getRegistryKey())
154
+                || !StringUtils.hasText(registryParam.getRegistryValue())) {
155
+            return new ReturnT<String>(ReturnT.FAIL_CODE, "Illegal Argument.");
156
+        }
157
+
141
         int ret = xxlJobRegistryDao.registryDelete(registryParam.getRegistGroup(), registryParam.getRegistryKey(), registryParam.getRegistryValue());
158
         int ret = xxlJobRegistryDao.registryDelete(registryParam.getRegistGroup(), registryParam.getRegistryKey(), registryParam.getRegistryValue());
142
         if (ret > 0) {
159
         if (ret > 0) {
143
 
160
 

+ 29 - 43
xxl-job-admin/src/test/java/com/xxl/job/adminbiz/AdminBizTest.java ファイルの表示

1
 package com.xxl.job.adminbiz;
1
 package com.xxl.job.adminbiz;
2
 
2
 
3
 import com.xxl.job.core.biz.AdminBiz;
3
 import com.xxl.job.core.biz.AdminBiz;
4
+import com.xxl.job.core.biz.client.AdminBizClient;
5
+import com.xxl.job.core.biz.model.HandleCallbackParam;
4
 import com.xxl.job.core.biz.model.RegistryParam;
6
 import com.xxl.job.core.biz.model.RegistryParam;
5
 import com.xxl.job.core.biz.model.ReturnT;
7
 import com.xxl.job.core.biz.model.ReturnT;
6
 import com.xxl.job.core.enums.RegistryConfig;
8
 import com.xxl.job.core.enums.RegistryConfig;
7
-import com.xxl.rpc.remoting.invoker.XxlRpcInvokerFactory;
8
-import com.xxl.rpc.remoting.invoker.call.CallType;
9
-import com.xxl.rpc.remoting.invoker.reference.XxlRpcReferenceBean;
10
-import com.xxl.rpc.remoting.invoker.route.LoadBalance;
11
-import com.xxl.rpc.remoting.net.NetEnum;
12
-import com.xxl.rpc.serialize.Serializer;
13
 import org.junit.Assert;
9
 import org.junit.Assert;
14
 import org.junit.Test;
10
 import org.junit.Test;
15
 
11
 
12
+import java.util.Arrays;
13
+import java.util.List;
14
+
16
 /**
15
 /**
17
  * admin api test
16
  * admin api test
18
  *
17
  *
21
 public class AdminBizTest {
20
 public class AdminBizTest {
22
 
21
 
23
     // admin-client
22
     // admin-client
24
-    private static String addressUrl = "http://127.0.0.1:8080/xxl-job-admin".concat(AdminBiz.MAPPING);
23
+    private static String addressUrl = "http://127.0.0.1:8080/xxl-job-admin/";
25
     private static String accessToken = null;
24
     private static String accessToken = null;
26
 
25
 
26
+
27
+    @Test
28
+    public void callback() throws Exception {
29
+        AdminBiz adminBiz = new AdminBizClient(addressUrl, accessToken);
30
+
31
+        HandleCallbackParam param = new HandleCallbackParam();
32
+        param.setLogId(1);
33
+        param.setExecuteResult(ReturnT.SUCCESS);
34
+
35
+        List<HandleCallbackParam> callbackParamList = Arrays.asList(param);
36
+
37
+        ReturnT<String> returnT = adminBiz.callback(callbackParamList);
38
+
39
+        Assert.assertTrue(returnT.getCode() == ReturnT.SUCCESS_CODE);
40
+    }
41
+
27
     /**
42
     /**
28
      * registry executor
43
      * registry executor
29
      *
44
      *
30
      * @throws Exception
45
      * @throws Exception
31
      */
46
      */
32
     @Test
47
     @Test
33
-    public void registryTest() throws Exception {
34
-        addressUrl = addressUrl.replace("http://", "");
35
-        AdminBiz adminBiz = (AdminBiz) new XxlRpcReferenceBean(
36
-                NetEnum.NETTY_HTTP,
37
-                Serializer.SerializeEnum.HESSIAN.getSerializer(),
38
-                CallType.SYNC,
39
-                LoadBalance.ROUND,
40
-                AdminBiz.class,
41
-                null,
42
-                3000,
43
-                addressUrl,
44
-                accessToken,
45
-                null,
46
-                null).getObject();
47
-
48
-        // test executor registry
48
+    public void registry() throws Exception {
49
+        AdminBiz adminBiz = new AdminBizClient(addressUrl, accessToken);
50
+
49
         RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), "xxl-job-executor-example", "127.0.0.1:9999");
51
         RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), "xxl-job-executor-example", "127.0.0.1:9999");
50
         ReturnT<String> returnT = adminBiz.registry(registryParam);
52
         ReturnT<String> returnT = adminBiz.registry(registryParam);
51
-        Assert.assertTrue(returnT.getCode() == ReturnT.SUCCESS_CODE);
52
 
53
 
53
-        // stop invoker
54
-        XxlRpcInvokerFactory.getInstance().stop();
54
+        Assert.assertTrue(returnT.getCode() == ReturnT.SUCCESS_CODE);
55
     }
55
     }
56
 
56
 
57
     /**
57
     /**
61
      */
61
      */
62
     @Test
62
     @Test
63
     public void registryRemove() throws Exception {
63
     public void registryRemove() throws Exception {
64
-        addressUrl = addressUrl.replace("http://", "");
65
-        AdminBiz adminBiz = (AdminBiz) new XxlRpcReferenceBean(
66
-                NetEnum.NETTY_HTTP,
67
-                Serializer.SerializeEnum.HESSIAN.getSerializer(),
68
-                CallType.SYNC,
69
-                LoadBalance.ROUND,
70
-                AdminBiz.class,
71
-                null,
72
-                3000,
73
-                addressUrl,
74
-                accessToken,
75
-                null,
76
-                null).getObject();
77
-
78
-        // test executor registry remove
64
+        AdminBiz adminBiz = new AdminBizClient(addressUrl, accessToken);
65
+
79
         RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), "xxl-job-executor-example", "127.0.0.1:9999");
66
         RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), "xxl-job-executor-example", "127.0.0.1:9999");
80
         ReturnT<String> returnT = adminBiz.registryRemove(registryParam);
67
         ReturnT<String> returnT = adminBiz.registryRemove(registryParam);
68
+
81
         Assert.assertTrue(returnT.getCode() == ReturnT.SUCCESS_CODE);
69
         Assert.assertTrue(returnT.getCode() == ReturnT.SUCCESS_CODE);
82
 
70
 
83
-        // stop invoker
84
-        XxlRpcInvokerFactory.getInstance().stop();
85
     }
71
     }
86
 
72
 
87
 }
73
 }

+ 17 - 14
xxl-job-admin/src/test/java/com/xxl/job/executor/ExecutorBizTest.java ファイルの表示

9
 import com.xxl.rpc.remoting.invoker.call.CallType;
9
 import com.xxl.rpc.remoting.invoker.call.CallType;
10
 import com.xxl.rpc.remoting.invoker.reference.XxlRpcReferenceBean;
10
 import com.xxl.rpc.remoting.invoker.reference.XxlRpcReferenceBean;
11
 import com.xxl.rpc.remoting.invoker.route.LoadBalance;
11
 import com.xxl.rpc.remoting.invoker.route.LoadBalance;
12
-import com.xxl.rpc.remoting.net.NetEnum;
13
-import com.xxl.rpc.serialize.Serializer;
12
+import com.xxl.rpc.remoting.net.impl.netty_http.client.NettyHttpClient;
13
+import com.xxl.rpc.serialize.impl.HessianSerializer;
14
 
14
 
15
 /**
15
 /**
16
  * executor-api client, test
16
  * executor-api client, test
49
 
49
 
50
         // do remote trigger
50
         // do remote trigger
51
         String accessToken = null;
51
         String accessToken = null;
52
-        ExecutorBiz executorBiz = (ExecutorBiz) new XxlRpcReferenceBean(
53
-                NetEnum.NETTY_HTTP,
54
-                Serializer.SerializeEnum.HESSIAN.getSerializer(),
55
-                CallType.SYNC,
56
-                LoadBalance.ROUND,
57
-                ExecutorBiz.class,
58
-                null,
59
-                3000,
60
-                "127.0.0.1:9999",
61
-                null,
62
-                null,
63
-                null).getObject();
52
+
53
+        XxlRpcReferenceBean referenceBean = new XxlRpcReferenceBean();
54
+        referenceBean.setClient(NettyHttpClient.class);
55
+        referenceBean.setSerializer(HessianSerializer.class);
56
+        referenceBean.setCallType(CallType.SYNC);
57
+        referenceBean.setLoadBalance(LoadBalance.ROUND);
58
+        referenceBean.setIface(ExecutorBiz.class);
59
+        referenceBean.setVersion(null);
60
+        referenceBean.setTimeout(3000);
61
+        referenceBean.setAddress("127.0.0.1:9999");
62
+        referenceBean.setAccessToken(null);
63
+        referenceBean.setInvokeCallback(null);
64
+        referenceBean.setInvokerFactory(null);
65
+
66
+        ExecutorBiz executorBiz = (ExecutorBiz) referenceBean.getObject();
64
 
67
 
65
         ReturnT<String> runResult = executorBiz.run(triggerParam);
68
         ReturnT<String> runResult = executorBiz.run(triggerParam);
66
 
69
 

+ 0 - 2
xxl-job-core/src/main/java/com/xxl/job/core/biz/AdminBiz.java ファイルの表示

11
  */
11
  */
12
 public interface AdminBiz {
12
 public interface AdminBiz {
13
 
13
 
14
-    public static final String MAPPING = "/api";
15
-
16
 
14
 
17
     // ---------------------- callback ----------------------
15
     // ---------------------- callback ----------------------
18
 
16
 

+ 48 - 0
xxl-job-core/src/main/java/com/xxl/job/core/biz/client/AdminBizClient.java ファイルの表示

1
+package com.xxl.job.core.biz.client;
2
+
3
+import com.xxl.job.core.biz.AdminBiz;
4
+import com.xxl.job.core.biz.model.HandleCallbackParam;
5
+import com.xxl.job.core.biz.model.RegistryParam;
6
+import com.xxl.job.core.biz.model.ReturnT;
7
+import com.xxl.job.core.util.XxlJobRemotingUtil;
8
+
9
+import java.util.List;
10
+
11
+/**
12
+ * admin api test
13
+ *
14
+ * @author xuxueli 2017-07-28 22:14:52
15
+ */
16
+public class AdminBizClient implements AdminBiz {
17
+
18
+    public AdminBizClient() {
19
+    }
20
+    public AdminBizClient(String addressUrl, String accessToken) {
21
+        this.addressUrl = addressUrl;
22
+        this.accessToken = accessToken;
23
+
24
+        // valid
25
+        if (!this.addressUrl.endsWith("/")) {
26
+            this.addressUrl = this.addressUrl + "/";
27
+        }
28
+    }
29
+
30
+    private String addressUrl ;
31
+    private String accessToken;
32
+
33
+
34
+    @Override
35
+    public ReturnT<String> callback(List<HandleCallbackParam> callbackParamList) {
36
+        return XxlJobRemotingUtil.postBody(addressUrl+"api/callback", accessToken, callbackParamList, 3);
37
+    }
38
+
39
+    @Override
40
+    public ReturnT<String> registry(RegistryParam registryParam) {
41
+        return XxlJobRemotingUtil.postBody(addressUrl + "api/registry", accessToken, registryParam, 3);
42
+    }
43
+
44
+    @Override
45
+    public ReturnT<String> registryRemove(RegistryParam registryParam) {
46
+        return XxlJobRemotingUtil.postBody(addressUrl + "api/registryRemove", accessToken, registryParam, 3);
47
+    }
48
+}

+ 15 - 32
xxl-job-core/src/main/java/com/xxl/job/core/executor/XxlJobExecutor.java ファイルの表示

2
 
2
 
3
 import com.xxl.job.core.biz.AdminBiz;
3
 import com.xxl.job.core.biz.AdminBiz;
4
 import com.xxl.job.core.biz.ExecutorBiz;
4
 import com.xxl.job.core.biz.ExecutorBiz;
5
+import com.xxl.job.core.biz.client.AdminBizClient;
5
 import com.xxl.job.core.biz.impl.ExecutorBizImpl;
6
 import com.xxl.job.core.biz.impl.ExecutorBizImpl;
6
 import com.xxl.job.core.handler.IJobHandler;
7
 import com.xxl.job.core.handler.IJobHandler;
7
 import com.xxl.job.core.log.XxlJobFileAppender;
8
 import com.xxl.job.core.log.XxlJobFileAppender;
11
 import com.xxl.job.core.thread.TriggerCallbackThread;
12
 import com.xxl.job.core.thread.TriggerCallbackThread;
12
 import com.xxl.rpc.registry.ServiceRegistry;
13
 import com.xxl.rpc.registry.ServiceRegistry;
13
 import com.xxl.rpc.remoting.invoker.XxlRpcInvokerFactory;
14
 import com.xxl.rpc.remoting.invoker.XxlRpcInvokerFactory;
14
-import com.xxl.rpc.remoting.invoker.call.CallType;
15
-import com.xxl.rpc.remoting.invoker.reference.XxlRpcReferenceBean;
16
-import com.xxl.rpc.remoting.invoker.route.LoadBalance;
17
-import com.xxl.rpc.remoting.net.NetEnum;
15
+import com.xxl.rpc.remoting.net.impl.netty_http.server.NettyHttpServer;
18
 import com.xxl.rpc.remoting.provider.XxlRpcProviderFactory;
16
 import com.xxl.rpc.remoting.provider.XxlRpcProviderFactory;
19
 import com.xxl.rpc.serialize.Serializer;
17
 import com.xxl.rpc.serialize.Serializer;
18
+import com.xxl.rpc.serialize.impl.HessianSerializer;
20
 import com.xxl.rpc.util.IpUtil;
19
 import com.xxl.rpc.util.IpUtil;
21
 import com.xxl.rpc.util.NetUtil;
20
 import com.xxl.rpc.util.NetUtil;
22
 import org.slf4j.Logger;
21
 import org.slf4j.Logger;
105
         // destory TriggerCallbackThread
104
         // destory TriggerCallbackThread
106
         TriggerCallbackThread.getInstance().toStop();
105
         TriggerCallbackThread.getInstance().toStop();
107
 
106
 
108
-        // destory invoker
109
-        stopInvokerFactory();
110
     }
107
     }
111
 
108
 
112
 
109
 
113
     // ---------------------- admin-client (rpc invoker) ----------------------
110
     // ---------------------- admin-client (rpc invoker) ----------------------
114
     private static List<AdminBiz> adminBizList;
111
     private static List<AdminBiz> adminBizList;
115
-    private static Serializer serializer;
112
+    private static Serializer serializer = new HessianSerializer();
116
     private void initAdminBizList(String adminAddresses, String accessToken) throws Exception {
113
     private void initAdminBizList(String adminAddresses, String accessToken) throws Exception {
117
-        serializer = Serializer.SerializeEnum.HESSIAN.getSerializer();
118
         if (adminAddresses!=null && adminAddresses.trim().length()>0) {
114
         if (adminAddresses!=null && adminAddresses.trim().length()>0) {
119
             for (String address: adminAddresses.trim().split(",")) {
115
             for (String address: adminAddresses.trim().split(",")) {
120
                 if (address!=null && address.trim().length()>0) {
116
                 if (address!=null && address.trim().length()>0) {
121
 
117
 
122
-                    String addressUrl = address.concat(AdminBiz.MAPPING);
123
-
124
-                    AdminBiz adminBiz = (AdminBiz) new XxlRpcReferenceBean(
125
-                            NetEnum.NETTY_HTTP,
126
-                            serializer,
127
-                            CallType.SYNC,
128
-                            LoadBalance.ROUND,
129
-                            AdminBiz.class,
130
-                            null,
131
-                            3000,
132
-                            addressUrl,
133
-                            accessToken,
134
-                            null,
135
-                            null
136
-                    ).getObject();
118
+                    AdminBiz adminBiz = new AdminBizClient(address.trim(), accessToken);
137
 
119
 
138
                     if (adminBizList == null) {
120
                     if (adminBizList == null) {
139
                         adminBizList = new ArrayList<AdminBiz>();
121
                         adminBizList = new ArrayList<AdminBiz>();
143
             }
125
             }
144
         }
126
         }
145
     }
127
     }
146
-    private void stopInvokerFactory(){
147
-        // stop invoker factory
148
-        try {
149
-            XxlRpcInvokerFactory.getInstance().stop();
150
-        } catch (Exception e) {
151
-            logger.error(e.getMessage(), e);
152
-        }
153
-    }
154
     public static List<AdminBiz> getAdminBizList(){
128
     public static List<AdminBiz> getAdminBizList(){
155
         return adminBizList;
129
         return adminBizList;
156
     }
130
     }
171
         serviceRegistryParam.put("address", address);
145
         serviceRegistryParam.put("address", address);
172
 
146
 
173
         xxlRpcProviderFactory = new XxlRpcProviderFactory();
147
         xxlRpcProviderFactory = new XxlRpcProviderFactory();
174
-        xxlRpcProviderFactory.initConfig(NetEnum.NETTY_HTTP, Serializer.SerializeEnum.HESSIAN.getSerializer(), ip, port, accessToken, ExecutorServiceRegistry.class, serviceRegistryParam);
148
+
149
+        xxlRpcProviderFactory.setServer(NettyHttpServer.class);
150
+        xxlRpcProviderFactory.setSerializer(HessianSerializer.class);
151
+        xxlRpcProviderFactory.setCorePoolSize(30);
152
+        xxlRpcProviderFactory.setMaxPoolSize(200);
153
+        xxlRpcProviderFactory.setIp(ip);
154
+        xxlRpcProviderFactory.setPort(port);
155
+        xxlRpcProviderFactory.setAccessToken(accessToken);
156
+        xxlRpcProviderFactory.setServiceRegistry(ExecutorServiceRegistry.class);
157
+        xxlRpcProviderFactory.setServiceRegistryParam(serviceRegistryParam);
175
 
158
 
176
         // add services
159
         // add services
177
         xxlRpcProviderFactory.addService(ExecutorBiz.class.getName(), null, new ExecutorBizImpl());
160
         xxlRpcProviderFactory.addService(ExecutorBiz.class.getName(), null, new ExecutorBizImpl());

+ 122 - 0
xxl-job-core/src/main/java/com/xxl/job/core/util/XxlJobRemotingUtil.java ファイルの表示

1
+package com.xxl.job.core.util;
2
+
3
+import com.xxl.job.core.biz.model.ReturnT;
4
+import com.xxl.registry.client.util.json.BasicJson;
5
+import org.slf4j.Logger;
6
+import org.slf4j.LoggerFactory;
7
+
8
+import java.io.BufferedReader;
9
+import java.io.DataOutputStream;
10
+import java.io.InputStreamReader;
11
+import java.net.HttpURLConnection;
12
+import java.net.URL;
13
+import java.util.Map;
14
+
15
+/**
16
+ * @author xuxueli 2018-11-25 00:55:31
17
+ */
18
+public class XxlJobRemotingUtil {
19
+    private static Logger logger = LoggerFactory.getLogger(XxlJobRemotingUtil.class);
20
+    public static String XXL_RPC_ACCESS_TOKEN = "XXL_RPC_ACCESS_TOKEN";
21
+
22
+    /**
23
+     * post
24
+     *
25
+     * @param url
26
+     * @param accessToken
27
+     * @param requestObj
28
+     * @return
29
+     */
30
+    public static ReturnT<String> postBody(String url, String accessToken, Object requestObj, int timeout) {
31
+        HttpURLConnection connection = null;
32
+        BufferedReader bufferedReader = null;
33
+        try {
34
+            // connection
35
+            URL realUrl = new URL(url);
36
+            connection = (HttpURLConnection) realUrl.openConnection();
37
+
38
+            // connection setting
39
+            connection.setRequestMethod("POST");
40
+            connection.setDoOutput(true);
41
+            connection.setDoInput(true);
42
+            connection.setUseCaches(false);
43
+            connection.setReadTimeout(timeout * 1000);
44
+            connection.setConnectTimeout(3 * 1000);
45
+            connection.setRequestProperty("connection", "Keep-Alive");
46
+            connection.setRequestProperty("Content-Type", "application/json;charset=UTF-8");
47
+            connection.setRequestProperty("Accept-Charset", "application/json;charset=UTF-8");
48
+
49
+            if(accessToken!=null && accessToken.trim().length()>0){
50
+                connection.setRequestProperty(XXL_RPC_ACCESS_TOKEN, accessToken);
51
+            }
52
+
53
+            // do connection
54
+            connection.connect();
55
+
56
+            // write requestBody
57
+            String requestBody = BasicJson.toJson(requestObj);
58
+
59
+            DataOutputStream dataOutputStream = new DataOutputStream(connection.getOutputStream());
60
+            dataOutputStream.writeBytes(requestBody);
61
+            dataOutputStream.flush();
62
+            dataOutputStream.close();
63
+
64
+            /*byte[] requestBodyBytes = requestBody.getBytes("UTF-8");
65
+            connection.setRequestProperty("Content-Length", String.valueOf(requestBodyBytes.length));
66
+            OutputStream outwritestream = connection.getOutputStream();
67
+            outwritestream.write(requestBodyBytes);
68
+            outwritestream.flush();
69
+            outwritestream.close();*/
70
+
71
+            // valid StatusCode
72
+            int statusCode = connection.getResponseCode();
73
+            if (statusCode != 200) {
74
+                return new ReturnT<String>(ReturnT.FAIL_CODE, "xxl-rpc remoting fail, StatusCode("+ statusCode +") invalid. for url : " + url);
75
+            }
76
+
77
+            // result
78
+            bufferedReader = new BufferedReader(new InputStreamReader(connection.getInputStream()));
79
+            StringBuilder result = new StringBuilder();
80
+            String line;
81
+            while ((line = bufferedReader.readLine()) != null) {
82
+                result.append(line);
83
+            }
84
+            String resultJson = result.toString();
85
+
86
+            // parse returnT
87
+            try {
88
+                Map<String, Object> resultMap = BasicJson.parseMap(resultJson);
89
+
90
+                ReturnT<String> returnT = new ReturnT<String>();
91
+                if (resultMap==null) {
92
+                    returnT.setCode(ReturnT.FAIL_CODE);
93
+                    returnT.setMsg("AdminBizClient Remoting call fail.");
94
+                } else {
95
+                    returnT.setCode(Integer.valueOf(String.valueOf(resultMap.get("code"))));
96
+                    returnT.setMsg(String.valueOf(resultMap.get("msg")));
97
+                    returnT.setContent(String.valueOf(resultMap.get("content")));
98
+                }
99
+                return returnT;
100
+            } catch (Exception e) {
101
+                e.printStackTrace();
102
+                return new ReturnT<String>(ReturnT.FAIL_CODE, "xxl-rpc remoting response content invalid("+ resultJson +"), for url : " + url);
103
+            }
104
+
105
+        } catch (Exception e) {
106
+            logger.error(e.getMessage(), e);
107
+            return new ReturnT<String>(ReturnT.FAIL_CODE, "xxl-rpc remoting error("+ e.getMessage() +"), for url : " + url);
108
+        } finally {
109
+            try {
110
+                if (bufferedReader != null) {
111
+                    bufferedReader.close();
112
+                }
113
+                if (connection != null) {
114
+                    connection.disconnect();
115
+                }
116
+            } catch (Exception e2) {
117
+                logger.error(e2.getMessage(), e2);
118
+            }
119
+        }
120
+    }
121
+
122
+}

+ 16 - 14
xxl-job-core/src/test/java/com/xxl/job/core/biz/impl/ExecutorBizImplTest.java ファイルの表示

10
 import com.xxl.rpc.remoting.invoker.call.CallType;
10
 import com.xxl.rpc.remoting.invoker.call.CallType;
11
 import com.xxl.rpc.remoting.invoker.reference.XxlRpcReferenceBean;
11
 import com.xxl.rpc.remoting.invoker.reference.XxlRpcReferenceBean;
12
 import com.xxl.rpc.remoting.invoker.route.LoadBalance;
12
 import com.xxl.rpc.remoting.invoker.route.LoadBalance;
13
-import com.xxl.rpc.remoting.net.NetEnum;
14
-import com.xxl.rpc.serialize.Serializer;
13
+import com.xxl.rpc.remoting.net.impl.netty_http.client.NettyHttpClient;
14
+import com.xxl.rpc.serialize.impl.HessianSerializer;
15
 import org.junit.After;
15
 import org.junit.After;
16
 import org.junit.Assert;
16
 import org.junit.Assert;
17
 import org.junit.Before;
17
 import org.junit.Before;
44
         TimeUnit.SECONDS.sleep(3);
44
         TimeUnit.SECONDS.sleep(3);
45
 
45
 
46
         // init executor biz proxy
46
         // init executor biz proxy
47
-        executorBiz = (ExecutorBiz) new XxlRpcReferenceBean(
48
-                NetEnum.NETTY_HTTP,
49
-                Serializer.SerializeEnum.HESSIAN.getSerializer(),
50
-                CallType.SYNC,
51
-                LoadBalance.ROUND,
52
-                ExecutorBiz.class,
53
-                null,
54
-                3000,
55
-                "127.0.0.1:9999",
56
-                null,
57
-                null,
58
-                null).getObject();
47
+        XxlRpcReferenceBean referenceBean = new XxlRpcReferenceBean();
48
+        referenceBean.setClient(NettyHttpClient.class);
49
+        referenceBean.setSerializer(HessianSerializer.class);
50
+        referenceBean.setCallType(CallType.SYNC);
51
+        referenceBean.setLoadBalance(LoadBalance.ROUND);
52
+        referenceBean.setIface(ExecutorBiz.class);
53
+        referenceBean.setVersion(null);
54
+        referenceBean.setTimeout(3000);
55
+        referenceBean.setAddress("127.0.0.1:9999");
56
+        referenceBean.setAccessToken(null);
57
+        referenceBean.setInvokeCallback(null);
58
+        referenceBean.setInvokerFactory(null);
59
+
60
+        executorBiz = (ExecutorBiz) referenceBean.getObject();
59
     }
61
     }
60
 
62
 
61
     @After
63
     @After