手撸XXL-JOB（四）——远程调用定时任务

Java Socket网络编程

网络编程是Java编程中的重要组成部分，包括服务端和客户端两部分内容。Socket是Java网络编程的基本组件之一，用于在应用程序之间提供双向通信，Socket提供了一种标准的接口，允许应用程序通过网络发送和接收数据，在Java中，Socket可以分为客户端Socket和服务端Socket两种类型。
客户端Socket：客户端 Socket 用于与服务端 Socket 进行通信。客户端 Socket 通过指定服务端的 IP 地址和端口号，连接到服务端 Socket，然后发送数据到服务端 Socket。
服务端Socket：服务端 Socket 用于接收来自客户端 Socket 的连接请求，并在连接成功后，与客户端 Socket 进行通信。服务端 Socket 首先需要创建一个 ServerSocket 对象，并通过 bind 方法绑定到一个本地端口，然后等待客户端 Socket 的连接请求。
下面是Socket的一个示例：
服务端：

package org.example.demo1;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;

public class Server {
    public static void main(String[] args) {
        try {
            ServerSocket serverSocket = new ServerSocket(8000);
            System.out.println("Server started, waiting for client...");

            Socket socket = serverSocket.accept();
            System.out.println("Client connected.");

            BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            PrintWriter out = new PrintWriter(socket.getOutputStream(), true);

            String message;
            while ((message = in.readLine()) != null) {
                System.out.println("Client:" + message);
                out.println("Server received message:" + message);
            }

            in.close();
            out.close();
            socket.close();
            serverSocket.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

客户端：

package org.example.demo1;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;

public class Client {
    public static void main(String[] args) {
        try {
            Socket socket = new Socket("localhost", 8000);
            System.out.println("Connected to server.");

            BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            PrintWriter out = new PrintWriter(socket.getOutputStream(), true);

            BufferedReader consoleIn = new BufferedReader(new InputStreamReader(System.in));
            String message;
            while ((message = consoleIn.readLine()) != null) {
                out.println(message);
                System.out.println("Server:" + in.readLine());
            }
            consoleIn.close();
            in.close();
            out.close();
            socket.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

首先启动服务端，然后启动客户端，在客户端的控制台，输入数据，服务端能接收到数据并返回对应的响应。
image.png
image.png

远程调用定时任务

首先，我们创建两个模块，core模块包含yang-job的一些核心内容，比如IJobExecutor执行器、JobExecuteRequest执行器请求等；client模块依赖core模块，并封装和socket客户端调用相关的一些内容。
然后创建一个sample1模块，用于演示。
image.png

core模块

image.png
core目前定义了定时任务执行类和其入参、出参等信息，其中，YangJobTransferDTO包含任务类路径和任务请求，如下所示：

package com.yang.job.dto;

import com.yang.job.execute.YangJobExecuteRequest;

import java.io.Serializable;

public class YangJobTransferDTO implements Serializable {
    private String className;

    private YangJobExecuteRequest yangJobExecuteRequest;

    public String getClassName() {
        return className;
    }

    public void setClassName(String className) {
        this.className = className;
    }

    public YangJobExecuteRequest getYangJobExecuteRequest() {
        return yangJobExecuteRequest;
    }

    public void setYangJobExecuteRequest(YangJobExecuteRequest yangJobExecuteRequest) {
        this.yangJobExecuteRequest = yangJobExecuteRequest;
    }
}
client模块

image.png
client模块定义了客户端所需要的一些类，其中，YangJob为注解类，对于每一个定时任务，需要加上YangJob注解，才能被正确调用。

package com.yang.job.client.annotations;

import java.lang.annotation.*;

@Documented
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface YangJob {
}

YangJobClientProperty为配置信息类，目前需要两个配置信息，客户端socket的ip和端口号

package com.yang.job.client.configuration;


import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
public class YangJobClientProperty {
    @Value("${yang-job.executor.port}")
    private Integer port;

    @Value("${yang-job.executor.ip}")
    private String ip;


    public Integer getPort() {
        return port;
    }

    public void setPort(Integer port) {
        this.port = port;
    }

    public String getIp() {
        return ip;
    }

    public void setIp(String ip) {
        this.ip = ip;
    }
}

YangJobClientPostProcessor在SpringBoot加载完毕后，扫描bean，将实现IYongJobExecutor的bean，注册到YangJobClientManager的map中，方便后续调用

package com.yang.job.client.schema;


import com.yang.job.client.annotations.YangJob;
import com.yang.job.client.YangJobClientManager;
import com.yang.job.execute.IYangJobExecutor;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;

public class YangJobClientPostProcessor implements BeanPostProcessor {
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        if (!(bean instanceof IYangJobExecutor)) {
            return bean;
        }
        YangJob annotation = bean.getClass().getAnnotation(YangJob.class);
        if (annotation == null) {
            return bean;
        }
        YangJobClientManager.putJobExecutor(bean.getClass().getName(), (IYangJobExecutor) bean);
        return bean;
    }
}

YangJobClientManager负责监听端口和管理定时任务的执行，它会监听我们配置的yang-job.execute.port端口号，然后当接收到消息时，将消息转为入参，并取出对应的定时任务执行类，执行对应的代码。

package com.yang.job.client;


import com.alibaba.fastjson.JSONObject;
import com.yang.job.client.dto.YangJobClientPropertyDTO;
import com.yang.job.core.dto.YangJobTransferDTO;
import com.yang.job.core.dto.ResultT;
import com.yang.job.core.execute.IYangJobExecutor;
import com.yang.job.core.execute.YangJobExecuteRequest;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class YangJobClientManager {
    private static Map<String, IYangJobExecutor> className2JobExecutorMap = new ConcurrentHashMap<>();

    private YangJobClientPropertyDTO yangJobClientPropertyDTO;

    private ServerSocket serverSocket;

    public YangJobClientManager(YangJobClientPropertyDTO yangJobClientPropertyDTO) {
        this.yangJobClientPropertyDTO = yangJobClientPropertyDTO;
    }

    public void init() {
        Integer port = this.yangJobClientPropertyDTO.getPort();
        try {
            this.serverSocket = new ServerSocket(port);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        System.out.println("init success============");
        new Thread(() -> {
            while (true) {
                try {
                    Socket socket = serverSocket.accept();
                    BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                    PrintWriter printWriter = new PrintWriter(socket.getOutputStream(), true);
                    String params = bufferedReader.readLine();
                    YangJobTransferDTO yangJobTransferDTO = JSONObject.parseObject(params, YangJobTransferDTO.class);
                    System.out.println(yangJobTransferDTO);
                    String className = yangJobTransferDTO.getClassName();
                    YangJobExecuteRequest yangJobExecuteRequest = yangJobTransferDTO.getYangJobExecuteRequest();
                    IYangJobExecutor jobExecutor = getJobExecutor(className);
                    if (jobExecutor != null) {
                        ResultT response = jobExecutor.execute(yangJobExecuteRequest);
                        printWriter.println(JSONObject.toJSONString(response));
                    }
                    bufferedReader.close();
                    printWriter.close();
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                if (serverSocket.isClosed() || serverSocket == null) {
                    break;
                }
            }
        }).start();
    }

    public void shutdown() {
        if (this.serverSocket != null) {
            try {
                this.serverSocket.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public YangJobClientPropertyDTO getYangJobPropertyDTO() {
        return this.yangJobClientPropertyDTO;
    }

    public static void putJobExecutor(String className, IYangJobExecutor iJobExecutor) {
        className2JobExecutorMap.put(className, iJobExecutor);
    }

    public static IYangJobExecutor getJobExecutor(String className) {
        return className2JobExecutorMap.get(className);
    }

}

YangJobClientContext为客户端的上下文，负责监听SpringBoot刷新消息和关闭消息，并执行对应的操作。

package com.yang.job.client;


import com.yang.job.client.utils.SpringContextUtils;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ApplicationContextEvent;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.event.ContextRefreshedEvent;

public class YangJobClientContext implements ApplicationListener<ApplicationContextEvent> {
    private static YangJobClientContext instance;

    private ApplicationContext applicationContext;

    @Override
    public void onApplicationEvent(ApplicationContextEvent event) {
        if (event instanceof ContextRefreshedEvent) {
            System.out.println("刷新了=========");
            YangJobClientContext.instance = this;
            instance.applicationContext = applicationContext;
            init();
        } else if (event instanceof ContextClosedEvent) {
            System.out.println("销毁了=========");
            shutdown();
        }
    }

    private void init() {
        YangJobClientManager yangJobClientManager = SpringContextUtils.getBeanOfType(YangJobClientManager.class);
        yangJobClientManager.init();
    }

    private void shutdown() {
        YangJobClientManager yangJobClientManager = SpringContextUtils.getBeanOfType(YangJobClientManager.class);
        yangJobClientManager.shutdown();
    }
}

YangJobClientConfiguration为配置类，负责对YangJobClientPostProcessor、YangJobClientManager和YangJobClientContext的统一配置管理。

package com.yang.job.client.configuration;


import com.yang.job.client.YangJobClientManager;
import com.yang.job.client.YangJobClientContext;
import com.yang.job.client.dto.YangJobClientPropertyDTO;
import com.yang.job.client.schema.YangJobClientPostProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class YangJobClientConfiguration {
    @Autowired
    private YangJobClientProperty yangJobClientProperty;

    @Bean
    public YangJobClientPostProcessor yangJobPostProcessor() {
        return new YangJobClientPostProcessor();
    }

    @Bean
    public YangJobClientManager yangJobClientManager() {
        YangJobClientPropertyDTO yangJobClientPropertyDTO = new YangJobClientPropertyDTO();
        yangJobClientPropertyDTO.setIp(yangJobClientProperty.getIp());
        yangJobClientPropertyDTO.setPort(yangJobClientProperty.getPort());
        return new YangJobClientManager(yangJobClientPropertyDTO);
    }

    @Bean
    public YangJobClientContext yangJobContext() {
        return new YangJobClientContext();
    }
}

最后，为了使引入client依赖的应用，能自动装配我们提供的bean，我们在resources目录下创建META-INF目录，在该目录下创建spring.factories文件，文件内容如下：

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
  com.yang.job.client.utils.SpringContextUtils,\
  com.yang.job.client.configuration.YangJobClientProperty,\
  com.yang.job.client.configuration.YangJobClientConfiguration
sample1

我们创建一个sample1项目，引入spring-boot-starter-web依赖和yang-client，yang-core的依赖

  <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>com.yang</groupId>
            <artifactId>yang-job-core</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
        <dependency>
            <groupId>com.yang</groupId>
            <artifactId>yang-job-client</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
    </dependencies>

创建启动类

package com.yang.job.sample1;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class YangJobSample1App {
    public static void main(String[] args) {
        SpringApplication.run(YangJobSample1App.class, args);
    }
}

创建一个任务类：

package com.yang.job.sample1.task;

import com.yang.job.client.annotations.YangJob;
import com.yang.job.dto.ResultT;
import com.yang.job.execute.IYangJobExecutor;
import com.yang.job.execute.YangJobExecuteRequest;
import org.springframework.stereotype.Component;

@Component
@YangJob
public class TestTask1 implements IYangJobExecutor {
    @Override
    public ResultT execute(YangJobExecuteRequest yangJobExecuteRequest) {
        System.out.println("开启定时任务了，入参为:" + yangJobExecuteRequest);
        return ResultT.success();
    }
}

添加配置文件，因为client模块的YangJobClientProperty需要有yang-job.executor.port和yang-job.executor.ip这两个配置，如果我们的配置文件中，缺少这些配置，会导致报错，无法启动项目。

spring:
  application:
    name: YangJobSample1App
yang-job:
  executor:
    port: 9999
    ip: 127.0.0.1
server:
  port: 8001
测试

我们先启动刚才的sample1项目，然后执行下列代码，来远程调用TestTask1方法执行类。

 public static void main(String[] args) {
        try {
            Socket socket = new Socket("127.0.0.1", 9999);
            System.out.println("链接成功=============");
            PrintWriter printWriter = new PrintWriter(socket.getOutputStream(), true);
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            YangJobExecuteRequest yangJobExecuteRequest = new YangJobExecuteRequest();
            yangJobExecuteRequest.setJobId("1");
            yangJobExecuteRequest.addParam("num", "1");
            YangJobTransferDTO yangJobTransferDTO = new YangJobTransferDTO();
            yangJobTransferDTO.setClassName("com.yang.job.sample1.task.TestTask1");
            yangJobTransferDTO.setYangJobExecuteRequest(yangJobExecuteRequest);

            printWriter.println(JSONObject.toJSONString(yangJobTransferDTO));
            System.out.println("response:" + bufferedReader.readLine());
            bufferedReader.close();
            printWriter.close();
            socket.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

执行结果如下，说明我们能成功地进行远程调用。
image.png
image.png

添加远程任务

domain层

在上一篇文章中，我们操作的任务，都是本地任务，现在我们需要对远程任务进行操作，为了区分任务类型，我们首先在domain层添加一个任务类型枚举

package com.yang.job.admin.domain.enums;

public enum JobTypeEnum {
    LOCAL("local", "本地任务"),
    REMOTE("remote", "远程任务");

    private String name;

    private String description;

    JobTypeEnum(String name, String description) {
        this.name = name;
        this.description = description;
    }

    public String getName() {
        return name;
    }

    public String getDescription() {
        return description;
    }

    public static JobTypeEnum getJobTypeByName(String name) {
        for (JobTypeEnum value : values()) {
            if (value.getName().equals(name)) {
                return value;
            }
        }
        return null;
    }
}

然后修改YangJobModel，添加上任务类型枚举和远程任务信息

package com.yang.job.admin.domain.model;


import com.yang.job.admin.client.dto.common.BusinessException;
import com.yang.job.admin.client.dto.common.ErrorCode;
import com.yang.job.admin.domain.enums.JobExecuteStrategyEnum;
import com.yang.job.admin.domain.enums.JobTypeEnum;
import com.yang.job.admin.domain.event.SaveJobPostEvent;
import com.yang.job.admin.domain.event.SubmitJobPostEvent;
import com.yang.job.admin.domain.event.UpdateJobPostEvent;
import com.yang.job.admin.domain.valueobject.RemoteExecutorMessage;
import com.yang.job.admin.infra.event.EventCenter;
import com.yang.job.admin.infra.job.YangJobManager;
import com.yang.job.admin.infra.job.request.YangJobSubmitParam;
import com.yang.job.admin.infra.utils.CronUtils;
import com.yang.job.admin.infra.utils.SpringContextUtils;
import lombok.Data;

import java.io.Serializable;
import java.time.ZonedDateTime;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@Data
public class YangJobModel implements Serializable {
    private Integer jobId;

    private String jobName;

    private String description;

    private String cron;

    private String executeClassPath;

    private Runnable runnable;

    private JobExecuteStrategyEnum executeStrategy;

    private JobTypeEnum jobType;

    private RemoteExecutorMessage remoteExecutorMessage;

    private Integer enable;

    private Integer open;

    private Date createTime;

    private Date updateTime;

    private Map<String, String> featureMap = new HashMap<>();

    private Map<String, String> executeParamMap = new HashMap<>();

    public boolean isEnable() {
        if (this.enable == null) {
            return false;
        }
        return this.enable == 1;
    }

    public boolean isOpen() {
        if (!isEnable()) {
            return false;
        }
        if (this.open == null) {
            return false;
        }
        return this.open == 1;
    }

    public boolean isClose() {
        return !isOpen();
    }

    public boolean isLocalJob() {
        return JobTypeEnum.LOCAL == this.jobType;
    }

    public boolean isRemoteJob() {
        return JobTypeEnum.REMOTE == this.jobType;
    }

    public void setExecuteStrategy(JobExecuteStrategyEnum jobExecuteStrategyEnum) {
        if (jobExecuteStrategyEnum == null) {
            throw new BusinessException(ErrorCode.EXECUTE_STRATEGY_NO_EXIST);
        }
        this.executeStrategy = jobExecuteStrategyEnum;
    }


    public void submitJob() {
        YangJobSubmitParam yangJobSubmitParam = convert2YangJobSubmitParam();
        YangJobManager yangJobManager = getYangJobManager();
        yangJobManager.submitJob(yangJobSubmitParam);
        // 提交任务后，发送提交任务后置事件
        SubmitJobPostEvent submitJobPostEvent = new SubmitJobPostEvent(yangJobSubmitParam);
        getEventCenter().postEvent(submitJobPostEvent);
    }

    public void cancelJob() {
        YangJobManager yangJobManager = getYangJobManager();
        yangJobManager.cancelJob(this.jobId);
    }

    private YangJobSubmitParam convert2YangJobSubmitParam() {
        YangJobSubmitParam yangJobBuildParam = new YangJobSubmitParam();
        yangJobBuildParam.setJobId(this.jobId);
        yangJobBuildParam.setRunnable(this.runnable);
        ZonedDateTime nextExecutionTime = CronUtils.nextExecutionTime(this.cron, ZonedDateTime.now());
        ZonedDateTime nextNextExecutionTime = CronUtils.nextExecutionTime(this.cron, nextExecutionTime);
        long nowEochMill = ZonedDateTime.now().toInstant().toEpochMilli();
        long executeEochMill = nextExecutionTime.toInstant().toEpochMilli();
        long secondExecuteEochMill = nextNextExecutionTime.toInstant().toEpochMilli();
        yangJobBuildParam.setInitialDelay((int)(executeEochMill - nowEochMill) / 1000);
        yangJobBuildParam.setPeriod((int)(secondExecuteEochMill - executeEochMill) / 1000);
        yangJobBuildParam.setJobExecuteStrategy(this.executeStrategy);
        return yangJobBuildParam;
    }

    public void postSaveJobEvent() {
        SaveJobPostEvent saveJobPostEvent = new SaveJobPostEvent(this.jobId);
        getEventCenter().asyncPostEvent(saveJobPostEvent);
    }

    public void postUpdateJobEvent() {
        UpdateJobPostEvent updateJobPostEvent = new UpdateJobPostEvent(this.jobId);
        getEventCenter().asyncPostEvent(updateJobPostEvent);
    }

    public void postDeleteJobEvent() {
        UpdateJobPostEvent updateJobPostEvent = new UpdateJobPostEvent(this.jobId);
        getEventCenter().asyncPostEvent(updateJobPostEvent);
    }

    private YangJobManager getYangJobManager() {
        return SpringContextUtils.getBeanOfType(YangJobManager.class);
    }

    private EventCenter getEventCenter() {
        return SpringContextUtils.getBeanOfType(EventCenter.class);
    }

}

远程任务信息类：

package com.yang.job.admin.domain.valueobject;

import lombok.Data;

import java.io.Serializable;

@Data
public class RemoteExecutorMessage implements Serializable {
    private String ip;

    private Integer port;
}

接着我们添加一个features枚举，用于记录映射features字段中各个key表示的含义，因为我们现在表的设计中没有任务类型字段和远程信息相关的字段，所以会将这些信息添加到features字段中

package com.yang.job.admin.domain.enums;

public enum JobModelFeatureEnum {
    JOB_TYPE("jobType", "任务类型"),
    REMOTE_EXECUTOR_IP("executorIp", "执行器ip"),
    REMOTE_EXECUTOR_PORT("executorPort", "执行器端口"),
    REMOTE_EXECUTOR_MESSAGE("r_executor_m", "远程执行器的信息");

    private String name;

    private String description;

    JobModelFeatureEnum(String name, String description) {
        this.name = name;
        this.description = description;
    }


    public String getName() {
        return name;
    }

    public String getDescription() {
        return description;
    }
}
client层

我们修改原先的NewYangJobCommand类，加上任务类型属性

package com.yang.job.admin.client.dto.command;


import lombok.Data;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;

@Data
public class NewYangJobCommand implements Serializable {
    private String jobName;

    private String description;

    private String cron;

    private String executeStrategy;

    private String jobType;

    private String executeClassPath;

    private Integer open;

    private Map<String, String> params = new HashMap<>();
}

然后修改YangJobDTO类，也加上jobType属性

package com.yang.job.admin.client.dto;


import lombok.Data;

import java.io.Serializable;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@Data
public class YangJobDTO implements Serializable {
    private Integer jobId;

    private String jobName;

    private String description;

    private String cron;

    private String executeStrategy;

    private String executeClassPath;

    private String jobType;

    private Integer enable;

    private Integer open;

    private Date createTime;

    private Date updateTime;

    private Map<String, String> featureMap = new HashMap<>();

    private Map<String, String> executeParamMap = new HashMap<>();
}
application层

接着修改YangJobApplicationService类的convertYangJobModel方法，将jobType任务类型和远程任务信息添加到YangJobModel中

 private YangJobModel convert2YangJobModel(NewYangJobCommand newYangJobCommand) {
        String jobType = newYangJobCommand.getJobType();
        JobTypeEnum jobTypeEnum = JobTypeEnum.getJobTypeByName(jobType);
        if (jobType == null) {
            throw new BusinessException(ErrorCode.PARAM_VALID_ERROR);
        }
        YangJobModel yangJobModel = new YangJobModel();
        yangJobModel.setJobName(newYangJobCommand.getJobName());
        yangJobModel.setDescription(newYangJobCommand.getDescription());
        yangJobModel.setCron(newYangJobCommand.getCron());
        yangJobModel.setOpen(newYangJobCommand.getOpen());
        yangJobModel.setExecuteStrategy(JobExecuteStrategyEnum.getJobExecuteStrategyByName(newYangJobCommand.getExecuteStrategy()));
        yangJobModel.setExecuteClassPath(newYangJobCommand.getExecuteClassPath());
        yangJobModel.setExecuteParamMap(newYangJobCommand.getParams());
        yangJobModel.setJobType(jobTypeEnum);
        if (jobTypeEnum == JobTypeEnum.REMOTE) {
            String ip = newYangJobCommand.getParams().get(JobModelFeatureEnum.REMOTE_EXECUTOR_IP.getName());
            String port = newYangJobCommand.getParams().get(JobModelFeatureEnum.REMOTE_EXECUTOR_PORT.getName());
            if (ip == null || port == null) {
                throw new BusinessException(ErrorCode.PARAM_VALID_ERROR);
            }
            RemoteExecutorMessage remoteExecutorMessage = new RemoteExecutorMessage();
            remoteExecutorMessage.setIp(ip);
            remoteExecutorMessage.setPort(Integer.valueOf(port));
            yangJobModel.setRemoteExecutorMessage(remoteExecutorMessage);
        } else {
            if (yangJobModel.getExecuteClassPath() == null || yangJobModel.getExecuteClassPath().isEmpty()) {
                throw new BusinessException(ErrorCode.UN_LEGAL_CLASS_PATH);
            }
            try {
                Class.forName(yangJobModel.getExecuteClassPath());
            } catch (ClassNotFoundException e) {
                e.printStackTrace();
                throw new BusinessException(ErrorCode.UN_LEGAL_CLASS_PATH);
            }
        }
        return yangJobModel;
    }
infra层

最后修改基础设施层，首先修改YangJobModelConvertor类，将RemoteMessage和JobType转化到features中，以及从features中取出

package com.yang.job.admin.infra.gatewayimpl.repository.convertor;

import com.alibaba.fastjson.JSONObject;
import com.yang.job.admin.domain.enums.JobExecuteStrategyEnum;
import com.yang.job.admin.domain.enums.JobModelFeatureEnum;
import com.yang.job.admin.domain.enums.JobTypeEnum;
import com.yang.job.admin.domain.model.YangJobModel;
import com.yang.job.admin.domain.valueobject.RemoteExecutorMessage;
import com.yang.job.admin.infra.data.YangJobData;
import com.yang.job.admin.infra.job.thread.RemoteJobExecuteThread;
import com.yang.job.admin.infra.utils.FeaturesUtils;
import com.yang.job.core.dto.YangJobTransferDTO;
import com.yang.job.core.execute.IYangJobExecutor;
import com.yang.job.core.execute.YangJobExecuteRequest;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
public class YangJobModelConvertor {
    public YangJobData convert2Data(YangJobModel yangJobModel) {
        if (yangJobModel == null) {
            return null;
        }
        YangJobData yangJobData = new YangJobData();
        yangJobData.setJobId(yangJobModel.getJobId());
        yangJobData.setJobName(yangJobModel.getJobName());
        yangJobData.setDescription(yangJobModel.getDescription());
        yangJobData.setCron(yangJobModel.getCron());
        yangJobData.setExecuteClassPath(yangJobModel.getExecuteClassPath());
        yangJobData.setEnable(yangJobModel.getEnable());
        yangJobData.setOpen(yangJobModel.getOpen());
        yangJobData.setCreateTime(yangJobModel.getCreateTime());
        yangJobData.setUpdateTime(yangJobModel.getUpdateTime());
        Map<String, String> featureMap = yangJobModel.getFeatureMap();
        featureMap.put(JobModelFeatureEnum.JOB_TYPE.getName(), yangJobModel.getJobType().getName());
        featureMap.put(JobModelFeatureEnum.REMOTE_EXECUTOR_MESSAGE.getName(), JSONObject.toJSONString(yangJobModel.getRemoteExecutorMessage()));
        yangJobData.setFeatures(FeaturesUtils.convert2Features(featureMap));
        yangJobData.setExecuteParams(FeaturesUtils.convert2Features(yangJobModel.getExecuteParamMap()));
        yangJobData.setExecuteStrategy(yangJobModel.getExecuteStrategy().getName());
        return yangJobData;
    }

    public YangJobModel convert2Model(YangJobData yangJobData) {
        if (yangJobData == null) {
            return null;
        }
        YangJobModel yangJobModel = new YangJobModel();
        yangJobModel.setJobId(yangJobData.getJobId());
        yangJobModel.setDescription(yangJobData.getDescription());
        yangJobModel.setCron(yangJobData.getCron());
        yangJobModel.setJobName(yangJobData.getJobName());
        yangJobModel.setExecuteClassPath(yangJobData.getExecuteClassPath());
        yangJobModel.setEnable(yangJobData.getEnable());
        yangJobModel.setOpen(yangJobData.getOpen());
        yangJobModel.setCreateTime(yangJobData.getCreateTime());
        yangJobModel.setUpdateTime(yangJobData.getUpdateTime());
        yangJobModel.setFeatureMap(FeaturesUtils.convert2FeatureMap(yangJobData.getFeatures()));
        yangJobModel.setExecuteParamMap(FeaturesUtils.convert2FeatureMap(yangJobData.getExecuteParams()));
        JobExecuteStrategyEnum executeStrategy = JobExecuteStrategyEnum.getJobExecuteStrategyByName(yangJobData.getExecuteStrategy());
        if (executeStrategy == null) {
            throw new RuntimeException("执行策略有误!");
        }

        JobTypeEnum jobType = JobTypeEnum.getJobTypeByName(yangJobModel.getFeatureMap().get(JobModelFeatureEnum.JOB_TYPE.getName()));
        yangJobModel.setJobType(jobType);
        String remoteMessageStr = yangJobModel.getFeatureMap().get(JobModelFeatureEnum.REMOTE_EXECUTOR_MESSAGE.getName());
        RemoteExecutorMessage remoteExecutorMessage = JSONObject.parseObject(remoteMessageStr, RemoteExecutorMessage.class);
        yangJobModel.setRemoteExecutorMessage(remoteExecutorMessage);

        yangJobModel.setExecuteStrategy(executeStrategy);
        yangJobModel.setRunnable(buildRunnable(yangJobModel));

        return yangJobModel;
    }

    private Runnable buildRunnable(YangJobModel yangJobModel) {
        if (yangJobModel.isLocalJob()) {
            String executeClassPath = yangJobModel.getExecuteClassPath();
            try {
                Class<?> aClass = Class.forName(executeClassPath);
                if (!IYangJobExecutor.class.isAssignableFrom(aClass)) {
                    throw new RuntimeException("该类必须实现IYangJobExecutor接口");
                }
                IYangJobExecutor executor = (IYangJobExecutor) aClass.newInstance();
                YangJobExecuteRequest yangJobExecuteRequest = convert2YangJobExecuteRequest(yangJobModel);
                Runnable runnable = () -> executor.execute(yangJobExecuteRequest);
                return runnable;
            } catch (InstantiationException | IllegalAccessException e) {
                e.printStackTrace();
            } catch (ClassNotFoundException e) {
                System.out.println(String.format("%s 类路径对应的类不存在", executeClassPath));
                e.printStackTrace();
            }
        } else {
            RemoteExecutorMessage remoteExecutorMessage = yangJobModel.getRemoteExecutorMessage();
            String executeClassPath = yangJobModel.getExecuteClassPath();

            YangJobTransferDTO yangJobTransferDTO = new YangJobTransferDTO();
            yangJobTransferDTO.setClassName(executeClassPath);

            YangJobExecuteRequest yangJobExecuteRequest = convert2YangJobExecuteRequest(yangJobModel);
            yangJobTransferDTO.setYangJobExecuteRequest(yangJobExecuteRequest);

            return new RemoteJobExecuteThread(remoteExecutorMessage, yangJobTransferDTO);
        }
        return null;
    }

    private static YangJobExecuteRequest convert2YangJobExecuteRequest(YangJobModel yangJobModel) {
        YangJobExecuteRequest yangJobExecuteRequest = new YangJobExecuteRequest();
        yangJobExecuteRequest.setJobId(yangJobModel.getJobId().toString());
        yangJobExecuteRequest.setParams(yangJobModel.getExecuteParamMap());
        return yangJobExecuteRequest;
    }
}

然后添加一个RemoteJobExecuteThread类，该类实现runnable接口，当我们的任务类型为远程调用时，其YangJobModel的runnable属性为remoteJobExecuteThread类

package com.yang.job.admin.infra.job.thread;

import com.alibaba.fastjson.JSONObject;
import com.yang.job.admin.domain.valueobject.RemoteExecutorMessage;
import com.yang.job.core.dto.YangJobTransferDTO;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.net.UnknownHostException;

public class RemoteJobExecuteThread implements Runnable {
    private YangJobTransferDTO yangJobTransferDTO;

    private RemoteExecutorMessage remoteExecutorMessage;

    public RemoteJobExecuteThread(RemoteExecutorMessage remoteExecutorMessage, YangJobTransferDTO yangJobTransferDTO) {
        this.remoteExecutorMessage = remoteExecutorMessage;
        this.yangJobTransferDTO = yangJobTransferDTO;
    }

    @Override
    public void run() {
        try {
            String ip = remoteExecutorMessage.getIp();
            Integer port = remoteExecutorMessage.getPort();
            Socket socket = new Socket(ip, port);
            PrintWriter printWriter = new PrintWriter(socket.getOutputStream(), true);
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));

            printWriter.println(JSONObject.toJSONString(yangJobTransferDTO));

            bufferedReader.close();
            printWriter.close();
            socket.close();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
测试

我们先启动之前的sample1项目，然后启动yang-job-admin，调用http://localhost:8080/job添加任务，请求体如下：

{
  "jobName": "RemoteJobExecutor",
"description":"RemoteJobExecutor",
  "cron": "0/10 * * * * ?",
  "executeStrategy": "withFixedDelay",
  "executeClassPath": "com.yang.job.sample1.task.TestTask1",
"open":1,
"jobType":"remote",
"params":{
"executorIp":"127.0.0.1",
"executorPort":"9999"
}
}

image.png
添加成功后，我们查看Sample1项目的控制台，可以看到，每10秒，这个TestTask1任务会被调用一次
image.png

参考文章

https://www.yihuo.tech/programming/server-stack/exploring-the-java-network-programming-paradigm-socket-udp-nio-and-netty-in-focus/

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/625608.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

车载GPT爆红前夜：一场巨头竞逐的游戏

车载GPT爆红前夜：一场巨头竞逐的游戏

在基于GPT-3.5的ChatGPT问世之前&#xff0c;OpenAI作为深度学习领域并不大为人所看好的技术分支玩家&#xff0c;已经在GPT这个赛道默默耕耘了七八年的时间。 好几年的时间里&#xff0c;GPT始终没有跨越从“不能用”到“能用”的奇点。转折点发生在2020年6月份发布的GPT-3&a…
阅读更多...
实时“秒回”，像真人一样语音聊天，GPT-4o模型强到恐怖

实时“秒回”，像真人一样语音聊天，GPT-4o模型强到恐怖

今天凌晨OpenAl发布了 GPT-4o&#xff0c;这是一种新的人工智能模式&#xff0c;集合了文本、图片、视频、语音的全能模型。 能实时响应用户的需求&#xff0c;并通过语音来实时回答你&#xff0c;你可以随时打断它。还具有视觉能力&#xff0c;能识别物体并根据视觉做出快速的…
阅读更多...
Fabric实现多GPU运行

Fabric实现多GPU运行

官方的将pytorch转换为fabric简单分为五个步骤&#xff1a; 步骤 1&#xff1a; 在训练代码的开头创建 Fabric 对象 from lightning.fabric import Fabricfabric Fabric() 步骤 2&#xff1a; 如果打算使用多个设备&#xff08;例如多 GPU&#xff09;&#xff0c;就调用…
阅读更多...
gin导出excel文件

gin导出excel文件

go可以通过excelize 包实现对excel的操作 "github.com/xuri/excelize/v2"导出示例 service层 批量导出数据的&#xff0c;我们可以在dao层中返回一个切片。在service中新建一个excelize对象&#xff0c;单独设置表头。遍历切片往excelize上修改即可。 func (s *S…
阅读更多...
05-13 周一 量化是什么

05-13 周一 量化是什么

05-13 周一 量化是什么 时间版本修改人描述2024年5月13日11:27:25V0.1宋全恒新建文档2024年5月14日16:21:20V1.0宋全恒了解 简介 神经网络在运行时有较高的计算成本&#xff0c;而且随着大模型时代的到来&#xff0c;知识由一个巨大的LLM存储&#xff0c;为了获取知识&#xf…
阅读更多...
【STM32】状态机实现定时器按键消抖，处理单击、双击、三击、长按事件

【STM32】状态机实现定时器按键消抖，处理单击、双击、三击、长按事件

目录 一、简单介绍 二、模块与接线 三、cubemx配置 四、驱动编写 状态图 按键类型定义 参数初始化/复位 按键扫描 串口重定向 主函数 五、效果展示 六、驱动附录 key.c key.h 一、简单介绍 众所周知&#xff0c;普通的机械按键会产生抖动&#xff0c;可以采取硬件…
阅读更多...
【江南大学×朗汀留学】部分留学录取案例合集

【江南大学×朗汀留学】部分留学录取案例合集

朗汀留学 X 江南大学 尽管客观条件如此艰难&#xff0c;朗汀留学的同学们还是斩获众多名校的录取。成绩属于过去&#xff0c;我们继续努力创造更好未来。 以下为我们摘取的江南大学部分学生案例供大家参考&#xff0c;再次恭喜所有获得理想大学offer的学生们&#xff0c;你们…
阅读更多...
CAST: Cross-Attention in Space and Time for Video Action Recognition

CAST: Cross-Attention in Space and Time for Video Action Recognition

标题&#xff1a;CAST: 时空交叉注意力网络用于视频动作识别 原文链接&#xff1a;2311.18825v1 (arxiv.org)https://arxiv.org/pdf/2311.18825v1 源码链接&#xff1a;GitHub - KHU-VLL/CASThttps://github.com/KHU-VLL/CAST 发表&#xff1a;NeurIPS-2023&#xff08;CCF A…
阅读更多...
Linux第四节--常见的指令介绍集合（持续更新中）

Linux第四节--常见的指令介绍集合（持续更新中）

点赞关注不迷路&#xff01;本节涉及初识Linux第四节&#xff0c;主要为常见的几条指令介绍。 如果文章对你有帮助的话 欢迎 评论&#x1f4ac; 点赞&#x1f44d;&#x1f3fb; 收藏 ✨ 加关注&#x1f440; 期待与你共同进步! 1. more指令 语法&#xff1a;more [选项][文件]…
阅读更多...
MySQL文档_下载

MySQL文档_下载

可能需要&#xff1a;MySQL下载–》更新版本–》迁移数据库到MySQL 以下都不重要【只要确定好需要安装版本&#xff0c;找到对应的版本下载&#xff0c;安装&#xff0c;设置即可】 下载、安装&#xff1a; Determine whether MySQL runs and is supported on your platform…
阅读更多...
电脑刚开机的时候不卡，用一会就变卡顿了…怎么回事？

电脑刚开机的时候不卡，用一会就变卡顿了…怎么回事？

前言 昨天咱们聊到旧电脑更换了固态硬盘之后&#xff0c;开机就会变快的事情&#xff0c;这个确实是可行且有效的升级电脑办法之一。 看完这篇之后&#xff0c;切莫着急升级电脑硬件配置&#xff0c;因为这里的坑比你想象的还要多。 从机械硬盘测试的数据和固态硬盘的测试数…
阅读更多...
【C语言】—— 动态内存管理

【C语言】—— 动态内存管理

【C语言】——动态内存管理 一、动态内存管理概述1.1、动态内存的概念1.2、动态内存的必要性 二、 m a l l o c malloc malloc 函数2.1、函数介绍2.2、应用举例 三、 c a l l o c calloc calloc 函数四、 f r e e free free 函数4.1、函数介绍4.2、应用举例 五、 r e a l l o …
阅读更多...
SSL协议

SSL协议

SSL 安全传输协议&#xff08;安全套接层&#xff09; 也叫TLS ---- 传输层安全协议 SSL的工作原理&#xff1a;SSL协议因为是基于TCP协议工作的&#xff0c;通信双方需要先建立TCP会话。因为SSL协议需要进行安全保证&#xff0c;需要协商安全参数&#xff0c;所以也需要建立…
阅读更多...
【数据结构】栈和队列专题

【数据结构】栈和队列专题

前言 上篇博客我们讨论了栈和队列的有关结构&#xff0c;本篇博客我们继续来讨论有关栈和队列习题 这些题算是经典了 &#x1f493; 个人主页&#xff1a;小张同学zkf ⏩ 文章专栏&#xff1a;数据结构 若有问题 评论区见&#x1f4dd; &#x1f389;欢迎大家点赞&#x1f44d…
阅读更多...
Oracle 临时表空间的管理

Oracle 临时表空间的管理

Oracle 临时表空间的管理 临时表空间的处理 1.创建一个新的temporary tablespace; create temporary tablespace tp tempfile ...... size 10m autoextend on; 2.改变数据库的默认临时表空间 alter database default temporary tablespace tp; 3。drop tablespace temp; …
阅读更多...
Zoho CRM企业成长的智能引擎，智能化销售自动化

Zoho CRM企业成长的智能引擎，智能化销售自动化

数字化时代&#xff0c;客户体验已成为企业竞争的核心要素。卓豪Zoho CRM&#xff0c;作为全球领先的SaaS云端客户关系管理平台&#xff0c;正引领着一场企业运营模式的变革&#xff0c;助力超过25万家企业跨越180多个国家&#xff0c;实现客户互动与业务增长的无缝对接。让我们…
阅读更多...
Verlog-流水灯-FPGA

Verlog-流水灯-FPGA

Verlog-流水灯-FPGA 引言&#xff1a; ​ 随着电子技术的飞速发展&#xff0c;现场可编程门阵列&#xff08;FPGA&#xff09;已成为电子设计自动化&#xff08;EDA&#xff09;领域中不可或缺的组件。FPGA以其高度的灵活性和可定制性&#xff0c;广泛应用于通信、图像处理、工…
阅读更多...
【C++】学习笔记——继承_2

【C++】学习笔记——继承_2

文章目录 十二、继承5. 继承与友元6. 继承与静态成员7. 复杂的菱形继承及菱形虚拟继承 未完待续 十二、继承 5. 继承与友元 友元关系不能继承&#xff0c;也就是说父类友元不能访问子类私有和保护成员 。除非子类也设置成友元。 6. 继承与静态成员 父类定义了 static 静态成…
阅读更多...
单用户模式破解root密码

单用户模式破解root密码

目录 一. 破解root密码 1. 查看操作系统版本 2.重启系统&#xff0c;进入grub菜单&#xff0c;选择要使用的内核&#xff0c;按e进入​编辑 3. 找到linux16那一行&#xff0c;把光标移动到最后&#xff0c;添加 init/bin/sh 然后ctrlx保存退出会自动进入系统 4. 进入系统后…
阅读更多...
Spring WebFlux：响应式编程

Spring WebFlux：响应式编程

在软件开发领域&#xff0c;随着互联网应用的规模和复杂性不断增加&#xff0c;传统的编程模型逐渐暴露出一些局限性&#xff0c;尤其是在面对高并发、大规模数据流处理等场景时。为了应对这些挑战&#xff0c;响应式编程&#xff08;Reactive Programming&#xff09;应运而生…
阅读更多...
最新文章

Copyright @ 2022~2023