xxljob定时在sftp上传下载
2024-05-12 05:32:06
@Data
@AllArgsConstructor
@NoArgsConstructor
@Component
public class SftpConfig {/*** sftp 服务器地址*/@Value("${sftp.host}")private String host;/*** sftp 服务器端口*/@Value("${sftp.port}")private int port;/*** sftp服务器登陆用户名*/@Value("${sftp.username}")private String user;/*** sftp 服务器登陆密码* 密码和私钥二选一*/@Value("${sftp.password}")private String password;/*** 私钥文件* 私钥和密码二选一*/@Value("${sftp.identity}")private String identity;
}
public class SftpClient {public static boolean downloadFile(SftpConfig sftpConfig, SftpDownloadRequest request) {Session session = null;ChannelSftp channelSftp = null;try {session = getSession(sftpConfig);channelSftp = getChannelSftp(session);String remoteFileDir = getRemoteFileDir(request.getRemoteFilePath());String remoteFileName = getRemoteFileName(request.getRemoteFilePath());// 校验SFTP上文件是否存在if (!isFileExist(channelSftp, remoteFileDir, remoteFileName)) {return false;}if(StringUtils.isBlank(remoteFileDir)){remoteFileDir = "\\";}// 切换到SFTP文件目录channelSftp.cd(remoteFileDir);// 下载文件File localFile = new File(request.getLocalFilePath());FileUtils.forceMkdir(localFile);FileUtils.deleteQuietly(localFile);channelSftp.get(remoteFileName, request.getLocalFilePath());channelSftp.rm(request.getRemoteFilePath());return true;} catch (JSchException jSchException) {throw new RuntimeException("sftp connect failed:" + JsonUtils.toJson(sftpConfig), jSchException);} catch (SftpException sftpException) {throw new RuntimeException("sftp download file failed:" + JsonUtils.toJson(request), sftpException);} catch (IOException e) {throw new RuntimeException(e);} finally {disconnect(channelSftp, session);}}public static boolean uploadFile(SftpConfig sftpConfig, SftpUploadRequest request) {Session session = null;ChannelSftp channelSftp = null;try {session = getSession(sftpConfig);channelSftp = getChannelSftp(session);String remoteFileDir = getRemoteFileDir(request.getRemoteFilePath());String remoteFileName = getRemoteFileName(request.getRemoteFilePath());// 切换到SFTP文件目录cdOrMkdir(channelSftp, remoteFileDir);// 上传文件channelSftp.put(request.getLocalFilePath(), remoteFileName);// if (StringUtils.isBlank(request.getEndFlag())) {
// channelSftp.put(request.getLocalFilePath() + request.getEndFlag(),
// remoteFileName + request.getEndFlag());
// }return true;} catch (JSchException jSchException) {throw new RuntimeException("sftp connect failed: " + JsonUtils.toJson(sftpConfig), jSchException);} catch (SftpException sftpException) {throw new RuntimeException("sftp upload file failed: " + JsonUtils.toJson(request), sftpException);} finally {disconnect(channelSftp, session);}}public static Session getSession(SftpConfig sftpConfig) throws JSchException {
// Session session;
// JSch jsch = new JSch();/*if (StringUtils.isBlank(sftpConfig.getIdentity())) {jsch.addIdentity(sftpConfig.getIdentity());}*//*if (sftpConfig.getPort() <= 0) {// 默认端口session = jsch.getSession(sftpConfig.getUser(), sftpConfig.getHost());} else {// 指定端口session = jsch.getSession(sftpConfig.getUser(), sftpConfig.getHost(), sftpConfig.getPort());}if (StringUtils.isBlank(sftpConfig.getPassword())) {session.setPassword(sftpConfig.getPassword());}session.setConfig("StrictHostKeyChecking", "no");session.setTimeout(10 * 1000); // 设置超时时间10ssession.connect();
*/Session session = null;String config = "Port "+sftpConfig.getPort()+"\n" +"\n" +"Host foo\n" +" User "+sftpConfig.getUser()+"\n" +" Hostname "+sftpConfig.getHost()+"\n" +"Host *\n" +" ConnectTime 3000\n" +" PerferredAuthentications Keyboard-interact,password,publicKey\n" +" #ForwardAgent yes\n" +" #StrictHostKeyChecking no\n" +" #identiyFile ~/.ssh/id_rsa\n" +" #UserKnownHostFile ~/.shh/known_hosts";ConfigRepository configRepository = null;try {configRepository = OpenSSHConfig.parse(config);JSch jSch = new JSch();jSch.setConfigRepository(configRepository);//"foo" is from "Host foo" in the above configsession = jSch.getSession("foo");session.setPassword(sftpConfig.getPassword());session.setConfig("StrictHostKeyChecking", "no");session.connect(50000);return session;} catch (IOException e) {throw new RuntimeException(e);}}public static ChannelSftp getChannelSftp(Session session) throws JSchException {ChannelSftp channelSftp = (ChannelSftp) session.openChannel("sftp");channelSftp.connect();return channelSftp;}/*** SFTP文件是否存在* true：存在；false：不存在*/private static boolean isFileExist(ChannelSftp channelSftp,String fileDir,String fileName,String endFlag) throws SftpException {if (StringUtils.isBlank(endFlag)) {if (!isFileExist(channelSftp, fileDir, fileName + endFlag)) {return false;}} else {if (!isFileExist(channelSftp, fileDir, fileName)) {return false;}}return true;}/*** SFTP文件是否存在* true：存在；false：不存在*/public static boolean isFileExist(ChannelSftp channelSftp,String fileDir,String fileName) throws SftpException {if (StringUtils.isBlank(fileDir)){fileDir = "\\";}if (!isDirExist(channelSftp, fileDir)) {return false;}Vector vector = channelSftp.ls(fileDir);for (int i = 0; i < vector.size(); ++i) {ChannelSftp.LsEntry entry = (ChannelSftp.LsEntry) vector.get(i);if (fileName.equals(entry.getFilename())) {return true;}}return false;}/*** sftp上目录是否存在* true：存在；false：不存在*/private static boolean isDirExist(ChannelSftp channelSftp, String fileDir) {try {SftpATTRS sftpATTRS = channelSftp.lstat(fileDir);return sftpATTRS.isDir();} catch (SftpException e) {return false;}}private static void cdOrMkdir(ChannelSftp channelSftp, String fileDir) throws SftpException {if (StringUtils.isBlank(fileDir)) {return;}for (String dirName : fileDir.split(File.separator)) {if (StringUtils.isBlank(dirName)) {dirName = File.separator;}if (!isDirExist(channelSftp, dirName)) {channelSftp.mkdir(dirName);}channelSftp.cd(dirName);}}public static String getRemoteFileDir(String remoteFilePath) {int remoteFileNameindex = remoteFilePath.lastIndexOf(File.separator);return remoteFileNameindex == -1? "": remoteFilePath.substring(0, remoteFileNameindex);}public static String getRemoteFileName(String remoteFilePath) {int remoteFileNameindex = remoteFilePath.lastIndexOf(File.separator);if (remoteFileNameindex == -1) {return remoteFilePath;}String remoteFileName = remoteFileNameindex == -1? remoteFilePath: remoteFilePath.substring(remoteFileNameindex + 1);if (StringUtils.isBlank(remoteFileName)) {throw new RuntimeException("remoteFileName is blank");}return remoteFileName;}private static void disconnect(ChannelSftp channelSftp, Session session) {if (channelSftp != null) {channelSftp.disconnect();}if (session != null) {session.disconnect();}}/*** 获取文件夹下的文件** @param directory 路径* @return*/public static Vector<?> listFiles(ChannelSftp channelSftp, String directory) {try {if (isDirExist(channelSftp,directory)) {Vector<?> vector = null;vector = channelSftp.ls(directory);//移除上级目录和根目录："." ".."vector.remove(0);vector.remove(0);return vector;}} catch (SftpException e) {throw new RuntimeException(e);}return null;}
}
@Data
@AllArgsConstructor
@NoArgsConstructor
public class SftpDownloadRequest {/*** sftp上完整文件名*/private String remoteFilePath;/*** 本地完整文件名*/private String localFilePath;/*** 文件完成标识* 非必选*/private String endFlag;
}
@Data
@AllArgsConstructor
@NoArgsConstructor
public class SftpUploadRequest {/*** 本地完整文件名*/private String localFilePath;/*** sftp上完整文件名*/private String remoteFilePath;/*** 文件完成标识* 非必选*/private String endFlag;
}
@Data
@AllArgsConstructor
@NoArgsConstructor
public class returnParm<T> {private int state;private String msg;private T data;
}
@Configuration
public class RestTemplateConfig {@Beanpublic RestTemplate restTemplate(ClientHttpRequestFactory factory) {return new RestTemplate(factory);}@Beanpublic ClientHttpRequestFactory simpleClientHttpRequestFactory() {SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();factory.setReadTimeout(150000); // msfactory.setConnectTimeout(150000); // msreturn factory;}
}
@Configuration
public class XxlJobConfigSftp {private Logger logger = LoggerFactory.getLogger(XxlJobConfigSftp.class);@Value("${xxl.job.admin.addresses}")private String adminAddresses;@Value("${xxl.job.accessToken}")private String accessToken;@Value("${xxl.job.executor.appname}")private String appname;@Value("${xxl.job.executor.address}")private String address;@Value("${xxl.job.executor.ip}")private String ip;@Value("${xxl.job.executor.port}")private int port;@Value("${xxl.job.executor.logpath}")private String logPath;@Value("${xxl.job.executor.logretentiondays}")private int logRetentionDays;@Beanpublic XxlJobSpringExecutor xxlJobExecutor() {logger.info(">>>>>>>>>>> xxl-job config init.");XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();xxlJobSpringExecutor.setAdminAddresses(adminAddresses);xxlJobSpringExecutor.setAppname(appname);xxlJobSpringExecutor.setAddress(address);xxlJobSpringExecutor.setIp(ip);xxlJobSpringExecutor.setPort(port);xxlJobSpringExecutor.setAccessToken(accessToken);xxlJobSpringExecutor.setLogPath(logPath);xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);return xxlJobSpringExecutor;}/*** 针对多网卡、容器内部署等情况，可借助 "spring-cloud-commons" 提供的 "InetUtils" 组件灵活定制注册IP；** 1、引入依赖：* <dependency>* <groupId>org.springframework.cloud</groupId>* <artifactId>spring-cloud-commons</artifactId>* <version>${version}</version>* </dependency>** 2、配置文件，或者容器启动变量* spring.cloud.inetutils.preferred-networks: 'xxx.xxx.xxx.'** 3、获取IP* String ip_ = inetUtils.findFirstNonLoopbackHostInfo().getIpAddress();*/}
public class FileUtils {/*** <h1>获取指定文件夹下所有文件，不含文件夹</h1>* @param dirFilePath 文件夹路径* @return*/public static List<File> getAllFile(String dirFilePath){if(StringUtils.isBlank(dirFilePath))return null;return getAllFile(new File(dirFilePath));}/*** <h1>获取指定文件夹下所有文件，不含文件夹</h1>* @param dirFile 文件夹* @return*/public static List<File> getAllFile(File dirFile){// 如果文件夹不存在或着不是文件夹，则返回 nullif(Objects.isNull(dirFile) || !dirFile.exists() || dirFile.isFile())return null;File[] childrenFiles = dirFile.listFiles();if(Objects.isNull(childrenFiles) || childrenFiles.length == 0)return null;List<File> files = new ArrayList<>();for(File childFile : childrenFiles) {// 如果时文件，直接添加到结果集合if(childFile.isFile()) {files.add(childFile);}else {// 如果是文件夹，则将其内部文件添加进结果集合List<File> cFiles = getAllFile(childFile);if(Objects.isNull(cFiles) || cFiles.isEmpty()) continue;files.addAll(cFiles);}}return files;}public static void copyFileUsingFileStreams(File source, File dest)throws IOException {InputStream input = null;OutputStream output = null;try {input = new FileInputStream(source);output = new FileOutputStream(dest);byte[] buf = new byte[1024];int bytesRead;while ((bytesRead = input.read(buf)) > 0) {output.write(buf, 0, bytesRead);}} finally {input.close();output.close();}}}
@Slf4j
public class JsonUtils {private static ObjectMapper om = new ObjectMapper();static {// 对象的所有字段全部列入，还是其他的选项，可以忽略null等om.setSerializationInclusion(Include.ALWAYS);// 设置Date类型的序列化及反序列化格式om.setDateFormat(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));// 忽略空Bean转json的错误om.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);// 忽略未知属性，防止json字符串中存在，java对象中不存在对应属性的情况出现错误om.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);// 注册一个时间序列化及反序列化的处理模块，用于解决jdk8中localDateTime等的序列化问题om.registerModule(new JavaTimeModule());}/*** 对象 => json字符串** @param obj 源对象*/public static <T> String toJson(T obj) {String json = null;if (obj != null) {try {json = om.writeValueAsString(obj);} catch (JsonProcessingException e) {log.warn(e.getMessage(), e);throw new IllegalArgumentException(e.getMessage());}}return json;}/*** json字符串 => 对象** @param json 源json串* @param clazz 对象类* @param <T> 泛型*/public static <T> T parse(String json, Class<T> clazz) {return parse(json, clazz, null);}/*** json字符串 => 对象** @param json 源json串* @param type 对象类型* @param <T> 泛型*/public static <T> T parse(String json, TypeReference type) {return parse(json, null, type);}/*** json => 对象处理方法* <br>* 参数clazz和type必须一个为null，另一个不为null* <br>* 此方法不对外暴露，访问权限为private** @param json 源json串* @param clazz 对象类* @param type 对象类型* @param <T> 泛型*/private static <T> T parse(String json, Class<T> clazz, TypeReference type) {T obj = null;if (!StringUtils.isEmpty(json)) {try {if (clazz != null) {obj = om.readValue(json, clazz);} else {obj = (T) om.readValue(json, type);}} catch (IOException e) {log.warn(e.getMessage(), e);throw new IllegalArgumentException(e.getMessage());}}return obj;}
}
public class ZipUtils
{private static final int BUFFER_SIZE = 2 * 1024;/*** 压缩成ZIP 方法1** @param srcDir 压缩文件夹路径* @param out 压缩文件输出流* @param KeepDirStructure 是否保留原来的目录结构,true:保留目录结构;* false:所有文件跑到压缩包根目录下(注意：不保留目录结构可能会出现同名文件,会压缩失败)* @throws RuntimeException 压缩失败会抛出运行时异常*/public static void toZip(String srcDir, OutputStream out, boolean KeepDirStructure)throws RuntimeException{long start = System.currentTimeMillis();ZipOutputStream zos = null;try{zos = new ZipOutputStream(out);File sourceFile = new File(srcDir);compress(sourceFile, zos, sourceFile.getName(), KeepDirStructure);long end = System.currentTimeMillis();System.out.println("压缩完成，耗时：" + (end - start) + " ms");} catch (Exception e){throw new RuntimeException("zip error from ZipUtils", e);} finally{if (zos != null){try{zos.close();} catch (IOException e){e.printStackTrace();}}}}/*** 压缩成ZIP 方法2** @param srcFiles 需要压缩的文件列表* @param out 压缩文件输出流* @throws RuntimeException 压缩失败会抛出运行时异常*/public static void toZip(List<File> srcFiles, OutputStream out) throws RuntimeException{long start = System.currentTimeMillis();ZipOutputStream zos = null;try{zos = new ZipOutputStream(out);for (File srcFile : srcFiles){byte[] buf = new byte[BUFFER_SIZE];zos.putNextEntry(new ZipEntry(srcFile.getName()));int len;FileInputStream in = new FileInputStream(srcFile);while ((len = in.read(buf)) != -1){zos.write(buf, 0, len);}zos.closeEntry();in.close();}long end = System.currentTimeMillis();System.out.println("压缩完成，耗时：" + (end - start) + " ms");} catch (Exception e){throw new RuntimeException("zip error from ZipUtils", e);} finally{if (zos != null){try{zos.close();} catch (IOException e){e.printStackTrace();}}}}/*** 递归压缩方法** @param sourceFile 源文件* @param zos zip输出流* @param name 压缩后的名称* @param KeepDirStructure 是否保留原来的目录结构,true:保留目录结构;* false:所有文件跑到压缩包根目录下(注意：不保留目录结构可能会出现同名文件,会压缩失败)* @throws Exception*/private static void compress(File sourceFile, ZipOutputStream zos, String name,boolean KeepDirStructure) throws Exception{byte[] buf = new byte[BUFFER_SIZE];if (sourceFile.isFile()){// 向zip输出流中添加一个zip实体，构造器中name为zip实体的文件的名字zos.putNextEntry(new ZipEntry(name));// copy文件到zip输出流中int len;FileInputStream in = new FileInputStream(sourceFile);while ((len = in.read(buf)) != -1){zos.write(buf, 0, len);}// Complete the entryzos.closeEntry();in.close();} else{File[] listFiles = sourceFile.listFiles();if (listFiles == null || listFiles.length == 0){// 需要保留原来的文件结构时,需要对空文件夹进行处理if (KeepDirStructure){// 空文件夹的处理zos.putNextEntry(new ZipEntry(name + "/"));// 没有文件，不需要文件的copyzos.closeEntry();}} else{for (File file : listFiles){// 判断是否需要保留原来的文件结构if (KeepDirStructure){// 注意：file.getName()前面需要带上父文件夹的名字加一斜杠,// 不然最后压缩包中就不能保留原来的文件结构,即：所有文件都跑到压缩包根目录下了compress(file, zos, name + "/" + file.getName(), KeepDirStructure);} else{compress(file, zos, file.getName(), KeepDirStructure);}}}}}public static void unZipIt(String file, String outputFolder) throws IOException {ZipFile zipFile = null;InputStream in=null;OutputStream out=null;try {zipFile = new ZipFile(file);Enumeration <? extends ZipEntry> entries = zipFile.entries();while (entries.hasMoreElements()) {ZipEntry entry = entries.nextElement();File entryDestination = new File(outputFolder, entry.getName());if (entry.isDirectory()) {entryDestination.mkdirs();} else {entryDestination.getParentFile().mkdirs();try {out = new FileOutputStream(entryDestination);in = zipFile.getInputStream(entry);IOUtils.copy(in, out);} catch (IOException e) {throw e;}finally {IOUtils.closeQuietly(in);IOUtils.closeQuietly(out);}}}} finally {try {if(zipFile!=null){zipFile.close();zipFile=null;}} catch (IOException e) {}}}public static File newFile(File destinationDir, ZipEntry zipEntry){File destFile = null;try{destFile = new File(destinationDir, zipEntry.getName());String destDirPath = destinationDir.getCanonicalPath();String destFilePath = destFile.getCanonicalPath();if (!destFilePath.startsWith(destDirPath + File.separator)){destFile = null;}} catch (Exception e){}return destFile;}}
@Component
public class SampleXxlJobSftp {private static Logger logger = LoggerFactory.getLogger(SampleXxlJobSftp.class);@Value("${sftp.localFilePath}")private String localFilePath;@Value("${tdm.ip}")private String tdmIp;@Value("${tdm.port}")private String tdmPort;@Value("${ods.ip}")private String ip;@Value("${ods.port}")private String port;@Autowiredprivate SftpConfig sftpConfig;@Resourceprivate RestTemplate restTemplate;@XxlJob("demoJobHandler")public void demoJobHandler() throws Exception {String jobParam = XxlJobHelper.getJobParam();String[] methodParams = jobParam.split(",");String localFilePath = methodParams[0];String remoteFilePath = methodParams[1];logger.info("本地完整文件名..[{}]", localFilePath);logger.info("sftp上完整文件名...[{}]", remoteFilePath);boolean fileExist = false;try {String remoteFileDir = SftpClient.getRemoteFileDir(remoteFilePath);String remoteFileName = SftpClient.getRemoteFileName(remoteFilePath);Session session = SftpClient.getSession(sftpConfig);ChannelSftp channelSftp = SftpClient.getChannelSftp(session);fileExist = SftpClient.isFileExist(channelSftp, remoteFileDir, remoteFileName);} catch (JSchException e) {throw new RuntimeException(e);} catch (SftpException e) {throw new RuntimeException(e);}File file = new File(localFilePath);boolean exists = file.exists();if (exists) {logger.info("当前文件已下载成功");return;}if (fileExist) {SftpDownloadRequest sftpDownloadRequest = new SftpDownloadRequest();sftpDownloadRequest.setLocalFilePath(localFilePath);sftpDownloadRequest.setRemoteFilePath(remoteFilePath);boolean result = SftpClient.downloadFile(sftpConfig, sftpDownloadRequest);if (result) {XxlJobHelper.handleSuccess();}}}@XxlJob("packageJobHandler")public void packageJobHandler() throws Exception {String strFilter = XxlJobHelper.getJobParam();String strResult = "";String[] arrTestIds = strFilter.split(",");byte[] resultBytes = null;try {ExportRequest.Builder exportBuilder = ExportRequest.newBuilder();ExportRequest.AtfxSelectStatement.Builder statementBuilder = ExportRequest.AtfxSelectStatement.newBuilder();statementBuilder.setElementName("TPR_T_TEST");SelectStatement.ConditionItem.Builder conditionItemBuilder = SelectStatement.ConditionItem.newBuilder();SelectStatement.ConditionItem.Condition.Builder conditionBuilder = SelectStatement.ConditionItem.Condition.newBuilder();conditionBuilder.setAid(7341971803063440745L);conditionBuilder.setAttribute("GID");conditionBuilder.setOperator(SelectStatement.ConditionItem.Condition.OperatorEnum.OP_EQ);StringArray.Builder stringArrayBuilder = StringArray.newBuilder();List<String> attTestIdList = new ArrayList<>();for (String strTestId : arrTestIds) {attTestIdList.add(strTestId);}stringArrayBuilder.addAllValues(attTestIdList);conditionBuilder.setStringArray(stringArrayBuilder);conditionItemBuilder.setCondition(conditionBuilder);List<SelectStatement.ConditionItem> conditionItems = new ArrayList<>();conditionItems.add(conditionItemBuilder.build());statementBuilder.addAllWhere(conditionItems);exportBuilder.setStatement(statementBuilder);//导出方式exportBuilder.setExportType(ExportRequest.ExportTypeEnum.RELATION);logger.info("即将调用Atfx导出接口，构建的exportBuilder为:" + exportBuilder.toString());byte[] param = exportBuilder.build().toByteArray();HttpHeaders headers = new HttpHeaders();headers.setContentType(MediaType.APPLICATION_JSON_UTF8);HttpEntity httpEntity = new HttpEntity<>(headers);HttpEntity<String> response = restTemplate.exchange("http://" + ip + ":" + port + "/ods", HttpMethod.POST, httpEntity, String.class);String[] split = response.getHeaders().get("Location").get(0).split("/");String conI = split[split.length - 1];HttpHeaders httpHeaders = new HttpHeaders();headers.setContentType(MediaType.valueOf("application/x-asamods+protobuf"));HttpEntity<byte[]> byteBody = new HttpEntity<>(param, httpHeaders);//param 加到 body 中ResponseEntity<ByteArrayResource> exchange = restTemplate.exchange("http://" + ip + ":" + port + "/ods/" + conI + "/export-atfx", HttpMethod.POST, byteBody, ByteArrayResource.class);resultBytes = exchange.getBody().getByteArray();if (resultBytes != null) {ExportResponse exportResponse = ExportResponse.parseFrom(resultBytes);if (exportResponse == null) {//失败}String atfxFilePath = exportResponse.getAtfxFilePath();//strResult是返回文件的路径//类似于D:\mapletr4j\MapleTR4J_springboot\mapletr4j-ods\target\classes\temp\atfx\20220525150458\all(20220525150458).atfx//strResult = new String(resultBytes, java.nio.charset.StandardCharsets.UTF_8);strResult = new String(resultBytes, StandardCharsets.US_ASCII);logger.info("调用Atfx导出接口，返回数据为:" + strResult);String strPrefix = strResult.substring(0, strResult.indexOf(":") - 1);if (!StringUtils.isBlank(strPrefix)) {strResult = strResult.replace(strPrefix, "");}//对导出的文件进行打包压缩File atfxFile = new File(atfxFilePath);String strParentPath = atfxFile.getParent();ZipUtils zipUtils = new ZipUtils();FileOutputStream fos1 = new FileOutputStream(new File(strParentPath + ".zip"));logger.info("正在为导出生成的文件进行打包压缩");zipUtils.toZip(strParentPath, fos1, true);strResult = strParentPath + ".zip";logger.info("打包压缩已完成，地址为" + strResult);//删除之前的文件File fileDir = new File(strParentPath);deleteDirectory(fileDir);logger.info("已清理之前的文件夹");//通知调用方完成打包。
// String callbackUrl="";} else {logger.info("调用Atfx导出接口，未返回任何数据:");}} catch (Exception ex) {logger.error("ODSTODSTestCenterAccessService 导出Atfx过程中遇到错误:" + ex.getMessage());ex.printStackTrace();}
// restTemplate.postForObject("http://" + tdmIp + ":" + tdmPort + "/ods")XxlJobHelper.handleSuccess();}@XxlJob("downloadJobHandler")public void downloadJobHandler() throws Exception {
// String jobParam = XxlJobHelper.getJobParam();
// String[] methodParams = jobParam.split(",");
// String localFilePath = methodParams[0];
// String remoteFilePath = methodParams[0];Session session = null;try {session = SftpClient.getSession(sftpConfig);ChannelSftp channelSftp = null;try {channelSftp = SftpClient.getChannelSftp(session);String remoteFilePath = "\\";Vector vector = channelSftp.ls("\\");vector.remove(0);if (vector.size()>0){// Vector<?> vector = SftpClient.listFiles(channelSftp, "\\");String filename = ((ChannelSftp.LsEntry) vector.get(0)).getFilename();remoteFilePath = remoteFilePath + filename;String localFilePathName = localFilePath + "\\" + filename;logger.info("本地完整文件名..[{}]", localFilePathName);logger.info("sftp上完整文件名...[{}]", remoteFilePath);boolean fileExist = false;try{String remoteFileDir = SftpClient.getRemoteFileDir(remoteFilePath);String remoteFileName = SftpClient.getRemoteFileName(remoteFilePath);// Session session = SftpClient.getSession(sftpConfig);// ChannelSftp channelSftp = SftpClient.getChannelSftp(session);fileExist = SftpClient.isFileExist(channelSftp, remoteFileDir, remoteFileName);// } catch (JSchException e) {// throw new RuntimeException(e);} catch (SftpException e){throw new RuntimeException(e);}File file = new File(localFilePathName);boolean exists = file.exists();if (exists){logger.info("当前文件已下载成功");return;}if (fileExist){SftpDownloadRequest sftpDownloadRequest = new SftpDownloadRequest();sftpDownloadRequest.setLocalFilePath(localFilePathName);sftpDownloadRequest.setRemoteFilePath(remoteFilePath);boolean result = SftpClient.downloadFile(sftpConfig, sftpDownloadRequest);if (result){HttpHeaders headers = new HttpHeaders();headers.setContentType(MediaType.APPLICATION_JSON_UTF8);HttpEntity httpEntity = new HttpEntity<>(headers);HttpEntity<String> response = restTemplate.exchange("http://" + ip + ":" + port + "/ods", HttpMethod.POST, httpEntity, String.class);String[] split = response.getHeaders().get("Location").get(0).split("/");String conI = split[split.length - 1];ImportRequest.Builder importBuilder = ImportRequest.newBuilder();File file1 = new File(localFilePath+"\\zip");file1.mkdir();ZipUtils.unZipIt(file.getAbsolutePath(),file1.getAbsolutePath());List<File> allFile = FileUtils.getAllFile(file1);String atfx = null;for (File file2 : allFile) {String[] split1 = file2.getName().split("\\.");String s = split1[split1.length - 1];if (s.equals("atfx")){atfx = file2.getAbsolutePath();}}importBuilder.setAtfx(atfx);byte[] param = importBuilder.build().toByteArray();HttpHeaders httpHeaders = new HttpHeaders();headers.setContentType(MediaType.valueOf("application/x-asamods+protobuf"));HttpEntity<byte[]> byteBody = new HttpEntity<>(param, httpHeaders);//param 加到 body 中ResponseEntity<ByteArrayResource> exchange = restTemplate.exchange("http://" + ip + ":" + port + "/ods/" + conI + "/import-atfx", HttpMethod.POST, byteBody, ByteArrayResource.class);if (exchange.getBody() == null){XxlJobHelper.handleSuccess();}else {XxlJobHelper.handleSuccess("ods接口");}/*byte[] byteArray = exchange.getBody().getByteArray();if (byteArray.length > 0){XxlJobHelper.handleSuccess();} else{XxlJobHelper.handleFail("ods返回的字节流为空");}*/} else{XxlJobHelper.handleFail("下载失败");}}}else {// XxlJobHelper.handleFail("没有文件，等待");}} catch (JSchException e) {throw new RuntimeException(e);}finally{channelSftp.disconnect();}} catch (JSchException e) {throw new RuntimeException(e);}finally{session.disconnect();}}@XxlJob("uploadJobHandler")public void uploadJobHandler() throws Exception {String jobParam = XxlJobHelper.getJobParam();String[] methodParams = jobParam.split(",");String localFilePath = methodParams[0];String[] split = localFilePath.split("\\\\");String remoteFilePath = "\\";remoteFilePath = remoteFilePath + split[split.length - 1];logger.info("本地完整文件名..[{}]", localFilePath);logger.info("sftp上完整文件名...[{}]", remoteFilePath);boolean fileExist = false;ChannelSftp channelSftp = null;Session session = null;try {String remoteFileDir = SftpClient.getRemoteFileDir(remoteFilePath);String remoteFileName = SftpClient.getRemoteFileName(remoteFilePath);session = SftpClient.getSession(sftpConfig);channelSftp = SftpClient.getChannelSftp(session);fileExist = SftpClient.isFileExist(channelSftp, remoteFileDir, remoteFileName);File file = new File(localFilePath);boolean exists = file.exists();if (fileExist) {logger.info("当前文件已上传成功");}if (exists) {SftpUploadRequest sftpUploadRequest = new SftpUploadRequest();sftpUploadRequest.setLocalFilePath(localFilePath);sftpUploadRequest.setRemoteFilePath(remoteFilePath);boolean result = SftpClient.uploadFile(sftpConfig, sftpUploadRequest);file.delete();if (result) {XxlJobHelper.handleSuccess();}}} catch (JSchException e) {throw new RuntimeException(e);} catch (SftpException e) {throw new RuntimeException(e);}finally{channelSftp.disconnect();session.disconnect();}}@XxlJob("shardingJobHandler")public void shardingJobHandler() throws Exception {// 分片参数int shardIndex = XxlJobHelper.getShardIndex();int shardTotal = XxlJobHelper.getShardTotal();XxlJobHelper.log("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);// 业务逻辑for (int i = 0; i < shardTotal; i++) {if (i == shardIndex) {XxlJobHelper.log("第 {} 片, 命中分片开始处理", i);} else {XxlJobHelper.log("第 {} 片, 忽略", i);}}}/*** 3、命令行任务*/@XxlJob("commandJobHandler")public void commandJobHandler() throws Exception {String command = XxlJobHelper.getJobParam();int exitValue = -1;BufferedReader bufferedReader = null;try {// command processProcessBuilder processBuilder = new ProcessBuilder();processBuilder.command(command);processBuilder.redirectErrorStream(true);Process process = processBuilder.start();//Process process = Runtime.getRuntime().exec(command);BufferedInputStream bufferedInputStream = new BufferedInputStream(process.getInputStream());bufferedReader = new BufferedReader(new InputStreamReader(bufferedInputStream));// command logString line;while ((line = bufferedReader.readLine()) != null) {XxlJobHelper.log(line);}// command exitprocess.waitFor();exitValue = process.exitValue();} catch (Exception e) {XxlJobHelper.log(e);} finally {if (bufferedReader != null) {bufferedReader.close();}}if (exitValue == 0) {// default success} else {XxlJobHelper.handleFail("command exit value(" + exitValue + ") is failed");}}/*** 4、跨平台Http任务* 参数示例：* "url: http://www.baidu.com\n" +* "method: get\n" +* "data: content\n";*/@XxlJob("httpJobHandler")public void httpJobHandler() throws Exception {// param parseString param = XxlJobHelper.getJobParam();if (param == null || param.trim().length() == 0) {XxlJobHelper.log("param[" + param + "] invalid.");XxlJobHelper.handleFail();return;}String[] httpParams = param.split("\n");String url = null;String method = null;String data = null;for (String httpParam : httpParams) {if (httpParam.startsWith("url:")) {url = httpParam.substring(httpParam.indexOf("url:") + 4).trim();}if (httpParam.startsWith("method:")) {method = httpParam.substring(httpParam.indexOf("method:") + 7).trim().toUpperCase();}if (httpParam.startsWith("data:")) {data = httpParam.substring(httpParam.indexOf("data:") + 5).trim();}}// param validif (url == null || url.trim().length() == 0) {XxlJobHelper.log("url[" + url + "] invalid.");XxlJobHelper.handleFail();return;}if (method == null || !Arrays.asList("GET", "POST").contains(method)) {XxlJobHelper.log("method[" + method + "] invalid.");XxlJobHelper.handleFail();return;}boolean isPostMethod = method.equals("POST");// requestHttpURLConnection connection = null;BufferedReader bufferedReader = null;try {// connectionURL realUrl = new URL(url);connection = (HttpURLConnection) realUrl.openConnection();// connection settingconnection.setRequestMethod(method);connection.setDoOutput(isPostMethod);connection.setDoInput(true);connection.setUseCaches(false);connection.setReadTimeout(5 * 1000);connection.setConnectTimeout(3 * 1000);connection.setRequestProperty("connection", "Keep-Alive");connection.setRequestProperty("Content-Type", "application/json;charset=UTF-8");connection.setRequestProperty("Accept-Charset", "application/json;charset=UTF-8");// do connectionconnection.connect();// dataif (isPostMethod && data != null && data.trim().length() > 0) {DataOutputStream dataOutputStream = new DataOutputStream(connection.getOutputStream());dataOutputStream.write(data.getBytes("UTF-8"));dataOutputStream.flush();dataOutputStream.close();}// valid StatusCodeint statusCode = connection.getResponseCode();if (statusCode != 200) {throw new RuntimeException("Http Request StatusCode(" + statusCode + ") Invalid.");}// resultbufferedReader = new BufferedReader(new InputStreamReader(connection.getInputStream(), "UTF-8"));StringBuilder result = new StringBuilder();String line;while ((line = bufferedReader.readLine()) != null) {result.append(line);}String responseMsg = result.toString();XxlJobHelper.log(responseMsg);return;} catch (Exception e) {XxlJobHelper.log(e);XxlJobHelper.handleFail();return;} finally {try {if (bufferedReader != null) {bufferedReader.close();}if (connection != null) {connection.disconnect();}} catch (Exception e2) {XxlJobHelper.log(e2);}}}/*** 5、生命周期任务示例：任务初始化与销毁时，支持自定义相关逻辑；*/@XxlJob(value = "demoJobHandler2", init = "init", destroy = "destroy")public void demoJobHandler2() throws Exception {XxlJobHelper.log("XXL-JOB, Hello World.");}public void init() {logger.info("init");}public void destroy() {logger.info("destory");}}
# web port server.port=8082 # no web #spring.main.web-environment=false# log config logging.config=classpath:logback.xml### xxl-job admin address list, such as "http://address" or "http://address01,http://address02" xxl.job.admin.addresses=http://127.0.0.1:2500/xxl-job-admin ### xxl-job, access token xxl.job.accessToken=### xxl-job executor appname xxl.job.executor.appname=xxl-job-executor-sftp ### xxl-job executor registry-address: default use address to registry , otherwise use ip:port if address is null xxl.job.executor.address= ### xxl-job executor server-info xxl.job.executor.ip= xxl.job.executor.port=10002 ### xxl-job executor log-path xxl.job.executor.logpath=/data/applogs/xxl-job/jobhandler ### xxl-job executor log-retention-days xxl.job.executor.logretentiondays=30#????????????????????? spring.main.allow-bean-definition-overriding=truesftp.host=localhost sftp.port=2022 sftp.username=zhou sftp.password=12138..xx sftp.identity= sftp.localFilePath=C:\\Users\\lenovo\\Desktop\\123ods.ip=localhost ods.port=3000tdm.ip=localhost tdm.port=3000
xxljob定时在sftp上传下载相关推荐
- SFTP上传下载文件
secureCRT SFTP上传/下载文件 远程登陆IP secureCRT会话中点击SFTP 3.cd /home/dowload linux平台切换到/home/dowload目录 ...
- CRT偶乐还是用到的-自带的sftp上传下载功能
使用CRT自带sftp功能 在CRT界面按快捷键 Alt + p打开一个新的sftp界面在sftp>界面输入"help"查看命令详情,左侧为命令,右侧为命令的说明.3.列举常 ...
- bat定时进行ftp上传下载文件
bat进行ftp上传下载文件 参考文章: https://blog.csdn.net/yongzai666/article/details/86488761 背景: 由于公司某个系统原本硬盘损坏 , ...
- C# 使用SFTP的上传下载文件时如何使用代理
最近遇到一个需求,使用SFTP上传下载文件,由于安全性,需要使用内部代理,在网上找了下,未找到相关代码.就自己整理了一份,实现原理基于 Tamir.SharpSsh.jsch; 部分代码仅供参考. ...
- 如何在Linux中使用sFTP上传或下载文件与文件夹
如何在Linux中使用sFTP上传或下载文件与文件夹 sFTP(安全文件传输程序)是一种安全的交互式文件传输程序,其工作方式与 FTP(文件传输协议)类似. 然而,sFTP 比 FTP 更安全;它通过 ...
- Java使用SFTP和FTP两种连接服务器的方式实现对文件的上传下载
一.Java实现对SFTP服务器的文件的上传下载: 1.添加maven依赖: <dependency><groupId>com.jcraft</groupId>&l ...
- uniapp 定时执行_ftp上传，完成ftp定时上传、下载只需3步
FTP[File Transfer Protocol]中文译为文件传输协议,是Internet上的另一项主要服务,这项服务让使用者能通过Internet来传输各式各样的文件.FTP上传是与WEB上传相 ...
- 能过ChilkatDotNet4.dll组件，开发SFTP，上传下载功能。
/// <summary>/// SFTP文件上传下载/// </summary>public class SFtp{/// <summary>/// http:/ ...
- sftp文件上传下载改名压缩解压
希望能帮到大家,有疑问联系 package com.qb.modules.organtrans.interactive.jinshang;import java.io.File; import jav ...
最新文章
- html从入门到卖电脑（六）
- Putty常用属性设置
- 最终在学校的职业发展目标
- OpenCASCADE：教程概述
- Games101现代图形学入门Lecture 4: Transformation Cont知识点总结
- ArcGIS for Desktop入门教程_第八章_Desktop学习资源 - ArcGIS知乎-新一代ArcGIS问答社区...
- 论文浅尝 | 探索将预训练语言模型用于事件抽取和事件生成
- 查看一个定义的方法在哪些地方被使用过(vs2008)
- 高效排错系列--摘要
- 遨博机器人展示_高交会：智能机器人走入大众生活
- 当当图书排行榜html,最-当当图书
- L314 单音节词读音规则（二）-元音字母发音规则
- beautify插件实现.wxml文件属性自动换行
- html是什么1003无标题,爱特漫画1003无标题
- springMVC+mybatis+maven搭建过程
- 把执行结果转成json对象报错_关于JSON转换成对象 报错LinkedHashMap不能直接转成对象...
- Python教程：函数多个返回值与丢弃返回值
- CC++ 输入十六进制字符转十六进制数
- ajax传json对象到后端，明明有数据，后端缺显示null
- PowerBuilder 进度条