Skip to content

B
byit-myth-job
Commit 682a0e40 by huangfusuper
Options

增加管理服务根据路由策略选取服务，并执行调度

parent 0942268b
Showing with 141 additions and 43 deletions
byit-myth-core/myth-admin-core/pom.xml
......@@ -19,15 +19,6 @@
<artifactId>netty-all</artifactId>
</dependency>
<dependency>
<groupId>myth-job</groupId>
<artifactId>myth-executor-core</artifactId>
</dependency>
<dependency>
<groupId>myth-job</groupId>
<artifactId>byit-myth-rpc</artifactId>
</dependency>
<dependency>
<groupId>myth-job</groupId>
......@@ -39,10 +30,7 @@
<artifactId>myth-core-common</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
</dependency>
</dependencies>
</project>
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaBeanJobTask.java
......@@ -3,9 +3,14 @@ package com.byit.task;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.byit.job.exceptions.plugin.PluginException;
import com.byit.job.model.JavaBeanJobInfo;
import com.byit.job.utils.IpUtil;
import com.byit.rpc.remoting.invoker.route.LoadBalance;
import com.byit.rpc.remoting.invoker.route.RpcLoadBalance;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import lombok.extern.slf4j.Slf4j;
/**
* @program: byit-myth-job->JvavBeanJobTask
......@@ -13,6 +18,7 @@ import io.netty.util.TimerTask;
* @author: huangfu
* @date: 2019/11/20 15:08
**/
@Slf4j
public class JavaBeanJobTask implements TimerTask {
private JavaBeanJobInfo javaBeanJobInfo;
......@@ -21,14 +27,25 @@ public class JavaBeanJobTask implements TimerTask {
}
@Override
public void run(Timeout timeout) throws Exception {
String url = javaBeanJobInfo.getUrl( );
String jobHandelName = javaBeanJobInfo.getJobHandelName( );
String param = javaBeanJobInfo.getParam( );
public void run(Timeout timeout) {
//根据负责均衡方案获取对应IP
RpcLoadBalance rpcInvokerRouter = LoadBalance.match(javaBeanJobInfo.getRoutingStrategy( ), LoadBalance.ROUND).rpcInvokerRouter;
try{
String url = IpUtil.electiveUrl(javaBeanJobInfo.getUrl( ), rpcInvokerRouter);
String jobHandelName = javaBeanJobInfo.getJobHandelName( );
String param = javaBeanJobInfo.getParam( );
JSONObject jsonObject = new JSONObject();
jsonObject.put("jobHandelName",jobHandelName);
jsonObject.put("param",param);
String result = HttpUtil.post(url, JSON.toJSONString(jsonObject),10*1000);
log.info("---------------{}------------",result);
}catch (PluginException ignored){
log.error("--------------------{}，{}-----------------",ignored.getIEnum().getCode(),ignored.getIEnum().getMsg());
}catch (Exception e){
e.printStackTrace();
}
JSONObject jsonObject = new JSONObject();
jsonObject.put("jobHandelName",jobHandelName);
jsonObject.put("param",param);
String result = HttpUtil.post(url, JSON.toJSONString(jsonObject),10*1000);
}
}
byit-myth-core/myth-core-common/pom.xml
......@@ -31,6 +31,16 @@
<artifactId>hutool-all</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
</dependency>
<dependency>
<groupId>myth-job</groupId>
<artifactId>byit-myth-rpc</artifactId>
</dependency>
</dependencies>
</project>
\ No newline at end of file
byit-myth-core/myth-core-common/src/main/java/com/byit/job/enums/plugin/PluginEnum.java
......@@ -8,8 +8,8 @@ import com.byit.job.enums.IEnum;
*/
public enum PluginEnum implements IEnum {
REQUEST_PORT_OR_IP_IS_MISSING("请求ip或者port为null",10000);
REQUEST_PORT_OR_IP_IS_MISSING("请求ip或者port为null",10000),
NO_SERVICE_AVAILABLE("无可用的服务",11000);
private String msg;
private int code;
......
byit-myth-core/myth-core-common/src/main/java/com/byit/job/model/JavaBeanJobInfo.java
......@@ -40,14 +40,20 @@ public class JavaBeanJobInfo {
*/
private String gatewayToken;
/**
* 添加任务的地址
* 添加任务的IP
*/
private String requestIP;
/**
* 添加任务的port
*/
private String requestPort;
/**
* 添加任务的参数
*/
private String param;
@Override
public String toString() {
return "JavaBeanJobInfo{" +
......
byit-myth-core/myth-core-common/src/main/java/com/byit/job/model/ReturnResult.java
......@@ -41,6 +41,7 @@ public class ReturnResult<T> implements Serializable {
*/
public ReturnResult(T content) {
this.code = JobResultEnum.SUCCESS.getCode();
this.msg = JobResultEnum.SUCCESS.getMsg();
this.content = content;
}
}
byit-myth-core/myth-core-common/src/main/java/com/byit/job/utils/IpUtil.java 0 → 100644
package com.byit.job.utils;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.byit.job.enums.plugin.PluginEnum;
import com.byit.job.exceptions.plugin.PluginException;
import com.byit.rpc.remoting.invoker.route.RpcLoadBalance;
import lombok.extern.slf4j.Slf4j;
import java.util.Arrays;
import java.util.Collections;
import java.util.TreeSet;
/**
* @program: byit-myth-job->IpUtil
* @description: IP相关工具类
* @author: huangfu
* @date: 2019/11/28 13:48
**/
@Slf4j
public class IpUtil {
/**
* 存活测试
* @param url 测试url
* @return return
*/
public static boolean survivalTest(String url){
try{
log.info("------------------开始发送心跳包--------------------");
JSONObject jsonObject = new JSONObject();
jsonObject.put("heartbeat","PENG");
String heartbeatRes = HttpUtil.post(url, JSON.toJSONString(jsonObject), 2 * 1000);
log.info("------------------接收心跳包--------------------");
return "PONG".equals(heartbeatRes);
}catch (Exception e){
log.error("--------------------{},服务不可用------------------",url);
return false;
}
}
/**
* 根据路由策略选择地址
* @param urlsStr
* @param rpcLoadBalance
* @return
*/
public static String electiveUrl(String urlsStr, RpcLoadBalance rpcLoadBalance) throws PluginException {
log.info("-----------------开始根据路由规则选取服务地址--------------------");
String[] urls = urlsStr.split(",");
TreeSet<String> treeSet = new TreeSet<>(Arrays.asList(urls));
boolean flag = false;
String electiveIp = "";
while (!flag){
if(CollectionUtil.isEmpty(treeSet)) {
throw new PluginException(PluginEnum.NO_SERVICE_AVAILABLE);
}
electiveIp = rpcLoadBalance.route("PLUGIN-SERVER", treeSet);
//删除这个IP
treeSet.remove(electiveIp);
flag = survivalTest(electiveIp);
}
return electiveIp;
}
}
byit-myth-executor/myth-exector-plugin/src/main/java/com/byit/rpc/RunJobServerHandler.java
......@@ -12,6 +12,7 @@ import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.*;
import io.netty.util.CharsetUtil;
import io.netty.util.internal.StringUtil;
import lombok.extern.slf4j.Slf4j;
/**
* @program: byit-myth-job->RunJobServer
......@@ -19,28 +20,38 @@ import io.netty.util.internal.StringUtil;
* @author: huangfu
* @date: 2019/11/18 16:33
**/
@Slf4j
public class RunJobServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
private static final String PENG = "PENG";
private static final String PONG = "PONG";
@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {
String responseBody = "";
if(req instanceof HttpRequest){
JSONObject jsonObject = analysisParam(req.content( ));
if(null == jsonObject){
throw new Exception("核心参数 为 null");
}
String jobHandelName = ((String)(jsonObject.get("jobHandelName")));
String param = ((String)(jsonObject.get("param")));
if(StringUtil.isNullOrEmpty(jobHandelName)){
throw new Exception("jobHandelName 为 null");
//检测是否有心跳参数，有心跳参数则为测试参数，且为PENG的话，服务端回复 PONG
String heartbeat = (String)(jsonObject.get("heartbeat"));
if(null == heartbeat){
String jobHandelName = ((String)(jsonObject.get("jobHandelName")));
String param = ((String)(jsonObject.get("param")));
if(StringUtil.isNullOrEmpty(jobHandelName)){
throw new Exception("jobHandelName 为 null");
}
ReturnResult<String> stringReturnResult = runJob(jobHandelName, param);
assert stringReturnResult != null;
responseBody = stringReturnResult.getMsg();
log.info("---------服务器端:{},{}-------------",stringReturnResult.getCode(),stringReturnResult.getMsg());
}else if(PENG.equals(heartbeat)){
log.info("----------调度平台心跳检测-------------");
responseBody = PONG;
}
ReturnResult<String> stringReturnResult = runJob(jobHandelName, param);
System.out.println("---------服务器端-------------"+stringReturnResult.getCode());
//----------------------------------消息发送-----------------------------------
ByteBuf byteBuf = Unpooled.copiedBuffer("Hello World", CharsetUtil.UTF_8);
ByteBuf byteBuf = Unpooled.copiedBuffer(responseBody, CharsetUtil.UTF_8);
//HTTP响应
FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,byteBuf);
//设置头信息
......@@ -58,10 +69,6 @@ public class RunJobServerHandler extends SimpleChannelInboundHandler<FullHttpReq
try {
IJobHandler iJobHandler = jobClass.newInstance( );
return iJobHandler.execute(param);
} catch (InstantiationException e) {
e.printStackTrace( );
} catch (IllegalAccessException e) {
e.printStackTrace( );
} catch (Exception e) {
e.printStackTrace( );
}
......
byit-myth-gateway/pom.xml
......@@ -41,7 +41,6 @@
<dependency>
<groupId>myth-job</groupId>
<artifactId>byit-myth-rpc</artifactId>
<version>${project.version}</version>
</dependency>
......
byit-myth-rpc/pom.xml
......@@ -106,7 +106,6 @@
<scope>provided</scope>
</dependency>
<!-- ********************** registry (default=xxl-registry) ********************** -->
<!-- registry-client -->
......
demo-client/byit-demo-client/src/main/java/com/byit/job/Mains.java
......@@ -15,7 +15,7 @@ import java.io.IOException;
**/
public class Mains {
public static void main(String[] args) throws IOException {
String plServerUrl = "http://localhost:9999";
String plServerUrl = "http://127.0.0.1:9999,http://127.0.0.1:8888";
String mythCron = "时间";
String routingStrategy = "路由策略";
String blockingStrategy = "阻塞策略";
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment