innov
/
aggregation-platform
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge branch 'web_backend_develope' of

Browse Source 
https://git.trustie.net/fhx569287825/aggregation-platform into
web_backend_develope

Conflicts:
	src/com/platform/oracle/OracleConnector.java
web_backend_develope
chenlw 9 years ago
parent ad628fdc2a 4138b8bf6b
commit 2bd489878d
13 changed files with 1294 additions and 888 deletions
Show Stats Download Patch File Download Diff File

44
WebContent/WEB-INF/config/config.properties
Unescape View File

@ -83,30 +83,30 @@ oracle-psw=oracle
# windows path # windows path
#============================================================================================================= #=============================================================================================================
#extract-log-localtion=D:\\test\\log\\ extract-log-localtion=D:\\test\\log\\
#extract-standard-log-localtion=D:\\test\\log2\\ extract-standard-log-localtion=D:\\test\\log2\\
#
#file_upload_path=D:\\test\\
#file_download_path=D:\\test\\export.xlsx
#
#package_download_path=D:\\test\\
#package_name=sql_script_standard
#
#sql_script_path_last=D:\\test\\sql_script_last\\
#sql_script_path_standard=D:\\test\\sql_script_standard\\
#============================================================================================================= file_upload_path=D:\\test\\
# linux path file_download_path=D:\\test\\export.xlsx
#=============================================================================================================
extract-log-localtion=/home/web_manage/log/ package_download_path=D:\\test\\
extract-standard-log-localtion=/home/web_manage/log2/ package_name=sql_script_standard
file_upload_path=/excel_import_dir/ sql_script_path_last=D:\\test\\sql_script_last\\
file_download_path=/excel_export_dir/export.xlsx sql_script_path_standard=D:\\test\\sql_script_standard\\
package_download_path=/ #=============================================================================================================
package_name=DefaultDescription # linux path
#=============================================================================================================
sql_script_path_last=/DefaultDescription_last/ #extract-log-localtion=/home/web_manage/log/
sql_script_path_standard=/DefaultDescription/ #extract-standard-log-localtion=/home/web_manage/log2/
#
#file_upload_path=/excel_import_dir/
#file_download_path=/excel_export_dir/export.xlsx
#
#package_download_path=/
#package_name=DefaultDescription
#
#sql_script_path_last=/DefaultDescription_last/
#sql_script_path_standard=/DefaultDescription/

142
src/com/platform/controller/DataModelController.java
Unescape View File

@ -23,30 +23,32 @@ import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.bind.annotation.ResponseBody;
import com.base.BaseController; import com.base.BaseController;
import com.platform.entities.BasedTask;
import com.platform.entities.DataInfoEntity; import com.platform.entities.DataInfoEntity;
import com.platform.entities.DataInfoEntityMoveTmp; import com.platform.entities.DataInfoEntityMoveTmp;
import com.platform.entities.GatherOracleInfo; import com.platform.entities.GatherOracleInfo;
import com.platform.entities.OracleConnectorParams;
import com.platform.entities.OracleExtractExecuter;
import com.platform.form.PagerOptions; import com.platform.form.PagerOptions;
import com.platform.form.oracleForm; import com.platform.form.oracleForm;
import com.platform.form.volumeMoveForm; import com.platform.form.volumeMoveForm;
import com.platform.http.HttpUtils;
import com.platform.service.DataInfoService; import com.platform.service.DataInfoService;
import com.platform.service.ICodeService; import com.platform.service.ICodeService;
import com.platform.service.IGfsService; import com.platform.service.IGatherOracleService;
import com.platform.service.ILogRead; import com.platform.service.ILogRead;
import com.platform.service.IMoveDataService; import com.platform.service.IMoveDataService;
import com.platform.service.IGatherOracleService;
import com.platform.service.IOracleExtractService; import com.platform.service.IOracleExtractService;
import com.platform.service.IVolumeService; import com.platform.service.IVolumeService;
import com.platform.service.OracleExtractTask;
import com.platform.service.OracleStatusService; import com.platform.service.OracleStatusService;
import com.platform.service.thread.ThreadExtractOracle;
import com.platform.service.thread.ThreadGainOracleConnect;
import com.platform.utils.Constant; import com.platform.utils.Constant;
import com.platform.utils.UtilsHelper; import com.platform.utils.UtilsHelper;
/** 数据管理 /**
* 数据管理
*
* @author chen * @author chen
* *
*/ */
@Controller @Controller
public class DataModelController extends BaseController { public class DataModelController extends BaseController {
@ -94,12 +96,14 @@ public class DataModelController extends BaseController {
*/ */
@Resource(name = "logReadService") @Resource(name = "logReadService")
private ILogRead logReadService; private ILogRead logReadService;
public void setDfsImp(DataInfoService dfs) { public void setDfsImp(DataInfoService dfs) {
this.dfs = dfs; this.dfs = dfs;
} }
/** 数据管理--分页查询数据 /**
* 数据管理--分页查询数据
*
* @param res * @param res
* @param req * @param req
* @return * @return
@ -124,17 +128,19 @@ public class DataModelController extends BaseController {
sb.append(str).append(":").append("null").append(","); sb.append(str).append(":").append("null").append(",");
} }
} }
log.info(sb.deleteCharAt(sb.length() - 1) log.info(sb.deleteCharAt(sb.length() - 1).append("}").toString());
.append("}").toString());
PagerOptions pagerOptions = (PagerOptions) UtilsHelper PagerOptions pagerOptions = (PagerOptions) UtilsHelper
.newObjAndSetAttrsByClass(PagerOptions.class, params); .newObjAndSetAttrsByClass(PagerOptions.class, params);
pagerOptions.setCurrentPageNum(Integer.valueOf(params.get("currentPageNum"))); pagerOptions.setCurrentPageNum(Integer.valueOf(params
//冷热区查询字段mark .get("currentPageNum")));
// 冷热区查询字段mark
pagerOptions.setMark(pagerOptions.getVolumeType().trim()); pagerOptions.setMark(pagerOptions.getVolumeType().trim());
return dfs.getPagerTableData(pagerOptions); return dfs.getPagerTableData(pagerOptions);
} }
/** 数据管理--删除数据 /**
* 数据管理--删除数据
*
* @param res * @param res
* @param req * @param req
* @throws Exception * @throws Exception
@ -149,7 +155,9 @@ public class DataModelController extends BaseController {
req.setStatus(200); req.setStatus(200);
} }
/** 连接oracle /**
* 连接oracle
*
* @param res * @param res
* @param req * @param req
* @throws UnsupportedEncodingException * @throws UnsupportedEncodingException
@ -165,7 +173,8 @@ public class DataModelController extends BaseController {
log.info("执行连接\t" + rcName); log.info("执行连接\t" + rcName);
String cmd = "kubectl label --overwrite rc " + rcName String cmd = "kubectl label --overwrite rc " + rcName
+ " status=0"; + " status=0";
List<String> rList = Constant.ganymedSSH.execCmdWaitAcquiescent(cmd); List<String> rList = Constant.ganymedSSH
.execCmdWaitAcquiescent(cmd);
StringBuffer sb = new StringBuffer(); StringBuffer sb = new StringBuffer();
for (String string : rList) for (String string : rList)
sb.append(string).append("\n"); sb.append(string).append("\n");
@ -175,7 +184,9 @@ public class DataModelController extends BaseController {
req.setStatus(200); req.setStatus(200);
} }
/** 断开oracle连接 /**
* 断开oracle连接
*
* @param res * @param res
* @param req * @param req
* @throws UnsupportedEncodingException * @throws UnsupportedEncodingException
@ -196,7 +207,9 @@ public class DataModelController extends BaseController {
req.setStatus(200); req.setStatus(200);
} }
/** gfs的 volume节点的查询 /**
* gfs的 volume节点的查询
*
* @return * @return
* @throws Exception * @throws Exception
*/ */
@ -208,7 +221,9 @@ public class DataModelController extends BaseController {
return rest; return rest;
} }
/** 数据迁移功能 /**
* 数据迁移功能
*
* @param res * @param res
* @param req * @param req
* @param form * @param form
@ -233,7 +248,9 @@ public class DataModelController extends BaseController {
req.setStatus(200); req.setStatus(200);
} }
/** oracle的 汇总功能 /**
* oracle的 汇总功能
*
* @param res * @param res
* @param req * @param req
* @param form * @param form
@ -241,39 +258,32 @@ public class DataModelController extends BaseController {
*/ */
@RequestMapping(value = "/oracle/{name}/extract", method = RequestMethod.POST) @RequestMapping(value = "/oracle/{name}/extract", method = RequestMethod.POST)
public void oracleExtract(HttpServletRequest res, HttpServletResponse req, public void oracleExtract(HttpServletRequest res, HttpServletResponse req,
@RequestBody oracleForm form) throws Exception { @RequestBody oracleForm form) throws Exception {
log.error("/oracle/{name}/extract"); log.error("/oracle/{name}/extract");
boolean isConnect = false; if (null != form.getTarget()) { // 检查请求参数中是否包含汇总库信息
//5秒内是否能获得oracle连接，否则认为超时。 boolean isConnect = OracleExtract.isConnectTotalOracle(form
if (null != form.getTarget()) { .getTarget()); // 检查汇总库是否可以连接成功，连接成功返回200状态吗，连接失败返回500状态吗
ThreadGainOracleConnect thOrcl = new ThreadGainOracleConnect(form, OracleExtract); if (isConnect) {
thOrcl.start(); req.setStatus(200);
for (int i = 0; i < 10; i++) { if (null != form.getInneed() && form.getInneed().size() > 0) {
Thread.sleep(400); for (OracleConnectorParams oracleParams : form.getInneed()) {
isConnect = thOrcl.isConnect(); BasedTask task = new OracleExtractTask(
if (isConnect) { oracleParams.getName(), form.getTarget(),
break; oracleParams, OracleExtract);
} OracleExtractExecuter oee = new OracleExtractExecuter(
else { task);
if (thOrcl.isExcept()) { new Thread(oee, oracleParams.getName()).start();
break;
} }
Thread.sleep(100);
} }
} } else
} req.setStatus(500);
if (isConnect)
req.setStatus(200);
else
req.setStatus(500);
// 开始抽取数据到汇总库
if (isConnect && null != form.getInneed() && form.getInneed().size() > 0) {
ThreadExtractOracle thExtra = new ThreadExtractOracle(form, OracleExtract);
thExtra.start();
} }
} }
/** oracle汇总、抽取库的 查询 /**
* oracle汇总、抽取库的 查询
*
* @return * @return
* @throws Exception * @throws Exception
*/ */
@ -285,7 +295,9 @@ public class DataModelController extends BaseController {
return result; return result;
} }
/** oracle汇总、抽取库的 删除 /**
* oracle汇总、抽取库的 删除
*
* @param req * @param req
* @param res * @param res
* @param id * @param id
@ -301,7 +313,9 @@ public class DataModelController extends BaseController {
res.setStatus(200); res.setStatus(200);
} }
/** oracle汇总、抽取库的 新增 /**
* oracle汇总、抽取库的 新增
*
* @param res * @param res
* @param req * @param req
* @param id * @param id
@ -319,7 +333,9 @@ public class DataModelController extends BaseController {
req.setStatus(200); req.setStatus(200);
} }
/** oracle汇总、抽取库的 更新 /**
* oracle汇总、抽取库的 更新
*
* @param res * @param res
* @param req * @param req
* @param id * @param id
@ -337,7 +353,9 @@ public class DataModelController extends BaseController {
req.setStatus(200); req.setStatus(200);
} }
/** 迁移 数据 的查询 /**
* 迁移 数据 的查询
*
* @return * @return
* @throws Exception * @throws Exception
*/ */
@ -349,7 +367,9 @@ public class DataModelController extends BaseController {
return result; return result;
} }
/** 迁移完成后的 删除记录功能 /**
* 迁移完成后的 删除记录功能
*
* @param res * @param res
* @param req * @param req
* @param id * @param id
@ -368,7 +388,9 @@ public class DataModelController extends BaseController {
return result; return result;
} }
/** 迁移数据完成后新增一条数据（暂时去掉，新增功能不在此） /**
* 迁移数据完成后新增一条数据（暂时去掉，新增功能不在此）
*
* @param res * @param res
* @param req * @param req
* @param move * @param move
@ -381,12 +403,14 @@ public class DataModelController extends BaseController {
HttpServletResponse req, @RequestBody DataInfoEntity move) HttpServletResponse req, @RequestBody DataInfoEntity move)
throws Exception { throws Exception {
log.debug("---------/task/transfer/save-----------------------"); log.debug("---------/task/transfer/save-----------------------");
// int result = dfs.save(move); // int result = dfs.save(move);
req.setStatus(200); req.setStatus(200);
return 1; return 1;
} }
/** 地区和系统的 code 对应的名称 的 获取 /**
* 地区和系统的 code 对应的名称 的 获取
*
* @return * @return
* @throws Exception * @throws Exception
*/ */
@ -398,7 +422,9 @@ public class DataModelController extends BaseController {
return result; return result;
} }
/** oracle 汇总的 日志 读取 /**
* oracle 汇总的 日志 读取
*
* @param name * @param name
* @param res * @param res
* @param req * @param req
@ -410,8 +436,8 @@ public class DataModelController extends BaseController {
public Object getExtractLog(@RequestParam("rcName") String name, public Object getExtractLog(@RequestParam("rcName") String name,
HttpServletRequest res, HttpServletResponse req) throws Exception { HttpServletRequest res, HttpServletResponse req) throws Exception {
log.info("---------/oracle/extract/log-------------------"); log.info("---------/oracle/extract/log-------------------");
String result = logReadService.readLog(name); String result = logReadService.readLog(name);
// "查看相应日志" // "查看相应日志"
Map<String, String> log = new HashMap<String, String>(); Map<String, String> log = new HashMap<String, String>();
log.put(name, result); log.put(name, result);
return log; return log;

32
src/com/platform/entities/AbstractOracleExtractTask.java
Unescape View File

@ -0,0 +1,32 @@
package com.platform.entities;
public abstract class AbstractOracleExtractTask implements BasedTask {
private String name;
private int status; // 任务的执行状态,0未执行，1执行中，2完成，3失败
public AbstractOracleExtractTask() {
}
public AbstractOracleExtractTask(String name) {
this.name = name;
}
public abstract void handler();
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getStatus() {
return status;
}
public void setStatus(int status) {
this.status = status;
}
}

5
src/com/platform/entities/BasedTask.java
Unescape View File

@ -0,0 +1,5 @@
package com.platform.entities;
public interface BasedTask {
public void handler();
}

15
src/com/platform/entities/OracleExtractExecuter.java
Unescape View File

@ -0,0 +1,15 @@
package com.platform.entities;
public class OracleExtractExecuter implements Runnable {
private BasedTask task;
public OracleExtractExecuter(BasedTask task) {
this.task = task;
}
@Override
public void run() {
if (null != task)
task.handler();
}
}

5
src/com/platform/kubernetes/BasedPathEntity.java
Unescape View File

@ -0,0 +1,5 @@
package com.platform.kubernetes;
public class BasedPathEntity {
}

25
src/com/platform/kubernetes/SimpleKubeClient.java
Unescape View File

@ -4,6 +4,8 @@ import io.fabric8.kubernetes.api.model.Container;
import io.fabric8.kubernetes.api.model.ContainerPort; import io.fabric8.kubernetes.api.model.ContainerPort;
import io.fabric8.kubernetes.api.model.Pod; import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.ReplicationController; import io.fabric8.kubernetes.api.model.ReplicationController;
import io.fabric8.kubernetes.api.model.Volume;
import io.fabric8.kubernetes.api.model.VolumeMount;
import io.fabric8.kubernetes.client.Client; import io.fabric8.kubernetes.client.Client;
import io.fabric8.kubernetes.client.Config; import io.fabric8.kubernetes.client.Config;
import io.fabric8.kubernetes.client.ConfigBuilder; import io.fabric8.kubernetes.client.ConfigBuilder;
@ -156,4 +158,27 @@ public class SimpleKubeClient {
client.close(); client.close();
} }
/**
*
* @param rcId --> 资源Id
* @param ns --> 资源名字空间
* @return
*/
public ReplicationController createReplicas(String rcId, String ns) {
if (checkClientNull())
client = new DefaultKubernetesClient(Configs.KUBE_MASTER_URL);
ReplicationController rc = client.replicationControllers()
.inNamespace(ns).createNew().editMetadata().withName(rcId)
.endMetadata().done();
return rc;
}
public ReplicationController createReplicas(String rcId, String ns, Map<String, String> selector,Volume volume, VolumeMount volumeMounts){
if (checkClientNull())
client = new DefaultKubernetesClient(Configs.KUBE_MASTER_URL);
//Container container = new Container("", "", "", "", "", null, null, "oracle", ports, null, null, null, false, false, null, false, volumeMounts, null);
//container.setName("oracle");
//ReplicationController rController = client.replicationControllers().inNamespace(ns).createNew().editMetadata().withName(rcId).withLabels(selector).endMetadata().editSpec().withReplicas(1).withSelector(selector).editTemplate().editMetadata().withLabels(selector).endMetadata().editSpec().addNewContainer().a.
return null;
}
} }

357
src/com/platform/oracle/OracleConnector.java
Unescape View File

@ -1,175 +1,184 @@
package com.platform.oracle; package com.platform.oracle;
import java.sql.Connection; import java.sql.Connection;
import java.sql.DriverManager; import java.sql.DriverManager;
import java.sql.ResultSet; import java.sql.ResultSet;
import java.sql.SQLException; import java.sql.SQLException;
import java.sql.Statement; import java.sql.Statement;
import org.apache.log4j.Logger; import org.apache.log4j.Logger;
import com.base.Custom4exception; import com.base.Custom4exception;
import com.base.CustomException; import com.base.CustomException;
import com.platform.entities.OracleConnectorParams; import com.platform.entities.OracleConnectorParams;
import com.platform.utils.Configs; import com.platform.utils.Configs;
import com.platform.utils.FileOperateHelper; import com.platform.utils.FileOperateHelper;
public class OracleConnector { public class OracleConnector {
public static Logger log = Configs.DAILY_ROLLING_LOGGER.getLogger(OracleConnector.class); public static Logger log = Configs.DAILY_ROLLING_LOGGER
.getLogger(OracleConnector.class);
public OracleConnector() {
} public OracleConnector() {
}
static {
try { static {
Class.forName("oracle.jdbc.driver.OracleDriver"); try {
Configs.CONSOLE_LOGGER.info("Oracle驱动加载成功"); Class.forName("oracle.jdbc.driver.OracleDriver");
} catch (ClassNotFoundException e) { Configs.CONSOLE_LOGGER.info("Oracle驱动加载成功");
log.error(Custom4exception.OracleSQL_Except, e); } catch (ClassNotFoundException e) {
} log.error(Custom4exception.OracleSQL_Except, e);
} }
}
public synchronized static Connection connectionBuilder(String url, String user, public synchronized static Connection connectionBuilder(String url,
String password, OracleConnectorParams oc) throws CustomException { String user, String password, OracleConnectorParams oc)
Connection conn=null; throws CustomException {
try { Connection conn = null;
conn = DriverManager.getConnection(url, user, password); try {
} catch (SQLException e) { conn = DriverManager.getConnection(url, user, password);
Configs.CONSOLE_LOGGER.info("创建oracle连接失败: [" + e.getMessage() + "]"); } catch (SQLException e) {
if (null != oc) { Configs.CONSOLE_LOGGER.info("创建oracle连接失败: [" + e.getMessage()
FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION + "]");
+ oc.getName()+".log", "创建oracle连接失败: [" + e.getMessage() + "]\r\n"); if (null != oc) {
} FileOperateHelper.fileWrite(
throw new CustomException(Custom4exception.OracleSQL_Except, e); Configs.EXTRACT_LOG_LOCALTION + oc.getName() + ".log",
} "创建oracle连接失败: [" + e.getMessage() + "]\r\n");
return conn; }
} throw new CustomException(Custom4exception.OracleSQL_Except, e);
}
public synchronized static boolean canConnect(String url, String user, String password) { return conn;
Connection result = null; }
try {
result = connectionBuilder(url, user, password, null); public synchronized static boolean canConnect(String url, String user,
} catch (CustomException e) { String password) {
log.error(Custom4exception.OracleSQL_Except, e); Connection result = null;
} try {
return (null != result); result = connectionBuilder(url, user, password, null);
} } catch (CustomException e) {
log.error(Custom4exception.OracleSQL_Except, e);
public synchronized static ResultSet getSQLExecResultSet(Connection conn, String sql, String filePath) { }
ResultSet resultSet = null; return (null != result);
if (null != filePath) { }
filePath = filePath.replace(".log", "");
} public synchronized static ResultSet getSQLExecResultSet(Connection conn,
Statement statement = null; String sql, String filePath) {
try { ResultSet resultSet = null;
statement = conn if (null != filePath) {
.createStatement(ResultSet.TYPE_SCROLL_SENSITIVE, filePath = filePath.replace(".log", "");
ResultSet.CONCUR_UPDATABLE); }
resultSet = statement.executeQuery(sql); Statement statement = null;
try {
FileOperateHelper statement = conn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE,
.fileWrite(filePath+".log", sql+ "\r\n"+ "OK \r\n"); ResultSet.CONCUR_UPDATABLE);
} catch (SQLException e) { resultSet = statement.executeQuery(sql);
FileOperateHelper /*
.fileWrite(filePath+".log", sql+ "\r\n"+e.getMessage()+"\r\n"); * if(resultSet.next()){ System.out.println(resultSet.getInt(1)); }
log.error(Custom4exception.OracleSQL_Except, e); */
} FileOperateHelper.fileWrite(filePath + ".log", sql + "\r\n"
finally{ + "OK \r\n");
if (null != statement) { } catch (SQLException e) {
try { FileOperateHelper.fileWrite(filePath + ".log",
statement.close(); sql + "\r\n" + e.getMessage() + "\r\n");
} catch (SQLException e) { log.error(Custom4exception.OracleSQL_Except, e);
log.error(Custom4exception.OracleSQL_Except, e); } finally {
} if (null != statement) {
} try {
} statement.close();
} catch (SQLException e) {
return resultSet; log.error(Custom4exception.OracleSQL_Except, e);
} }
}
public synchronized static ResultSet getSQLExecResultSet(String url, String user, }
String password, String sql, String filePath) {
ResultSet result = null; return resultSet;
try { }
result = getSQLExecResultSet(connectionBuilder(url, user, password, null), sql, filePath);
} catch (CustomException e) { public synchronized static ResultSet getSQLExecResultSet(String url,
log.error(Custom4exception.OracleSQL_Except, e); String user, String password, String sql, String filePath) {
} ResultSet result = null;
return result; try {
} result = getSQLExecResultSet(
connectionBuilder(url, user, password, null), sql, filePath);
/** } catch (CustomException e) {
* 执行对oracle数据库的增、删 log.error(Custom4exception.OracleSQL_Except, e);
* @param conn }
* @param sql return result;
* @return true:执行的不返回集合数据的sql成功， 是否执行成功 }
*/
public synchronized static boolean execOracleSQL(Connection conn, String sql, String filePath) { /**
if (null != filePath) { * 执行对oracle数据库的增、删
filePath = filePath.replace(".log", ""); *
} * @param conn
boolean flag = false; * @param sql
Statement statement = null; * @return true:执行的不返回集合数据的sql成功， 是否执行成功
try { */
statement = conn.createStatement(); public synchronized static boolean execOracleSQL(Connection conn,
flag =statement.execute(sql); String sql, String filePath) {
FileOperateHelper if (null != filePath) {
.fileWrite(filePath+".log", sql+ "\r\n"+ flag +" \r\n"); filePath = filePath.replace(".log", "");
} catch (SQLException e) { }
flag = false; boolean flag = false;
FileOperateHelper Statement statement = null;
.fileWrite(filePath+".log", sql+ "\r\n"+e.getMessage()+"\r\n"); try {
log.error(Custom4exception.OracleSQL_Except, e); statement = conn.createStatement();
} statement.executeUpdate(sql);
finally{ flag = true;
if (null != statement) { FileOperateHelper.fileWrite(filePath + ".log", sql + "\r\n" + flag
try { + " \r\n");
statement.close(); } catch (SQLException e) {
} catch (SQLException e) { flag = false;
log.error(Custom4exception.OracleSQL_Except, e); FileOperateHelper.fileWrite(filePath + ".log",
} sql + "\r\n" + e.getMessage() + "\r\n");
} log.error(Custom4exception.OracleSQL_Except, e);
} } finally {
if (null != statement) {
return flag; try {
} statement.close();
/** } catch (SQLException e) {
* 执行对oracle数据库的返回集合数据的sql log.error(Custom4exception.OracleSQL_Except, e);
* @param conn }
* @param sql }
* @return true:执行结果大于1，即有数据 是否执行成功 }
*/
public synchronized static boolean execUpdateOracleSQL(Connection conn, String sql, String filePath) { return flag ;
if (null != filePath) { }
filePath = filePath.replace(".log", "");
} /**
boolean flag = false; * 执行对oracle数据库的返回集合数据的sql
Statement statement = null; *
try { * @param conn
statement = conn.createStatement(); * @param sql
if(statement.executeUpdate(sql) > 0) * @return true:执行结果大于1，即有数据 是否执行成功
{ */
flag = true; public synchronized static boolean execUpdateOracleSQL(Connection conn,
FileOperateHelper String sql, String filePath) {
.fileWrite(filePath+".log", sql+ "\r\n"+"OK \r\n"); if (null != filePath) {
} filePath = filePath.replace(".log", "");
} catch (SQLException e) { }
flag = false; boolean flag = false;
FileOperateHelper Statement statement = null;
.fileWrite(filePath+".log", sql+ "\r\n"+e.getMessage()+"\r\n"); try {
log.error(Custom4exception.OracleSQL_Except, e); statement = conn.createStatement();
} statement.executeUpdate(sql);
finally{ flag = true;
if (null != statement) { FileOperateHelper.fileWrite(filePath + ".log", sql + "\r\n"
try { + "OK \r\n");
statement.close();
} catch (SQLException e) { } catch (SQLException e) {
log.error(Custom4exception.OracleSQL_Except, e); flag = false;
} FileOperateHelper.fileWrite(filePath + ".log",
} sql + "\r\n" + e.getMessage() + "\r\n");
} log.error(Custom4exception.OracleSQL_Except, e);
return flag; } finally {
} if (null != statement) {
try {
statement.close();
} catch (SQLException e) {
log.error(Custom4exception.OracleSQL_Except, e);
}
}
}
return flag;
}
} }

6
src/com/platform/service/IOracleExtractService.java
Unescape View File

@ -32,5 +32,11 @@ public interface IOracleExtractService {
*/ */
boolean extractStandardTable(String name, List<OracleConnectorParams> dataInfolist, boolean extractStandardTable(String name, List<OracleConnectorParams> dataInfolist,
GatherOracleInfo oracleConnect) throws Exception; GatherOracleInfo oracleConnect) throws Exception;
/**
*
* @param ocp
*/
public void updateDataExtractStatus(OracleConnectorParams ocp, int status);
} }

763
src/com/platform/service/OracleExtractHelper.java
Unescape View File

@ -1,334 +1,429 @@
package com.platform.service; package com.platform.service;
import java.sql.Connection; import java.sql.Connection;
import java.sql.ResultSet; import java.sql.ResultSet;
import java.sql.SQLException; import java.sql.SQLException;
import java.sql.Statement; import java.sql.Statement;
import org.apache.log4j.Logger; import org.apache.log4j.Logger;
import com.base.Custom4exception; import com.base.Custom4exception;
import com.platform.entities.GatherOracleInfo; import com.platform.entities.GatherOracleInfo;
import com.platform.entities.OracleConnectorParams; import com.platform.entities.OracleConnectorParams;
import com.platform.oracle.OracleConnector; import com.platform.oracle.OracleConnector;
import com.platform.utils.Configs; import com.platform.utils.Configs;
import com.platform.utils.FileOperateHelper; import com.platform.utils.FileOperateHelper;
public class OracleExtractHelper { public class OracleExtractHelper {
public static Logger log = Configs.DAILY_ROLLING_LOGGER.getLogger(OracleExtractHelper.class); public static Logger log = Logger.getLogger(OracleExtractHelper.class);
/** /**
* 判断dblink是否已经存在 * 判断dblink是否已经存在
* *
* @param conn * @param conn
* @param linkName * @param linkName
* dblink的名称 * dblink的名称
* @return * @return
*/ */
private boolean hasSameNameDBLink(Connection conn, String linkName, String filePath) { private boolean hasSameNameDBLink(Connection conn, String linkName,
String filePath) {
boolean flag = false;
String sql = "SELECT * FROM ALL_DB_LINKS WHERE DB_LINK='" + linkName+"'"; boolean flag = false;
ResultSet rSet = null; String sql = "SELECT COUNT(*) c FROM ALL_DB_LINKS WHERE DB_LINK='"
try { + linkName + "'";
rSet = OracleConnector.getSQLExecResultSet(conn, sql, null); ResultSet rSet = null;
rSet.last(); Statement statement = null;
if (rSet.getRow() > 0) try {
flag = true; statement = conn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE,
FileOperateHelper ResultSet.CONCUR_UPDATABLE);
.fileWrite(filePath, sql+ "\r\n"+"OK \r\n"); rSet = statement.executeQuery(sql);
} catch (Exception e) { if (rSet.next()) {
FileOperateHelper if (rSet.getInt("c") > 0)
.fileWrite(filePath, sql+ "\r\n"+ e.getMessage() + "\r\n，连接异常！ \r\n"); flag = true;
log.error(Custom4exception.threadVolume_Oracle_Except, e); else
} flag = false;
return flag; }
} FileOperateHelper.fileWrite(filePath + ".log", sql + "\r\n"
+ "OK \r\n");
/**
* 创建dblink } catch (Exception e) {
* FileOperateHelper.fileWrite(filePath, sql + "\r\n" + e.getMessage()
* @param conn 汇总库的连接 + "\r\n，连接异常！ \r\n");
* @param oc 采集库 log.error(Custom4exception.threadVolume_Oracle_Except, e);
* dblink连接参数实体 } finally {
*/ if (null != statement) {
public void createDBLink(Connection conn, OracleConnectorParams oc) { try {
String linkName = "LINKTO" + oc.getName(); statement.close();
String sql = "CREATE PUBLIC DATABASE LINK " } catch (SQLException e) {
+ linkName log.error(Custom4exception.OracleSQL_Except, e);
+ " CONNECT TO " }
+ oc.getUser() }
+ " IDENTIFIED BY " }
+ oc.getPassword() return flag;
+ " USING \'(DESCRIPTION =(ADDRESS_LIST =(ADDRESS = (PROTOCOL = TCP)(HOST = " }
+ oc.getIp() + ")(PORT = " + oc.getPort()
+ ")))(CONNECT_DATA =(SERVICE_NAME =" + oc.getDatabaseName() /**
+ ")))\'"; * 创建dblink
if (null != oc) { *
if (hasSameNameDBLink(conn, linkName, Configs.EXTRACT_LOG_LOCALTION + oc.getName()+".log")) { // 如果dblink已经存在,先删除dblink，在创建dblink * @param conn
String deleteSql = "DROP PUBLIC DATABASE LINK " * 汇总库的连接
+ linkName; * @param oc
//删除 dblink * 采集库 dblink连接参数实体
if (OracleConnector.execOracleSQL(conn, deleteSql, Configs.EXTRACT_LOG_LOCALTION + oc.getName())) { */
if(!OracleConnector.execOracleSQL(conn, sql, Configs.EXTRACT_LOG_LOCALTION + oc.getName())){ public void createDBLink(Connection conn, OracleConnectorParams oc) {
OracleConnector.execUpdateOracleSQL(conn, sql, Configs.EXTRACT_LOG_LOCALTION + oc.getName()); String linkName = "LINKTO_J" + oc.getName().replaceAll("-", "_");
} String sql = "CREATE PUBLIC DATABASE LINK "
} else { + linkName
Configs.CONSOLE_LOGGER.error("删除已有的DBLink失败,无法创建新的DBLink!"); + " CONNECT TO "
FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION + oc.getUser()
+ oc.getName()+".log", "删除已有的DBLink失败,无法创建新的DBLink!"+"\r\n"); + " IDENTIFIED BY "
} + oc.getPassword()
+ " USING \'(DESCRIPTION =(ADDRESS_LIST =(ADDRESS = (PROTOCOL = TCP)(HOST = "
} else { + oc.getIp() + ")(PORT = " + oc.getPort()
// 否则，创建dblink + ")))(CONNECT_DATA =(SERVICE_NAME =" + oc.getDatabaseName()
if(!OracleConnector.execOracleSQL(conn, sql, Configs.EXTRACT_LOG_LOCALTION + oc.getName())){ + ")))\'";
OracleConnector.execUpdateOracleSQL(conn, sql, Configs.EXTRACT_LOG_LOCALTION + oc.getName()); if (null != oc) {
} if (hasSameNameDBLink(conn, linkName, Configs.EXTRACT_LOG_LOCALTION
} + oc.getName())) { // 如果dblink已经存在,先删除dblink，在创建dblink
} String deleteSql = "DROP PUBLIC DATABASE LINK " + linkName;
} // 删除 dblink
if (OracleConnector.execOracleSQL(conn, deleteSql,
/** Configs.EXTRACT_LOG_LOCALTION + oc.getName())) {
* 创建表空间 if (!OracleConnector.execOracleSQL(conn, sql,
* Configs.EXTRACT_LOG_LOCALTION + oc.getName())) {
* @param conn 汇总库连接 OracleConnector.execUpdateOracleSQL(conn, sql,
* @param collectOracle Configs.EXTRACT_LOG_LOCALTION + oc.getName());
* @param oc 汇总库信息 }
* @return } else {
*/ Configs.CONSOLE_LOGGER.error("删除已有的DBLink失败,无法创建新的DBLink!");
public boolean createTableSpace(Connection conn, OracleConnectorParams collectOracle, GatherOracleInfo oc) { FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION
String tmpSql = "select TABLESPACE_NAME from dba_tablespaces where TABLESPACE_NAME = '" + oc.getName() + ".log",
+ oc.getTableName() + "'"; "删除已有的DBLink失败,无法创建新的DBLink!" + "\r\n");
// 存在 表空间 }
if (OracleConnector.execUpdateOracleSQL(conn, tmpSql, Configs.EXTRACT_LOG_LOCALTION + collectOracle.getName())) {
return true; } else {
} else { // 否则，创建dblink
String sql = "create tablespace " + oc.getTableName() if (!OracleConnector.execOracleSQL(conn, sql,
+ " datafile '" + Configs.GATHER_TABLESPACE_PATH Configs.EXTRACT_LOG_LOCALTION + oc.getName())) {
+ oc.getTableName() + ".dbf" OracleConnector.execUpdateOracleSQL(conn, sql,
+ "' size 512M autoextend on next 512M maxsize unlimited"; Configs.EXTRACT_LOG_LOCALTION + oc.getName());
return OracleConnector.execOracleSQL(conn, sql, Configs.EXTRACT_LOG_LOCALTION + collectOracle.getName()); }
} }
} }
}
/**
* 创建用户并授权 /**
* * 创建表空间
* @param conn 汇总库连接 *
* @param oc 汇总库信息 * @param conn
*/ * 汇总库连接
public void createUser(Connection conn, OracleConnectorParams oc, GatherOracleInfo totalOracle) { * @param collectOracle
String strTUser = oc.getName() + totalOracle.getSuffix(); * @param oc
* 汇总库信息
String sql = "Create user " + strTUser + " default tablespace " * @return
+ totalOracle.getTableName() + " identified by " */
+ Configs.GATHER_TABLE_PASSWORD; public boolean createTableSpace(Connection conn,
String grantSql = "grant connect, resource, dba to " + strTUser; OracleConnectorParams collectOracle, GatherOracleInfo oc) {
String tmpSql = "select TABLESPACE_NAME from dba_tablespaces where TABLESPACE_NAME = '"
OracleConnector.execOracleSQL(conn, sql, Configs.EXTRACT_LOG_LOCALTION + oc.getName()); + oc.getTableName() + "'";
// 存在 表空间
OracleConnector.execOracleSQL(conn, grantSql, Configs.EXTRACT_LOG_LOCALTION + oc.getName()); if (OracleConnector.execUpdateOracleSQL(conn, tmpSql,
} Configs.EXTRACT_LOG_LOCALTION + collectOracle.getName())) {
return true;
/** } else {
* 创建用户并授权 String sql = "create tablespace " + oc.getTableName()
* + " datafile '" + Configs.GATHER_TABLESPACE_PATH
* @param conn 汇总库连接 + oc.getTableName() + ".dbf"
* @param oc 汇总库信息 + "' size 512M autoextend on next 512M maxsize unlimited";
*/ return OracleConnector.execOracleSQL(conn, sql,
public void createOnlyUser(Connection conn, OracleConnectorParams oc, GatherOracleInfo totalOracle) { Configs.EXTRACT_LOG_LOCALTION + collectOracle.getName());
String strTUser = Configs.GATHER_STANDARD_USER_NAME; }
}
String sql = "Create user " + strTUser + " default tablespace "
+ totalOracle.getTableName() + " identified by " /**
+ Configs.GATHER_TABLE_PASSWORD; * 创建用户并授权
String grantSql = "grant connect, resource, dba to " + strTUser; *
* @param conn
OracleConnector.execOracleSQL(conn, sql, Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName()); * 汇总库连接
* @param oc
OracleConnector.execOracleSQL(conn, grantSql, Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName()); * 汇总库信息
} */
public void createUser(Connection conn, OracleConnectorParams oc,
/** GatherOracleInfo totalOracle) {
* 创建用户并授权 String strTUser = "J" + oc.getName().replaceAll("-", "_")
* + totalOracle.getSuffix();
* @param conn 汇总库连接
* @param oc 汇总库信息 String sql = "Create user " + strTUser + " default tablespace "
*/ + totalOracle.getTableName() + " identified by "
public void createTable(Connection conn, OracleConnectorParams oc, GatherOracleInfo totalOracle) { + Configs.GATHER_TABLE_PASSWORD;
String strTUser = oc.getName() + totalOracle.getSuffix(); String grantSql = "grant connect, resource, dba to " + strTUser;
String sql = "Create table " + strTUser + " default tablespace " OracleConnector.execOracleSQL(conn, sql, Configs.EXTRACT_LOG_LOCALTION
+ totalOracle.getTableName() + " identified by " + oc.getName());
+ Configs.GATHER_TABLE_PASSWORD;
String grantSql = "grant connect, resource, dba to " + strTUser; OracleConnector.execOracleSQL(conn, grantSql,
Configs.EXTRACT_LOG_LOCALTION + oc.getName());
OracleConnector.execOracleSQL(conn, sql, Configs.EXTRACT_LOG_LOCALTION + oc.getName()); }
OracleConnector.execOracleSQL(conn, grantSql, Configs.EXTRACT_LOG_LOCALTION + oc.getName()); /**
} * 创建用户并授权
*
/** * @param conn
* 执行汇总操作 * 汇总库连接
* * @param oc
* @param conn 汇总库连接 * 汇总库信息
* @param oc 采集库 */
*/ public void createOnlyUser(Connection conn, OracleConnectorParams oc,
public void extractColleDB(Connection conn, OracleConnectorParams oc, GatherOracleInfo totalOracle) { GatherOracleInfo totalOracle) {
String strTUser = oc.getName() + totalOracle.getSuffix(); String strTUser = Configs.GATHER_STANDARD_USER_NAME;
String sql = "SELECT 'create table "
+ strTUser String sql = "Create user " + strTUser + " default tablespace "
+ ".J'|| substr(t.OWNER||'_'||t.TABLE_NAME,0,29)||' as select * from '||t.OWNER||'.'||t.TABLE_NAME||" + totalOracle.getTableName() + " identified by "
+ "'@LINKTO" + Configs.GATHER_TABLE_PASSWORD;
+ oc.getName() String grantSql = "grant connect, resource, dba to " + strTUser;
+ ";' FROM dba_tables@LINKTO"
+ oc.getName() OracleConnector.execOracleSQL(conn, sql,
+ " t WHERE t.TABLESPACE_NAME NOT IN ('SYSTEM','SYSAUX')" Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName());
+ " and t.owner||t.table_name not in (select owner||table_name from dba_tables@LINKTO"
+ oc.getName() + " where 'data_type'='CLOB')"; OracleConnector.execOracleSQL(conn, grantSql,
ResultSet rsSet = OracleConnector.getSQLExecResultSet(conn, sql, Configs.EXTRACT_LOG_LOCALTION + oc.getName()); Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName());
try { }
while (rsSet.next()) {
try { /**
rsSet.getRow(); * 创建用户并授权
String resultSql = rsSet.getString(1).replace(";", ""); *
OracleConnector.execUpdateOracleSQL(conn, resultSql, Configs.EXTRACT_LOG_LOCALTION + oc.getName()); * @param conn
} catch (SQLException e) { * 汇总库连接
log.error(Custom4exception.threadVolume_Oracle_Except, e); * @param oc
} * 汇总库信息
} */
} catch (SQLException e) { public void createTable(Connection conn, OracleConnectorParams oc,
log.error(Custom4exception.threadVolume_Oracle_Except, e); GatherOracleInfo totalOracle) {
} String strTUser = oc.getName() + totalOracle.getSuffix();
} String sql = "Create table " + strTUser + " default tablespace "
+ totalOracle.getTableName() + " identified by "
/** 测试是否能否连接上？ + Configs.GATHER_TABLE_PASSWORD;
* @return true：能连接上 String grantSql = "grant connect, resource, dba to " + strTUser;
*/
public boolean testConnect(Connection conn) { OracleConnector.execOracleSQL(conn, sql, Configs.EXTRACT_LOG_LOCALTION
String testSql = "select count(*) from user_tables"; + oc.getName());
boolean flag = false;
try { OracleConnector.execOracleSQL(conn, grantSql,
Statement statement = conn.createStatement(); Configs.EXTRACT_LOG_LOCALTION + oc.getName());
if(statement.executeUpdate(testSql) > 0) }
flag = true;
} catch (SQLException e) { /**
flag = false; * 执行汇总操作
log.error(Custom4exception.threadVolume_Oracle_Except, e); *
} * @param conn
return flag; * 汇总库连接
} * @param oc
* 采集库
/** */
* 执行抽取操作--支付表 public void extractColleDB(Connection conn, OracleConnectorParams oc,
* GatherOracleInfo totalOracle) {
* @param conn 汇总库连接 String strTUser = "J" + oc.getName().replaceAll("-", "_")
* @param oc 采集库 + totalOracle.getSuffix();
*/ String sql = "SELECT 'create table "
public void extractStandardPayTable(Connection conn, OracleConnectorParams oc, GatherOracleInfo totalOracle) { + strTUser
String strTUser = Configs.GATHER_STANDARD_USER_NAME; + ".J'|| substr(t.OWNER||'_'||t.TABLE_NAME,0,29)||' as select * from '||t.OWNER||'.'||t.TABLE_NAME||"
createPay(conn, oc); + "'@LINKTO_J"
String sql = "insert into " + strTUser + "." + Configs.GATHER_STANDARD_PAY_TABLE_NAME + " select * from " + oc.getName().replaceAll("-", "_")
+ strTUser + "." + Configs.GATHER_STANDARD_PAY_TABLE_NAME +"@LINKTOST" + ";' FROM dba_tables@LINKTO_J"
+ oc.getName(); + oc.getName().replaceAll("-", "_")
String resultSql = sql.replace(";", ""); + " t WHERE t.TABLESPACE_NAME NOT IN ('SYSTEM','SYSAUX')"
OracleConnector.execOracleSQL(conn, resultSql, Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName()); + " and t.owner||t.table_name not in (select owner||table_name from dba_tables@LINKTO_J"
} + oc.getName().replaceAll("-", "_")
+ " where 'data_type'='CLOB')";
/** 执行抽取操作--执行表 /*
* @param conn * ResultSet rsSet = OracleConnector.getSQLExecResultSet(conn, sql,
* @param collectOracle * Configs.EXTRACT_LOG_LOCALTION + oc.getName());
* @param oracleModel */
*/ Statement statement = null;
public void extractStandardExecTable(Connection conn, try {
OracleConnectorParams collectOracle, GatherOracleInfo oracleModel) { statement = conn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE,
createExec(conn, collectOracle); ResultSet.CONCUR_UPDATABLE);
String strTUser = Configs.GATHER_STANDARD_USER_NAME; ResultSet resultSet = statement.executeQuery(sql);
FileOperateHelper.fileWrite(
String sql = "insert into " + strTUser + "." + Configs.GATHER_STANDARD_EXEC_TABLE_NAME + " select * from " Configs.EXTRACT_LOG_LOCALTION + oc.getName() + ".log", sql
+ strTUser + "." + Configs.GATHER_STANDARD_EXEC_TABLE_NAME +"@LINKTOST" + "\r\n" + "OK \r\n");
+ collectOracle.getName();
String resultSql = sql.replace(";", ""); while (resultSet.next()) {
OracleConnector.execOracleSQL(conn, resultSql, Configs.EXTRACT_STANDARD_LOG_LOCALTION + collectOracle.getName()); try {
String resultSql = resultSet.getString(1).replace(";", "");
} OracleConnector.execUpdateOracleSQL(conn, resultSql,
Configs.EXTRACT_LOG_LOCALTION + oc.getName());
} catch (SQLException e) {
log.error(Custom4exception.threadVolume_Oracle_Except, e);
private void createPay(Connection conn, OracleConnectorParams oc) { }
String payCmd = "CREATE TABLE u_bzbjy.zfxxb(XZQHDM Varchar(255),XZQHMC Varchar(255),PZBH Varchar(255),LYZBKZH Varchar(255)," }
+ "ZFDATE Varchar(255),YSDWCODE Varchar(255),YSDWNAME Varchar(255),YWGKCS Varchar(255),XMCODE Varchar(255),XMNAME Varchar(255)," } catch (SQLException e) {
+"XMLBCODE Varchar(255),XMLBNAME Varchar(255),ZB_NO Varchar(255),GNFLCODE Varchar(255),GNFLNAME Varchar(255),JJFLCODE Varchar(255)," log.error(Custom4exception.threadVolume_Oracle_Except, e);
+"JJFLNAME Varchar(255),ZJXZCODE Varchar(255),ZJXZNAME Varchar(255),JSBFFSNAME Varchar(255),SKR Varchar(255),SKRYH Varchar(255)," }
+ "SKRZHZH Varchar(255),FKZHCODE Varchar(255),FKZHNAME Varchar(255),FKYHCODE Varchar(255),FKYHNAME Varchar(255),QSZHCODE Varchar(255),"
+ "QSZHNAME Varchar(255),QSYHCODE Varchar(255),QSYHNAME Varchar(255),JE Numeric(18,2), SFTK Varchar(255),NIAN Varchar(255),ZY Varchar(255))"; }
try { /**
OracleConnector.execOracleSQL(conn, payCmd, Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName()); * 测试是否能否连接上？
} catch (Exception e) { *
log.error(Custom4exception.threadVolume_Oracle_Except, e); * @return true：能连接上
} */
} public boolean testConnect(Connection conn) {
String testSql = "select count(*) from user_tables";
/** boolean flag = false;
* 创建dblink try {
* Statement statement = conn.createStatement();
* @param conn 汇总库的连接 if (statement.executeUpdate(testSql) > 0)
* @param oc 采集库 flag = true;
* dblink连接参数实体 } catch (SQLException e) {
*/ flag = false;
public void createStandardDBLink(Connection conn, OracleConnectorParams oc) { log.error(Custom4exception.threadVolume_Oracle_Except, e);
String linkName = "LINKTOST" + oc.getName(); }
String sql = "CREATE PUBLIC DATABASE LINK " return flag;
+ linkName }
+ " CONNECT TO "
+ oc.getUser() /**
+ " IDENTIFIED BY " * 执行抽取操作--支付表
+ oc.getPassword() *
+ " USING \'(DESCRIPTION =(ADDRESS_LIST =(ADDRESS = (PROTOCOL = TCP)(HOST = " * @param conn
+ oc.getIp() + ")(PORT = " + oc.getPort() * 汇总库连接
+ ")))(CONNECT_DATA =(SERVICE_NAME =" + oc.getDatabaseName() * @param oc
+ ")))\'"; * 采集库
if (null != oc) { */
if (hasSameNameDBLink(conn, linkName, Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName()+".log")) { // 如果dblink已经存在,先删除dblink，在创建dblink public void extractStandardPayTable(Connection conn,
String deleteSql = "DROP PUBLIC DATABASE LINK " OracleConnectorParams oc, GatherOracleInfo totalOracle) {
+ linkName; String strTUser = Configs.GATHER_STANDARD_USER_NAME;
//删除 dblink createPay(conn, oc);
if (OracleConnector.execOracleSQL(conn, deleteSql, Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName())) { String sql = "insert into " + strTUser + "."
// OracleConnector.execOracleSQL(conn, sql, Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName()); + Configs.GATHER_STANDARD_PAY_TABLE_NAME + " select * from "
if(!OracleConnector.execOracleSQL(conn, sql, Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName())){ + strTUser + "." + Configs.GATHER_STANDARD_PAY_TABLE_NAME
OracleConnector.execUpdateOracleSQL(conn, sql, Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName()); + "@LINKTOST" + oc.getName();
} String resultSql = sql.replace(";", "");
} else { OracleConnector.execOracleSQL(conn, resultSql,
Configs.CONSOLE_LOGGER.error("删除已有的DBLink失败,无法创建新的DBLink!"); Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName());
FileOperateHelper.fileWrite(Configs.EXTRACT_STANDARD_LOG_LOCALTION }
+ oc.getName()+".log", "删除已有的DBLink失败,无法创建新的DBLink!"+"\r\n");
} /**
* 执行抽取操作--执行表
} else { *
// 否则，创建dblink * @param conn
// OracleConnector.execOracleSQL(conn, sql, Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName()); * @param collectOracle
if(!OracleConnector.execOracleSQL(conn, sql, Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName())){ * @param oracleModel
OracleConnector.execUpdateOracleSQL(conn, sql, Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName()); */
} public void extractStandardExecTable(Connection conn,
} OracleConnectorParams collectOracle, GatherOracleInfo oracleModel) {
} createExec(conn, collectOracle);
} String strTUser = Configs.GATHER_STANDARD_USER_NAME;
private void createExec(Connection conn, OracleConnectorParams oc) { String sql = "insert into " + strTUser + "."
String execCmd = "CREATE TABLE u_bzbjy.kzxzb(XZQHDM Varchar(255),XZQHMC Varchar(255),YSND Varchar(255),ZBCODE Varchar(255),ZB_ID Varchar(255),ZB_NO Varchar(255)," + Configs.GATHER_STANDARD_EXEC_TABLE_NAME + " select * from "
+ "ZBDJLXCODE Varchar(255),ZBDJLXNAME Varchar(255),ZBLXNAME Varchar(255),DOCNO Varchar(255),ZBSM Varchar(255),ZBFWDATE Varchar(255),ZBYSLXCODE Varchar(255)," + strTUser + "." + Configs.GATHER_STANDARD_EXEC_TABLE_NAME
+ "ZBYSLXNAME Varchar(255),ZBYSLYNAME Varchar(255),YSDWCODE Varchar(255),YSDWNAME Varchar(255),GNFLCODE Varchar(255),GNFLNAME Varchar(255),JJFLCODE Varchar(255)," + "@LINKTOST" + collectOracle.getName();
+ "JJFLNAME Varchar(255),ZBGLCSNAME Varchar(255),SZGLCODE Varchar(255),SZGLNAME Varchar(255),XMCODE Varchar(255),XMNAME Varchar(255),GZBZ Varchar(255)," String resultSql = sql.replace(";", "");
+ "JJBZ Varchar(255),CGBZ Varchar(255),ZFFSCODE Varchar(255),ZFFSNAME Varchar(255),JZZFBZ Varchar(255),ZBJE Numeric(18,2),ZBTJJE Numeric(18,2),ZBDJJE Numeric(18,2)," OracleConnector.execOracleSQL(
+ "ZBKYJE Numeric(18,2),ZYZFBZ Varchar(255),BZ Varchar(255))"; conn,
resultSql,
try { Configs.EXTRACT_STANDARD_LOG_LOCALTION
OracleConnector.execOracleSQL(conn, execCmd, Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName()); + collectOracle.getName());
} catch (Exception e) {
log.error(Custom4exception.threadVolume_Oracle_Except, e); }
}
} private void createPay(Connection conn, OracleConnectorParams oc) {
} String payCmd = "CREATE TABLE u_bzbjy.zfxxb(XZQHDM Varchar(255),XZQHMC Varchar(255),PZBH Varchar(255),LYZBKZH Varchar(255),"
+ "ZFDATE Varchar(255),YSDWCODE Varchar(255),YSDWNAME Varchar(255),YWGKCS Varchar(255),XMCODE Varchar(255),XMNAME Varchar(255),"
+ "XMLBCODE Varchar(255),XMLBNAME Varchar(255),ZB_NO Varchar(255),GNFLCODE Varchar(255),GNFLNAME Varchar(255),JJFLCODE Varchar(255),"
+ "JJFLNAME Varchar(255),ZJXZCODE Varchar(255),ZJXZNAME Varchar(255),JSBFFSNAME Varchar(255),SKR Varchar(255),SKRYH Varchar(255),"
+ "SKRZHZH Varchar(255),FKZHCODE Varchar(255),FKZHNAME Varchar(255),FKYHCODE Varchar(255),FKYHNAME Varchar(255),QSZHCODE Varchar(255),"
+ "QSZHNAME Varchar(255),QSYHCODE Varchar(255),QSYHNAME Varchar(255),JE Numeric(18,2), SFTK Varchar(255),NIAN Varchar(255),ZY Varchar(255))";
try {
OracleConnector.execOracleSQL(conn, payCmd,
Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName());
} catch (Exception e) {
log.error(Custom4exception.threadVolume_Oracle_Except, e);
}
}
/**
* 创建dblink
*
* @param conn
* 汇总库的连接
* @param oc
* 采集库 dblink连接参数实体
*/
public void createStandardDBLink(Connection conn, OracleConnectorParams oc) {
String linkName = "LINKTOST" + oc.getName();
String sql = "CREATE PUBLIC DATABASE LINK "
+ linkName
+ " CONNECT TO "
+ oc.getUser()
+ " IDENTIFIED BY "
+ oc.getPassword()
+ " USING \'(DESCRIPTION =(ADDRESS_LIST =(ADDRESS = (PROTOCOL = TCP)(HOST = "
+ oc.getIp() + ")(PORT = " + oc.getPort()
+ ")))(CONNECT_DATA =(SERVICE_NAME =" + oc.getDatabaseName()
+ ")))\'";
if (null != oc) {
if (hasSameNameDBLink(conn, linkName,
Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName()
+ ".log")) { // 如果dblink已经存在,先删除dblink，在创建dblink
String deleteSql = "DROP PUBLIC DATABASE LINK " + linkName;
// 删除 dblink
if (OracleConnector.execOracleSQL(conn, deleteSql,
Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName())) {
// OracleConnector.execOracleSQL(conn, sql,
// Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName());
if (!OracleConnector.execOracleSQL(
conn,
sql,
Configs.EXTRACT_STANDARD_LOG_LOCALTION
+ oc.getName())) {
OracleConnector.execUpdateOracleSQL(
conn,
sql,
Configs.EXTRACT_STANDARD_LOG_LOCALTION
+ oc.getName());
}
} else {
Configs.CONSOLE_LOGGER.error("删除已有的DBLink失败,无法创建新的DBLink!");
FileOperateHelper.fileWrite(
Configs.EXTRACT_STANDARD_LOG_LOCALTION
+ oc.getName() + ".log",
"删除已有的DBLink失败,无法创建新的DBLink!" + "\r\n");
}
} else {
// 否则，创建dblink
// OracleConnector.execOracleSQL(conn, sql,
// Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName());
if (!OracleConnector.execOracleSQL(conn, sql,
Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName())) {
OracleConnector.execUpdateOracleSQL(
conn,
sql,
Configs.EXTRACT_STANDARD_LOG_LOCALTION
+ oc.getName());
}
}
}
}
private void createExec(Connection conn, OracleConnectorParams oc) {
String execCmd = "CREATE TABLE u_bzbjy.kzxzb(XZQHDM Varchar(255),XZQHMC Varchar(255),YSND Varchar(255),ZBCODE Varchar(255),ZB_ID Varchar(255),ZB_NO Varchar(255),"
+ "ZBDJLXCODE Varchar(255),ZBDJLXNAME Varchar(255),ZBLXNAME Varchar(255),DOCNO Varchar(255),ZBSM Varchar(255),ZBFWDATE Varchar(255),ZBYSLXCODE Varchar(255),"
+ "ZBYSLXNAME Varchar(255),ZBYSLYNAME Varchar(255),YSDWCODE Varchar(255),YSDWNAME Varchar(255),GNFLCODE Varchar(255),GNFLNAME Varchar(255),JJFLCODE Varchar(255),"
+ "JJFLNAME Varchar(255),ZBGLCSNAME Varchar(255),SZGLCODE Varchar(255),SZGLNAME Varchar(255),XMCODE Varchar(255),XMNAME Varchar(255),GZBZ Varchar(255),"
+ "JJBZ Varchar(255),CGBZ Varchar(255),ZFFSCODE Varchar(255),ZFFSNAME Varchar(255),JZZFBZ Varchar(255),ZBJE Numeric(18,2),ZBTJJE Numeric(18,2),ZBDJJE Numeric(18,2),"
+ "ZBKYJE Numeric(18,2),ZYZFBZ Varchar(255),BZ Varchar(255))";
try {
OracleConnector.execOracleSQL(conn, execCmd,
Configs.EXTRACT_STANDARD_LOG_LOCALTION + oc.getName());
} catch (Exception e) {
log.error(Custom4exception.threadVolume_Oracle_Except, e);
}
}
}

107
src/com/platform/service/OracleExtractTask.java
Unescape View File

@ -0,0 +1,107 @@
package com.platform.service;
import java.sql.Connection;
import java.util.Date;
import java.util.List;
import org.apache.log4j.Logger;
import com.base.Custom4exception;
import com.platform.entities.AbstractOracleExtractTask;
import com.platform.entities.GatherOracleInfo;
import com.platform.entities.OracleConnectorParams;
import com.platform.oracle.OracleConnector;
import com.platform.utils.Configs;
import com.platform.utils.Constant;
import com.platform.utils.DateForm;
import com.platform.utils.FileOperateHelper;
public class OracleExtractTask extends AbstractOracleExtractTask {
public static Logger log = Logger.getLogger(OracleExtractTask.class);
private GatherOracleInfo gatherOracleInfo;
private OracleConnectorParams oc;
private IOracleExtractService OracleExtract;
public OracleExtractTask(String name, GatherOracleInfo gatherOracleInfo,
OracleConnectorParams oc, IOracleExtractService OracleExtract) {
super(name);
this.gatherOracleInfo = gatherOracleInfo;
this.oc = oc;
this.OracleExtract = OracleExtract;
}
public OracleExtractTask(GatherOracleInfo gatherOracleInfo,
OracleConnectorParams oc, IOracleExtractService OracleExtract) {
this.gatherOracleInfo = gatherOracleInfo;
this.oc = oc;
this.OracleExtract = OracleExtract;
}
private OracleExtractHelper oracleExtract = new OracleExtractHelper();
@Override
public void handler() {
// TODO Auto-generated method stub
try {
Connection conn = OracleConnector.connectionBuilder(
createConnectUrl(), gatherOracleInfo.getUser(),
gatherOracleInfo.getPassword(), oc);
if (null != conn) {
String cmd = "kubectl label --overwrite rc " + oc.getName()
+ " isExtract=1";
OracleExtract.updateDataExtractStatus(oc, 1); // 更新数据库的状态
FileOperateHelper
.fileWrite(
Configs.EXTRACT_LOG_LOCALTION + oc.getName()
+ ".log",
"["
+ DateForm
.date2StringBysecond(new Date())
+ "]>>>>>>>>>>>>>>>>开始汇总 >>>>>>>>>>>>>>>>>>\r\n");
List<String> rList = Constant.ganymedSSH
.execCmdWaitAcquiescent(cmd);
StringBuffer sb = new StringBuffer();
for (String string : rList)
sb.append(string).append("\n");
Configs.CONSOLE_LOGGER.info(sb.toString());
oracleExtract.createDBLink(conn, oc); // 创建dblink
oracleExtract.createTableSpace(conn, oc, gatherOracleInfo); // 创建表空间
oracleExtract.createUser(conn, oc, gatherOracleInfo);// 创建用户并授权
oracleExtract.extractColleDB(conn, oc, gatherOracleInfo);// 执行抽取
cmd = "kubectl label --overwrite rc " + oc.getName()
+ " isExtract=2";
rList = Constant.ganymedSSH.execCmdWaitAcquiescent(cmd);
sb = new StringBuffer();
for (String string : rList)
sb.append(string).append("\n");
Configs.CONSOLE_LOGGER.info(sb.toString());
FileOperateHelper
.fileWrite(
Configs.EXTRACT_LOG_LOCALTION + oc.getName()
+ ".log",
"["
+ DateForm
.date2StringBysecond(new Date())
+ "]>>>>>>>>>>>>>>>>汇总结束 >>>>>>>>>>>>>>>>>>\r\n\r\n");
OracleExtract.updateDataExtractStatus(oc, 2); // 更新数据库的状态
}
} catch (Exception e) {
log.error(Custom4exception.OracleSQL_Except, e);
}
}
private String createConnectUrl() {
String answer = "";
if (null != gatherOracleInfo) {
answer = "jdbc:oracle:thin:@" + gatherOracleInfo.getIp() + ":"
+ gatherOracleInfo.getPort() + ":"
+ gatherOracleInfo.getDatabaseName();
}
return answer;
}
}

2
src/com/platform/service/impl/LogReadServiceImpl.java
Unescape View File

@ -12,7 +12,7 @@ public class LogReadServiceImpl implements ILogRead {
@Override @Override
public String readLog(String filename) throws Exception { public String readLog(String filename) throws Exception {
String result = FileOperateHelper.fileReader(Configs.EXTRACT_LOG_LOCALTION + "J" + filename.replace("-", "_")+".log"); String result = FileOperateHelper.fileReader(Configs.EXTRACT_LOG_LOCALTION + filename+".log");
return result; return result;
} }

679
src/com/platform/service/impl/OracleExtractServiceImpl.java
Unescape View File

@ -1,299 +1,380 @@
package com.platform.service.impl; package com.platform.service.impl;
import java.sql.Connection; import java.sql.Connection;
import java.util.Date; import java.util.Date;
import java.util.List; import java.util.List;
import javax.annotation.Resource; import javax.annotation.Resource;
import org.apache.log4j.Logger; import org.apache.log4j.Logger;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import com.base.Custom4exception; import com.base.Custom4exception;
import com.base.CustomException; import com.base.CustomException;
import com.platform.controller.OracleController; import com.platform.dao.DataInfoDao;
import com.platform.dao.DataInfoDao; import com.platform.entities.DataInfoEntity;
import com.platform.entities.DataInfoEntity; import com.platform.entities.GatherOracleInfo;
import com.platform.entities.GatherOracleInfo; import com.platform.entities.OracleConnectorParams;
import com.platform.entities.OracleConnectorParams; import com.platform.kubernetes.SimpleKubeClient;
import com.platform.kubernetes.SimpleKubeClient; import com.platform.oracle.OracleConnector;
import com.platform.oracle.OracleConnector; import com.platform.service.IOracleExtractService;
import com.platform.service.IOracleExtractService; import com.platform.service.OracleExtractHelper;
import com.platform.service.OracleExtractHelper; import com.platform.utils.Configs;
import com.platform.utils.Configs; import com.platform.utils.Constant;
import com.platform.utils.Constant; import com.platform.utils.DateForm;
import com.platform.utils.DateForm; import com.platform.utils.FileOperateHelper;
import com.platform.utils.FileOperateHelper;
@Service(value = "OracleExtract")
@Service(value = "OracleExtract") public class OracleExtractServiceImpl implements IOracleExtractService {
public class OracleExtractServiceImpl implements IOracleExtractService {
/**
/** * 日志
* 日志 */
*/ public final static Logger log = Logger
public final static Logger log = Logger.getLogger(OracleExtractServiceImpl.class); .getLogger(OracleExtractServiceImpl.class);
@Resource(name = "dataInfoDao") @Resource(name = "dataInfoDao")
private DataInfoDao dataInfoDao; private DataInfoDao dataInfoDao;
/** /**
* kubernetes client * kubernetes client
*/ */
private SimpleKubeClient client = new SimpleKubeClient(); private SimpleKubeClient client = new SimpleKubeClient();
/** /**
* 抽取 * 抽取
*/ */
private OracleExtractHelper oracleExtract = new OracleExtractHelper(); private OracleExtractHelper oracleExtract = new OracleExtractHelper();
/** /**
* 数据库连接实现类 * 数据库连接实现类
*/ */
private OracleConnector connect = new OracleConnector(); private OracleConnector connect = new OracleConnector();
@Override @Override
public boolean extractOracle(String name, List<OracleConnectorParams> dataInfolist, public boolean extractOracle(String name,
GatherOracleInfo oracleConnect) throws Exception { List<OracleConnectorParams> datainfos, GatherOracleInfo oracleModel)
boolean isSuccess = false; throws Exception {
try{ boolean isSuccess = false;
//map转 bean(汇总库信息-带tableName的) try {
GatherOracleInfo oracleModel = oracleConnect; // map转 bean(汇总库信息-带tableName的)
//采集库连接参数 // GatherOracleInfo oracleModel = oracleConnect;
List<OracleConnectorParams> datainfos = dataInfolist; // 采集库连接参数
if (datainfos.size() == 0) { // List<OracleConnectorParams> datainfos = dataInfolist;
return false; if (datainfos.size() == 0) {
} return false;
Connection conn = OracleConnector.connectionBuilder("jdbc:oracle:thin:@" + oracleModel.getIp() + ":" + oracleModel.getPort() + ":" }
+ oracleModel.getDatabaseName(), oracleModel.getUser(), oracleModel.getPassword(),dataInfolist.get(0)); Connection conn = OracleConnector.connectionBuilder(
if (null == conn) { "jdbc:oracle:thin:@" + oracleModel.getIp() + ":"
FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION + oracleModel.getPort() + ":"
+ dataInfolist.get(0).getName()+".log", "创建oracle连接失败: [" + conn + "]\r\n"); + oracleModel.getDatabaseName(),
return false; oracleModel.getUser(), oracleModel.getPassword(),
} datainfos.get(0));
for (OracleConnectorParams collectOracle : datainfos) { if (null == conn) {
String replicasName = collectOracle.getName(); FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION
DataInfoEntity data = new DataInfoEntity(); + datainfos.get(0).getName() + ".log",
try{ "创建oracle连接失败: [" + conn + "]\r\n");
if(null != collectOracle.getDataId() && !"".equals(collectOracle.getDataId())){ return false;
data.setId(Integer.valueOf(collectOracle.getDataId())); }
data.setExtractStatus(1); for (OracleConnectorParams collectOracle : datainfos) {
dataInfoDao.updateExtract(data); String replicasName = collectOracle.getName();
collectOracle.setName("J" + collectOracle.getName().replace("-", "_")); try {
String cmd = "kubectl label --overwrite rc " if (null != collectOracle.getDataId()
+ replicasName + " isExtract=1"; && !"".equals(collectOracle.getDataId())) {
//sql日志记录时间： DataInfoEntity data = new DataInfoEntity();
FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION data.setId(Integer.valueOf(collectOracle.getDataId()));
+ collectOracle.getName()+".log", "\r\n 开始汇总 \r\n"+ DateForm.date2StringBysecond(new Date()) +"\r\n"); data.setExtractStatus(1);
List<String> rList = Constant.ganymedSSH dataInfoDao.updateExtract(data);
.execCmdWaitAcquiescent(cmd); collectOracle.setName("J"
StringBuffer sb = new StringBuffer(); + collectOracle.getName().replace("-", "_"));
for (String string : rList) String cmd = "kubectl label --overwrite rc "
sb.append(string).append("\n"); + replicasName + " isExtract=1";
Configs.CONSOLE_LOGGER.info(sb.toString()); // sql日志记录时间：
// client.updateOrAddReplicasLabelById(collectOracle.getName(), "isExtract", "1"); //更新oracle汇总状态，0标示为未汇总，1标示汇总中，2标示汇总完成 FileOperateHelper
oracleExtract.createDBLink(conn, collectOracle); //创建dblink .fileWrite(
oracleExtract.createTableSpace(conn, collectOracle, oracleModel); //创建表空间 Configs.EXTRACT_LOG_LOCALTION
oracleExtract.createUser(conn, collectOracle, oracleModel);//创建用户并授权 + collectOracle.getName()
oracleExtract.extractColleDB(conn, collectOracle, oracleModel);//执行抽取 + ".log",
// client.updateOrAddReplicasLabelById(collectOracle.getName(), "isExtract", "2"); //更新oracle汇总状态，0标示为未汇总，1标示汇总中，2标示汇总完成 "\r\n 开始汇总 \r\n"
cmd = "kubectl label --overwrite rc " + DateForm
+ replicasName + " isExtract=2"; .date2StringBysecond(new Date())
rList = Constant.ganymedSSH.execCmdWaitAcquiescent(cmd); + "\r\n");
sb = new StringBuffer(); List<String> rList = Constant.ganymedSSH
for (String string : rList) .execCmdWaitAcquiescent(cmd);
sb.append(string).append("\n"); StringBuffer sb = new StringBuffer();
Configs.CONSOLE_LOGGER.info(sb.toString()); for (String string : rList)
data.setExtractStatus(2); sb.append(string).append("\n");
dataInfoDao.updateExtract(data); Configs.CONSOLE_LOGGER.info(sb.toString());
} // client.updateOrAddReplicasLabelById(collectOracle.getName(),
}catch(Exception e){ // "isExtract", "1");
log.error(Custom4exception.OracleSQL_Except, e); // //更新oracle汇总状态，0标示为未汇总，1标示汇总中，2标示汇总完成
} oracleExtract.createDBLink(conn, collectOracle); // 创建dblink
finally{ oracleExtract.createTableSpace(conn, collectOracle,
data.setExtractStatus(2); oracleModel); // 创建表空间
dataInfoDao.updateExtract(data); oracleExtract.createUser(conn, collectOracle,
//sql日志记录时间： oracleModel);// 创建用户并授权
FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION oracleExtract.extractColleDB(conn, collectOracle,
+ collectOracle.getName()+".log", "\r\n 汇总结束 \r\n"+ DateForm.date2StringBysecond(new Date()) +"\r\n"); oracleModel);// 执行抽取
String cmd = "kubectl label --overwrite rc " // client.updateOrAddReplicasLabelById(collectOracle.getName(),
+ replicasName + " isExtract=2"; // "isExtract", "2");
Constant.ganymedSSH.execCmdWaitAcquiescent(cmd); // //更新oracle汇总状态，0标示为未汇总，1标示汇总中，2标示汇总完成
} cmd = "kubectl label --overwrite rc " + replicasName
} + " isExtract=2";
isSuccess = true; rList = Constant.ganymedSSH.execCmdWaitAcquiescent(cmd);
}catch(Exception e){ sb = new StringBuffer();
new CustomException(Custom4exception.OracleSQL_Except, e); for (String string : rList)
} sb.append(string).append("\n");
return isSuccess; Configs.CONSOLE_LOGGER.info(sb.toString());
} data.setExtractStatus(2);
dataInfoDao.updateExtract(data);
@Override }
public boolean isConnectTotalOracle(GatherOracleInfo oracleModel) throws Exception { } catch (Exception e) {
boolean isConnect = false; log.error(Custom4exception.OracleSQL_Except, e);
Connection conn = OracleConnector.connectionBuilder("jdbc:oracle:thin:@" + oracleModel.getIp() + ":" + oracleModel.getPort() + ":" } finally {
+ oracleModel.getDatabaseName(), oracleModel.getUser(), oracleModel.getPassword(), null); // sql日志记录时间：
if (null == conn) { FileOperateHelper.fileWrite(
isConnect = false; Configs.EXTRACT_LOG_LOCALTION
throw new CustomException(Custom4exception.connect_Oracle_Except, null, oracleModel); + collectOracle.getName() + ".log",
// FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION "\r\n 汇总结束 \r\n"
// + dataInfolist.get(0).getName(), "创建oracle连接失败: [" + conn + "]\r\n"); + DateForm.date2StringBysecond(new Date())
}else { + "\r\n");
isConnect = oracleExtract.testConnect(conn); String cmd = "kubectl label --overwrite rc " + replicasName
} + " isExtract=2";
return isConnect; Constant.ganymedSSH.execCmdWaitAcquiescent(cmd);
} }
}
@Override isSuccess = true;
public boolean extractStandardTable(String name, List<OracleConnectorParams> dataInfolist, } catch (Exception e) {
GatherOracleInfo oracleConnect) throws Exception { new CustomException(Custom4exception.OracleSQL_Except, e);
boolean isSuccess = false; }
try{ return isSuccess;
//map转 bean(汇总库信息-带tableName的) }
GatherOracleInfo oracleModel = oracleConnect;
//采集库连接参数 @Override
List<OracleConnectorParams> datainfos = dataInfolist; public boolean isConnectTotalOracle(GatherOracleInfo oracleModel)
if (datainfos.size() == 0) { throws Exception {
return false; boolean isConnect = false;
} Connection conn = OracleConnector.connectionBuilder(
Connection conn = OracleConnector.connectionBuilder("jdbc:oracle:thin:@" + oracleModel.getIp() + ":" + oracleModel.getPort() + ":" "jdbc:oracle:thin:@" + oracleModel.getIp() + ":"
+ oracleModel.getDatabaseName(), oracleModel.getUser(), oracleModel.getPassword(),dataInfolist.get(0)); + oracleModel.getPort() + ":"
if (null == conn) { + oracleModel.getDatabaseName(), oracleModel.getUser(),
FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION oracleModel.getPassword(), null);
+ dataInfolist.get(0).getName()+".log", "创建oracle连接失败: [" + conn + "]\r\n"); if (null == conn) {
return false; isConnect = false;
} throw new CustomException(Custom4exception.connect_Oracle_Except,
for (OracleConnectorParams collectOracle : datainfos) { null, oracleModel);
String replicasName = collectOracle.getName(); // FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION
DataInfoEntity data = new DataInfoEntity(); // + dataInfolist.get(0).getName(), "创建oracle连接失败: [" + conn +
try{ // "]\r\n");
if(null != collectOracle.getDataId() && !"".equals(collectOracle.getDataId())){ } else {
data.setId(Integer.valueOf(collectOracle.getDataId())); isConnect = oracleExtract.testConnect(conn);
//设置为 标准表 抽取中 }
data.setCheckoutFlag(Constant.CHECKOUTFLAG_SIX); return isConnect;
data.setStandardExtractStatus("1"); }
dataInfoDao.update(data);
collectOracle.setName("CQ" + collectOracle.getName().replace("-", "_")); @Override
String cmd = "kubectl annotate --overwrite rc " public boolean extractStandardTable(String name,
+ replicasName + " standardExtractStatus=1"; List<OracleConnectorParams> dataInfolist,
//sql日志记录时间： GatherOracleInfo oracleConnect) throws Exception {
FileOperateHelper.fileWrite(Configs.EXTRACT_STANDARD_LOG_LOCALTION boolean isSuccess = false;
+ collectOracle.getName()+".log", "\r\n 开始抽取标准表 \r\n"+ DateForm.date2StringBysecond(new Date()) +"\r\n"); try {
List<String> rList = Constant.ganymedSSH // map转 bean(汇总库信息-带tableName的)
.execCmdWaitAcquiescent(cmd); GatherOracleInfo oracleModel = oracleConnect;
StringBuffer sb = new StringBuffer(); // 采集库连接参数
for (String string : rList) List<OracleConnectorParams> datainfos = dataInfolist;
sb.append(string).append("\n"); if (datainfos.size() == 0) {
Configs.CONSOLE_LOGGER.info(sb.toString()); return false;
// client.updateOrAddReplicasLabelById(collectOracle.getName(), "isExtract", "1"); //更新oracle汇总状态，0标示为未汇总，1标示汇总中，2标示汇总完成 }
oracleExtract.createStandardDBLink(conn, collectOracle); //创建dblink Connection conn = OracleConnector.connectionBuilder(
oracleExtract.createTableSpace(conn, collectOracle, oracleModel); //创建表空间 "jdbc:oracle:thin:@" + oracleModel.getIp() + ":"
oracleExtract.createOnlyUser(conn, collectOracle, oracleModel);//创建 抽取标准表的 用户并授权 + oracleModel.getPort() + ":"
DataInfoEntity tmpdata = dataInfoDao.findById(data.getId()); + oracleModel.getDatabaseName(),
if (null != tmpdata) { oracleModel.getUser(), oracleModel.getPassword(),
if (Constant.CHECKOUT_STATUS_THREE.equals(tmpdata.getPayResultLast()) dataInfolist.get(0));
|| Constant.CHECKOUT_STATUS_FIVE.equals(tmpdata.getPayResultLast()) if (null == conn) {
|| Constant.CHECKOUT_STATUS_SIX.equals(tmpdata.getPayResultLast())) { FileOperateHelper.fileWrite(Configs.EXTRACT_LOG_LOCALTION
//抽取中 + dataInfolist.get(0).getName() + ".log",
data.setPayResultLast(Constant.CHECKOUT_STATUS_SIX); "创建oracle连接失败: [" + conn + "]\r\n");
dataInfoDao.update(data); return false;
boolean isExtrac = true; }
try{ for (OracleConnectorParams collectOracle : datainfos) {
oracleExtract.extractStandardPayTable(conn, collectOracle, oracleModel);//执行抽取 String replicasName = collectOracle.getName();
}catch(Exception e){ try {
//改回 校验存在的状态 if (null != collectOracle.getDataId()
data.setPayResultLast(Constant.CHECKOUT_STATUS_THREE); && !"".equals(collectOracle.getDataId())) {
dataInfoDao.update(data); DataInfoEntity data = new DataInfoEntity();
isExtrac = false; data.setId(Integer.valueOf(collectOracle.getDataId()));
} // 设置为 标准表 抽取中
if (isExtrac) { data.setCheckoutFlag(Constant.CHECKOUTFLAG_SIX);
//抽取成功 data.setStandardExtractStatus("1");
data.setPayResultLast(Constant.CHECKOUT_STATUS_SEVEN); dataInfoDao.update(data);
dataInfoDao.update(data); collectOracle.setName("CQ"
} + collectOracle.getName().replace("-", "_"));
} String cmd = "kubectl annotate --overwrite rc "
if (Constant.CHECKOUT_STATUS_THREE.equals(tmpdata.getExecResultLast()) + replicasName + " standardExtractStatus=1";
|| Constant.CHECKOUT_STATUS_FIVE.equals(tmpdata.getExecResultLast()) // sql日志记录时间：
|| Constant.CHECKOUT_STATUS_SIX.equals(tmpdata.getExecResultLast())) { FileOperateHelper
//抽取中 .fileWrite(
data.setExecResultLast(Constant.CHECKOUT_STATUS_SIX); Configs.EXTRACT_STANDARD_LOG_LOCALTION
dataInfoDao.update(data); + collectOracle.getName()
boolean isExtrac = true; + ".log",
try{ "\r\n 开始抽取标准表 \r\n"
oracleExtract.extractStandardExecTable(conn, collectOracle, oracleModel);//执行抽取 + DateForm
}catch(Exception e){ .date2StringBysecond(new Date())
//改回 校验存在的状态 + "\r\n");
data.setExecResultLast(Constant.CHECKOUT_STATUS_THREE); List<String> rList = Constant.ganymedSSH
dataInfoDao.update(data); .execCmdWaitAcquiescent(cmd);
isExtrac = false; StringBuffer sb = new StringBuffer();
} for (String string : rList)
if (isExtrac) { sb.append(string).append("\n");
data.setExecResultLast(Constant.CHECKOUT_STATUS_SEVEN); Configs.CONSOLE_LOGGER.info(sb.toString());
dataInfoDao.update(data); // client.updateOrAddReplicasLabelById(collectOracle.getName(),
} // "isExtract", "1");
} // //更新oracle汇总状态，0标示为未汇总，1标示汇总中，2标示汇总完成
// client.updateOrAddReplicasLabelById(collectOracle.getName(), "isExtract", "2"); //更新oracle汇总状态，0标示为未汇总，1标示汇总中，2标示汇总完成 oracleExtract.createStandardDBLink(conn, collectOracle); // 创建dblink
cmd = "kubectl annotate --overwrite rc " oracleExtract.createTableSpace(conn, collectOracle,
+ replicasName + " standardExtractStatus=2"; oracleModel); // 创建表空间
rList = Constant.ganymedSSH.execCmdWaitAcquiescent(cmd); oracleExtract.createOnlyUser(conn, collectOracle,
sb = new StringBuffer(); oracleModel);// 创建 抽取标准表的 用户并授权
for (String string : rList) DataInfoEntity tmpdata = dataInfoDao.findById(data
sb.append(string).append("\n"); .getId());
Configs.CONSOLE_LOGGER.info(sb.toString()); if (null != tmpdata) {
data.setCheckoutFlag(Constant.CHECKOUTFLAG_SEVEN); if (Constant.CHECKOUT_STATUS_THREE.equals(tmpdata
data.setStandardExtractStatus("2"); .getPayResultLast())
dataInfoDao.update(data); || Constant.CHECKOUT_STATUS_FIVE
} .equals(tmpdata.getPayResultLast())
} || Constant.CHECKOUT_STATUS_SIX
}catch(Exception e){ .equals(tmpdata.getPayResultLast())) {
log.error(Custom4exception.OracleSQL_Except, e); // 抽取中
} data.setPayResultLast(Constant.CHECKOUT_STATUS_SIX);
finally{ dataInfoDao.update(data);
data.setCheckoutFlag(Constant.CHECKOUTFLAG_SEVEN); boolean isExtrac = true;
data.setStandardExtractStatus("2"); try {
dataInfoDao.update(data); oracleExtract.extractStandardPayTable(conn,
//sql日志记录时间： collectOracle, oracleModel);// 执行抽取
FileOperateHelper.fileWrite(Configs.EXTRACT_STANDARD_LOG_LOCALTION } catch (Exception e) {
+ collectOracle.getName()+".log", "\r\n 抽取标准表结束 \r\n"+ DateForm.date2StringBysecond(new Date()) +"\r\n"); // 改回 校验存在的状态
String cmd = "kubectl annotate --overwrite rc " data.setPayResultLast(Constant.CHECKOUT_STATUS_THREE);
+ replicasName + " standardExtractStatus=2"; dataInfoDao.update(data);
Constant.ganymedSSH.execCmdWaitAcquiescent(cmd); isExtrac = false;
} }
} if (isExtrac) {
isSuccess = true; // 抽取成功
}catch(Exception e){ data.setPayResultLast(Constant.CHECKOUT_STATUS_SEVEN);
log.error(Custom4exception.OracleSQL_Except, e); dataInfoDao.update(data);
} }
return isSuccess; }
} if (Constant.CHECKOUT_STATUS_THREE.equals(tmpdata
.getExecResultLast())
// @Override || Constant.CHECKOUT_STATUS_FIVE
// public boolean extractOracle(String name, List<OracleConnectorParams> dataInfos, GatherOracleInfo oracleConnect) throws Exception { .equals(tmpdata.getExecResultLast())
// boolean isSuccess = false; || Constant.CHECKOUT_STATUS_SIX
// try{ .equals(tmpdata.getExecResultLast())) {
// //map转 bean(汇总库信息-带tableName的) // 抽取中
//// GatherOracleInfo oracleModel = (GatherOracleInfo) Bean2MapUtils.convertMap(GatherOracleInfo.class, oracleConnect); data.setExecResultLast(Constant.CHECKOUT_STATUS_SIX);
// dataInfoDao.update(data);
// //采集库连接参数 boolean isExtrac = true;
//// List<OracleConnectorParams> datainfos = new ArrayList<OracleConnectorParams>(); try {
//// for (Map<String, String> map : dataInfoMap) { oracleExtract.extractStandardExecTable(
//// OracleConnectorParams dataInfoEntity = (OracleConnectorParams) Bean2MapUtils.convertMap(OracleConnectorParams.class, oracleConnect); conn, collectOracle, oracleModel);// 执行抽取
//// datainfos.add(dataInfoEntity); } catch (Exception e) {
//// } // 改回 校验存在的状态
// data.setExecResultLast(Constant.CHECKOUT_STATUS_THREE);
// Connection conn = OracleConnector.ConnectionBuilder("jdbc:oracle:thin:@" + oracleConnect.getIp() + ":" + oracleConnect.getPort() + "/" dataInfoDao.update(data);
// + oracleConnect.getDatabaseName(), oracleConnect.getUser(), oracleConnect.getPassword()); isExtrac = false;
// }
// for (OracleConnectorParams collectOracle : dataInfos) { if (isExtrac) {
// data.setExecResultLast(Constant.CHECKOUT_STATUS_SEVEN);
// oracleExtract.createDBLink(conn, collectOracle); dataInfoDao.update(data);
// oracleExtract.createTableSpace(conn, oracleConnect); }
// oracleExtract.createUser(conn, oracleConnect); }
// oracleExtract.extractColleDB(conn, collectOracle); // client.updateOrAddReplicasLabelById(collectOracle.getName(),
// } // "isExtract", "2");
// isSuccess = true; // //更新oracle汇总状态，0标示为未汇总，1标示汇总中，2标示汇总完成
// }catch(Exception e){ cmd = "kubectl annotate --overwrite rc "
// + replicasName + " standardExtractStatus=2";
// } rList = Constant.ganymedSSH
// return false; .execCmdWaitAcquiescent(cmd);
// } sb = new StringBuffer();
for (String string : rList)
} sb.append(string).append("\n");
Configs.CONSOLE_LOGGER.info(sb.toString());
data.setCheckoutFlag(Constant.CHECKOUTFLAG_SEVEN);
data.setStandardExtractStatus("2");
dataInfoDao.update(data);
}
}
} catch (Exception e) {
log.error(Custom4exception.OracleSQL_Except, e);
} finally {
// sql日志记录时间：
FileOperateHelper.fileWrite(
Configs.EXTRACT_STANDARD_LOG_LOCALTION
+ collectOracle.getName() + ".log",
"\r\n 抽取标准表结束 \r\n"
+ DateForm.date2StringBysecond(new Date())
+ "\r\n");
String cmd = "kubectl annotate --overwrite rc "
+ replicasName + " standardExtractStatus=2";
Constant.ganymedSSH.execCmdWaitAcquiescent(cmd);
}
}
isSuccess = true;
} catch (Exception e) {
log.error(Custom4exception.OracleSQL_Except, e);
}
return isSuccess;
}
@Override
public void updateDataExtractStatus(OracleConnectorParams ocp, int status) {
DataInfoEntity data = new DataInfoEntity();
data.setId(Integer.valueOf(ocp.getDataId()));
// 设置为 标准表 抽取中
data.setCheckoutFlag(Constant.CHECKOUTFLAG_SIX);
data.setExtractStatus(status);
try {
dataInfoDao.update(data);
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
// @Override
// public boolean extractOracle(String name, List<OracleConnectorParams>
// dataInfos, GatherOracleInfo oracleConnect) throws Exception {
// boolean isSuccess = false;
// try{
// //map转 bean(汇总库信息-带tableName的)
// // GatherOracleInfo oracleModel = (GatherOracleInfo)
// Bean2MapUtils.convertMap(GatherOracleInfo.class, oracleConnect);
//
// //采集库连接参数
// // List<OracleConnectorParams> datainfos = new
// ArrayList<OracleConnectorParams>();
// // for (Map<String, String> map : dataInfoMap) {
// // OracleConnectorParams dataInfoEntity = (OracleConnectorParams)
// Bean2MapUtils.convertMap(OracleConnectorParams.class, oracleConnect);
// // datainfos.add(dataInfoEntity);
// // }
//
// Connection conn = OracleConnector.ConnectionBuilder("jdbc:oracle:thin:@"
// + oracleConnect.getIp() + ":" + oracleConnect.getPort() + "/"
// + oracleConnect.getDatabaseName(), oracleConnect.getUser(),
// oracleConnect.getPassword());
//
// for (OracleConnectorParams collectOracle : dataInfos) {
//
// oracleExtract.createDBLink(conn, collectOracle);
// oracleExtract.createTableSpace(conn, oracleConnect);
// oracleExtract.createUser(conn, oracleConnect);
// oracleExtract.extractColleDB(conn, collectOracle);
// }
// isSuccess = true;
// }catch(Exception e){
//
// }
// return false;
// }
}

Write Preview
Loading…
Cancel
Save