执行流程：
1.从ftp判断文件是否存在
2.下载文件
3.创建ctl文件
4.创建可执行文件，bat或sh文件，文件中是sqlldr命令代码
5.备份表并创建临时表
6.执行文件，插入数据
7.检查数据完整性
8.创建主键和索引
9.将临时表修改为主表，删除临时表和备份表
1.
package com.iteye.aaa.job;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.FilenameFilter;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.iteye.mobilevideo.core.util.ftp.FtpServer;
import com.iteye.mobilevideo.core.util.ftp.FtpTemplate;
import com.iteye.aaa.constants.AaaConstants;
import com.iteye.aaa.dao.AaaSyncTestDao;
import com.iteye.aaa.mq.WSyncObject;
import com.iteye.aaa.util.AaaSyncUtil;
public class SyncAllTestJob {
private Log log = LogFactory.getLog(this.getClass().getName());
private String pkgSeq = "";
private String day = "";
private String filePath = "";
private String permissionFile = "";
private FtpServer ftpServer = null;
private List<String> fileList = null;
private List<String> commandFileList = null;
private long total = 0L;
private boolean mainTableEdit = false;
private String dropOld = "DROP TABLE W_OLD PURGE";
private String alterMain2Old = "ALTER TABLE W RENAME TO W_OLD";
private String createMain = "CREATE TABLE W(WID NUMBER NOT NULL,PKGSEQ VARCHAR2(100) NOT NULL,MOBILE VARCHAR2(20) NOT NULL,SERVTYPE VARCHAR2(4) NOT NULL,SERVICEID VARCHAR2(20) NOT NULL,SERVICECODE VARCHAR2(30) NOT NULL,ECID VARCHAR2(30) NOT NULL,OPRCODE VARCHAR2(4) NOT NULL,RESULTCODE VARCHAR2(10),RESULTDESC VARCHAR2(200),STATUS VARCHAR2(4) NOT NULL,BATCHEVENTID VARCHAR2(60),CREATETIME TIMESTAMP(6) NOT NULL,CREATOR VARCHAR2(30) NOT NULL)";
private String createTmp = "CREATE TABLE W_TMP(WID NUMBER NOT NULL,PKGSEQ VARCHAR2(100) NOT NULL,MOBILE VARCHAR2(20) NOT NULL,SERVTYPE VARCHAR2(4) NOT NULL,SERVICEID VARCHAR2(20) NOT NULL,SERVICECODE VARCHAR2(30) NOT NULL,ECID VARCHAR2(30) NOT NULL,OPRCODE VARCHAR2(4) NOT NULL,RESULTCODE VARCHAR2(10),RESULTDESC VARCHAR2(200),STATUS VARCHAR2(4) NOT NULL,BATCHEVENTID VARCHAR2(60),CREATETIME TIMESTAMP(6) NOT NULL,CREATOR VARCHAR2(30) NOT NULL)";
private String dropMain = "DROP TABLE W PURGE";
private String alterOld2Main = "ALTER TABLE W_OLD RENAME TO W";
private String dropTmp = "DROP TABLE W_TMP PURGE";
private String alterTmp2Main = "ALTER TABLE W_TMP RENAME TO W";
private String idIndex = "alter table W_TMP add primary key (WID) USING index";
private String timeIndex = "create index I_WM_CREATETIME_%s on W_TMP (CREATETIME DESC)";
private String creatorIndex = "create index I_WM_CREATOR_%s on W_TMP (CREATOR)";
private String serviceCodeIndex = "create index I_WM_SERVICECODE_%s on W_TMP (SERVICECODE)";
public SyncAllTestJob() {
init(null);
}
public SyncAllTestJob(String dayVal) {
init(dayVal);
}
public void init(String dayVal) {
Calendar calendar = Calendar.getInstance();
boolean createDay = true;
if (dayVal != null && !dayVal.trim().equals("")) {
try {
System.out.println(new SimpleDateFormat("yyyyMMdd").parse(dayVal.trim()).toString());
day = dayVal.trim();
createDay = false;
} catch (Exception e) {
}
}
if (createDay) {
day = new SimpleDateFormat("yyyyMMdd").format(calendar.getTime());
}
log.info("Sync all white list2:sync day of " + day + ".");
pkgSeq = "syncall" + new SimpleDateFormat("yyyyMMddHHmmssSSS").format(calendar.getTime());
calendar.add(Calendar.DATE, -1);
filePath = AaaConstants.localFtpFilePath + AaaConstants.syncallPath + day + "/";
permissionFile = AaaConstants.localFtpFilePath + AaaConstants.syncallPath + AaaConstants.syncallPermissionFile;
ftpServer = new FtpServer();
ftpServer.setIp(AaaConstants.mmFtpIp);
ftpServer.setPort(AaaConstants.mmFtpPort);
ftpServer.setFtpuser(AaaConstants.mmFtpUser);
ftpServer.setFtppasswd(AaaConstants.mmFtpPassword);
ftpServer.setFtpurl(AaaConstants.mmFtpSourcePath + AaaConstants.mmFtpMPAll);
}
public void executeSyncAllTest() {
int flag = 0;
log.info("Sync all white list:sync start.");
try {
viewFiles();
if (fileList != null && fileList.size() > 0) {
if (downloadFiles()) {
if (createControlFile()) {
createPerformFiles();
if (commandFileList != null && commandFileList.size() > 0) {
if (alterTable()) {
flag = 1;
if (perform()) {
if (checkData()) {
flag = 2;
}
}
}
}
}
}
} else {
log.error("Sync all white list:no any file by " + day + ".");
}
} catch (Exception e) {
log.error("Sync all white list:sync error.");
}
if (flag == 2) {
log.error("Sync all white list:commit.");
createIndex();
copyData(1);
commitTable();
} else if (flag == 1) {
log.error("Sync all white list:rollback.");
copyData(2);
rollbackTable();
}
}
public boolean alterTable() {
log.info("Sync all white list:alter table start.");
boolean flag = false;
try {
AaaSyncTestDao.excuteSql(dropOld);
} catch (Exception e) {
}
try {
AaaSyncTestDao.excuteSql(dropTmp);
} catch (Exception e) {
}
try {
if (AaaSyncTestDao.excuteSql(alterMain2Old)) {
mainTableEdit = true;
if (AaaSyncTestDao.excuteSql(createMain)) {
if (AaaSyncTestDao.excuteSql(createTmp)) {
flag = true;
} else {
log.info("Sync all white list:create tmp table error.");
}
} else {
log.info("Sync all white list:create table error.");
}
} else {
log.info("Sync all white list:alter main table error.");
}
} catch (Exception e) {
log.info("Sync all white list:alter table error.", e);
}
if (!flag && mainTableEdit) {
rollbackTable();
}
return flag;
}
private void rollbackTable() {
try {
AaaSyncTestDao.excuteSql(dropMain);
} catch (Exception e) {
}
try {
AaaSyncTestDao.excuteSql(dropTmp);
} catch (Exception e) {
}
try {
if (!AaaSyncTestDao.excuteSql(alterOld2Main)) {
log.info("Sync all white list:alter old table to main table error.");
}
} catch (Exception e) {
log.info("Sync all white list:alter old table to main table error.", e);
}
}
private void createIndex() {
Calendar calendar = Calendar.getInstance();
String seq = new SimpleDateFormat("yyyyMMddHHmm").format(calendar.getTime());
try {
AaaSyncTestDao.excuteSql(idIndex);
log.info("Sync all white list:create id index success.");
} catch (Exception e) {
log.error("Sync all white list:create id index error.", e);
}
try {
AaaSyncTestDao.excuteSql(String.format(serviceCodeIndex, seq));
log.info("Sync all white list:create serviceCode index success.");
} catch (Exception e) {
log.error("Sync all white list:create serviceCode index error.", e);
}
try {
AaaSyncTestDao.excuteSql(String.format(timeIndex, seq));
log.info("Sync all white list:create createTime index success.");
} catch (Exception e) {
log.error("Sync all white list:create createTime index error.", e);
}
try {
AaaSyncTestDao.excuteSql(String.format(creatorIndex, seq));
log.info("Sync all white list:create creator index success.");
} catch (Exception e) {
log.error("Sync all white list:create creator index error.", e);
}
}
private void copyData(int val) {
boolean flag = false;
try {
String sql = "";
if (val == 1) {
sql = "INSERT INTO W_TMP SELECT W_SEQ.NEXTVAL, PKGSEQ, MOBILE, SERVTYPE, SERVICEID, SERVICECODE, ECID, OPRCODE, RESULTCODE, RESULTDESC, STATUS, BATCHEVENTID, CREATETIME, CREATOR FROM W";
} else {
sql = "INSERT INTO W_OLD SELECT W_SEQ.NEXTVAL, PKGSEQ, MOBILE, SERVTYPE, SERVICEID, SERVICECODE, ECID, OPRCODE, RESULTCODE, RESULTDESC, STATUS, BATCHEVENTID, CREATETIME, CREATOR FROM W";
}
if (AaaSyncTestDao.transTestMain2Other(sql)) {
flag = true;
AaaSyncTestDao.excuteSql(dropMain);
} else {
log.error("Sync all white list:cope data from main to tmp error.");
}
} catch (Exception e) {
}
if (!flag) {
Calendar calendar = Calendar.getInstance();
String seq = new SimpleDateFormat("yyyyMMddHHmm").format(calendar.getTime());
log.info("Sync all white list:cope data from main to tmp error,alter main table to W_" + seq);
AaaSyncTestDao.excuteSql("ALTER TABLE W_TMP RENAME TO W_" + seq);
}
}
private void commitTable() {
try {
AaaSyncTestDao.excuteSql(dropMain);
} catch (Exception e) {
}
try {
AaaSyncTestDao.excuteSql(dropOld);
} catch (Exception e) {
}
try {
if (!AaaSyncTestDao.excuteSql(alterTmp2Main)) {
log.info("Sync all white list:alter old table to main table error.");
}
} catch (Exception e) {
log.info("Sync all white list:alter old table to main table error.", e);
}
}
private void viewFiles() {
fileList = null;
try {
String[] files = FtpTemplate.listNames(ftpServer, "/");
if (files != null && files.length > 0) {
fileList = new ArrayList<String>();
for (int i = 0; i < files.length; i++) {
String fileName = files[i].trim();
if (validateFileName(fileName, day) && !fileList.contains(fileName)) {
log.info("Sync all white list:add sync file " + fileName);
fileList.add(fileName);
}
}
if (fileList != null && fileList.size() > 0) {
if (!AaaSyncTestDao.insertWSyncFile(pkgSeq, day, "00")) {
log.error("Sync all white list:view files success,but insert data to db error.");
fileList = null;
}
}
}
} catch (Exception e) {
log.error("Sync all white list:view files error.", e);
fileList = null;
}
}
private boolean downloadFiles() {
int len = fileList.size();
int successLen = 0;
for (int i = 0; i < len; i++) {
WSyncObject so = new WSyncObject();
so.setPkgSeq(pkgSeq);
so.setFileName(fileList.get(i));
so.setCount(1);
log.info("Sync all white list:start download " + so.toString() + "," + Thread.currentThread().getName());
if (downloadFile(so, filePath)) {
successLen++;
log.info("Sync all white list:file " + so.getFileName() + " download finish.");
} else {
log.error("Sync all white list:file " + so.getFileName() + " download error.");
break;
}
}
if (successLen == len) {
if (!AaaSyncTestDao.updateWSyncFile(pkgSeq, "01")) {
log.info("Sync all white list:download success but update status to database fail.");
}
return true;
} else {
if (!AaaSyncTestDao.updateWSyncFile(pkgSeq, "10")) {
log.info("Sync all white list:download fail and update status to database fail.");
}
return false;
}
}
private boolean createControlFile() {
if (exportToFile(filePath, "control.ctl", AaaConstants.syncallControlFileList)) {
return true;
}
return false;
}
private void createPerformFiles() {
log.info("Sync all white list:start create perform files.");
try {
String fileType = AaaConstants.syncallType.equalsIgnoreCase("bat") ? ".bat" : ".sh";
commandFileList = new ArrayList<String>();
String command = AaaConstants.syncallType.equalsIgnoreCase("bat") ? AaaConstants.syncallSqlldrBat : AaaConstants.syncallSqlldrSh;
int len = fileList.size();
int index = 1;
String content = "";
String fileName = "";
String path = "";
String toFileName = "";
List<String> contentList = new ArrayList<String>();
for (int i = 0; i < len; i++) {
fileName = fileList.get(i);
path = filePath + fileName.substring(0, fileName.length() - 4);
long line = readFirstLine(fileName);
if (line > 0) {
total += line;
content = String.format(command, path, filePath, path, path, line);
contentList.add(content);
if (AaaConstants.syncallType.equalsIgnoreCase("bat")) {
contentList.add("exit");
} else {
contentList.add("exit");
}
toFileName = "command" + index + fileType;
if (exportToFile(filePath, toFileName, contentList)) {
index++;
commandFileList.add(filePath + toFileName);
contentList.clear();
}
}
}
if (contentList.size() > 0) {
toFileName = "command" + index + fileType;
if (exportToFile(filePath, toFileName, contentList)) {
index++;
commandFileList.add(filePath + toFileName);
log.info("Sync all white list:create perform files finish.");
contentList.clear();
} else {
log.error("Sync all white list:create perform files error.");
commandFileList = null;
}
}
} catch (Exception e) {
log.error("Sync all white list:create perform files error.");
commandFileList = null;
}
}
private long readFirstLine(String fileName) {
long result = -1;
BufferedReader reader = null;
try {
reader = new BufferedReader(new FileReader(filePath + fileName));
String headerLine = reader.readLine();
if (headerLine == null) {
headerLine = "";
}
result = Long.parseLong(headerLine.split(",")[1].trim());
reader.close();
} catch (Exception e) {
result = -1;
log.error("Sync all white list:read " + (filePath + fileName) + " first line error.", e);
} finally {
if (reader != null) {
try {
reader.close();
} catch (IOException e) {
reader = null;
}
}
}
if (result != -1) {
return result;
} else {
throw new RuntimeException("Sync all white list:read " + (filePath + fileName) + " first line error.");
}
}
private boolean perform() {
log.info("Sync all white list:create perform files finish,start perform.");
boolean flag = false;
try {
if (AaaConstants.syncallType.equalsIgnoreCase("sh")) {
log.info("Sync all white list:chmod " + permissionFile + " " + filePath + "*.sh");
AaaSyncUtil.callSh(permissionFile + " " + filePath);
}
if (AaaConstants.importDataThreadPool == null) {
AaaConstants.importDataThreadPool = new ImportDataThreadPool();
}
for (int i = 0; i < commandFileList.size(); i++) {
AaaConstants.importDataThreadPool.execute(new ImportDataThreadPool.Task(commandFileList.get(i)));
}
while (AaaConstants.importDataThreadPool.getActiveCount() != 0) {
}
flag = true;
log.info("Sync all white list:perform all success.");
AaaConstants.importDataThreadPool.shutdown();
if (!AaaSyncTestDao.updateWSyncFile(pkgSeq, "02")) {
log.info("Sync all white list:perform all success but update status to database fail.");
}
} catch (Exception e) {
log.info("Sync all white list:perform files error.", e);
flag = false;
if (!AaaSyncTestDao.updateWSyncFile(pkgSeq, "20")) {
log.info("Sync all white list:perform fail but update status to database fail.");
}
}
return flag;
}
private boolean exportToFile(String filePath, String fileName, List<String> contentList) {
boolean flag = false;
FileOutputStream fos = null;
OutputStreamWriter osw = null;
BufferedWriter bw = null;
try {
fos = new FileOutputStream(new File(filePath + fileName));
osw = new OutputStreamWriter(fos, "UTF-8");
bw = new BufferedWriter(osw);
String line = "";
for (int i = 0; i < contentList.size(); i++) {
line = contentList.get(i);
try {
bw.write(line + "\r\n");
} catch (IOException e) {
e.printStackTrace();
}
}
bw.close();
osw.close();
fos.close();
flag = true;
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (bw != null) {
bw.close();
}
if (osw != null) {
osw.close();
}
if (fos != null) {
fos.close();
}
} catch (Exception ee) {
ee.printStackTrace();
}
}
return flag;
}
private boolean validateFileName(String fileName, String day) {
String perfix = AaaConstants.mpAll_file_perfix + day;
String suffix = AaaConstants.mpAll_file_suffix;
if (fileName != null && fileName.startsWith(perfix) && fileName.endsWith(suffix)) {
String middle = fileName.substring(perfix.length());
middle = middle.substring(0, middle.length() - suffix.length());
if (middle.length() == 6) {
char[] chars = middle.toCharArray();
if (chars.length == 6) {
if (chars[0] < '0' || chars[0] > '2') {
return false;
}
if (chars[1] < '0' || chars[1] > '9') {
return false;
}
if (chars[0] == '2' && chars[1] > '3') {
return false;
}
if (chars[2] != '_') {
return false;
}
for (int i = 3; i < 6; i++) {
if (chars[i] < '0' || chars[i] > '9') {
return false;
}
}
return true;
}
}
}
return false;
}
private boolean downloadFile(WSyncObject so, String filePath) {
FtpServer ftpServer = new FtpServer();
ftpServer.setIp(AaaConstants.mmFtpIp);
ftpServer.setPort(AaaConstants.mmFtpPort);
ftpServer.setFtpuser(AaaConstants.mmFtpUser);
ftpServer.setFtppasswd(AaaConstants.mmFtpPassword);
ftpServer.setFtpurl(AaaConstants.mmFtpSourcePath + AaaConstants.mmFtpMPAll);
// 下载文件
boolean downSuccessFlag = false;
try {
FtpTemplate.downLoadFilesByPath(ftpServer, "/" + so.getFileName(), filePath);
File f = new File(filePath + so.getFileName());
if (f.exists() && f.length() > 0) {
downSuccessFlag = true;
}
} catch (Exception e) {
downSuccessFlag = false;
e.printStackTrace();
}
// 处理下载失败文件
if (!downSuccessFlag) {
if (so.getCount() <= 3) {
log.info("Sync all white list:repeat download fail file " + (AaaConstants.mpAllDownInterval / 1000) + "S,for " + so.getCount() + ","
+ so.toString());
try {
Thread.sleep(AaaConstants.mpAllDownInterval);
} catch (InterruptedException e) {
e.printStackTrace();
}
so.setCount(so.getCount() + 1);
// 重新下载
downloadFile(so, filePath);
}
}
return downSuccessFlag;
}
static class SqlldrFilter implements FilenameFilter {
private String type;
public SqlldrFilter(String type) {
this.type = type;
}
public boolean accept(File dir, String name) {
return name.endsWith(type);
}
}
public boolean checkData() {
log.info("Sync all white list:check data.");
boolean flag = false;
File file = new File(filePath);
if (file.isDirectory()) {
SqlldrFilter filter = new SqlldrFilter(".bad");
String[] files = file.list(filter);
if (files == null || files.length == 0) {
long count = AaaSyncTestDao.countSql();
if (count == total) {
log.info("Sync all white list:check success,total=" + total);
return true;
} else {
log.info("Sync all white list:check fail,count=" + count + ",total=" + total);
}
} else {
log.info("Sync all white list:check fail,bad files " + Arrays.toString(files));
}
} else {
log.info("Sync all white list:check fail," + filePath + "is not a dir.");
}
return flag;
}
public static void main(String[] args) {
long start = System.currentTimeMillis();
String day = null;
if (args != null && args.length > 0) {
if (args[0] != null && args[0].trim().length() > 0) {
day = args[0].trim();
}
}
try {
AaaSyncUtil.init();
SyncAllTestJob job = null;
if (day == null) {
job = new SyncAllTestJob();
} else {
job = new SyncAllTestJob(day);
}
job.executeSyncAllTest();
} catch (Exception e) {
e.printStackTrace();
}
long end = System.currentTimeMillis();
System.out.println("Host:" + (end - start));
}
}
2.
package com.job;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.iteye.aaa.constants.AaaConstants;
public class ImportDataThreadPool extends ThreadPoolExecutor {
public ImportDataThreadPool() {
super(AaaConstants.syncallThread, AaaConstants.syncallThread, 5, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000),
new ThreadPoolExecutor.CallerRunsPolicy());
}
public static class Task implements Runnable {
private Log log = LogFactory.getLog(this.getClass().getName());
private String filePath;
public Task(String filePath) {
this.filePath = filePath;
}
@Override
public void run() {
log.info("Sync all white list:perform " + filePath);
if (AaaConstants.syncallType.equalsIgnoreCase("bat")) {
callBat(filePath);
} else {
callSh(filePath);
}
log.info("Sync all white list:perform " + filePath + " finish.");
}
public void callSh(String command) {
try {
Runtime rt = Runtime.getRuntime();
Process pcs = rt.exec(command);
BufferedReader br = new BufferedReader(new InputStreamReader(pcs.getInputStream()));
while (br.readLine() != null) {
}
try {
pcs.waitFor();
} catch (InterruptedException e) {
}
br.close();
pcs.exitValue();
log.info("Sync all white list:perform " + command + " success.");
} catch (Exception e) {
log.info("Sync all white list:perform " + command + " error.");
}
}
public void callBat(String command) {
command = AaaConstants.syncallPathPrefix + command;
try {
Process child = Runtime.getRuntime().exec("cmd.exe /C start " + command);
InputStream in = child.getInputStream();
while (in.read() != -1) {
}
in.close();
child.destroy();
log.info("Sync all white list:perform " + command + " success.");
} catch (IOException e) {
log.info("Sync all white list:perform " + command + " error.");
e.printStackTrace();
}
}
}
public static void main(String[] args) {
AaaConstants.syncallThread = 1;
AaaConstants.syncallType = "bat";
String cmd1 = "e:/usr/local/aaaservice/syncall/20130909/1.bat";
String cmd2 = "e:/usr/local/aaaservice/syncall/20130909/2.bat";
ImportDataThreadPool pool = new ImportDataThreadPool();
pool.execute(new ImportDataThreadPool.Task(cmd1));
pool.execute(new ImportDataThreadPool.Task(cmd2));
while (pool.getActiveCount() != 0) {
}
pool.shutdown();
System.out.println("down");
}
}
3.
package com.util;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.UnknownHostException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.iteye.aaa.constants.AaaConstants;
public class AaaSyncUtil {
private static Log log = LogFactory.getLog(AaaSyncUtil.class.getName());
public static void init() throws UnknownHostException, IOException {
AaaConstants.conf = new Configuration("/config.properties");
org.hibernate.cfg.Configuration cfg = new org.hibernate.cfg.Configuration();
AaaConstants.sessionFactory = cfg.configure().buildSessionFactory();
AaaConstants.mmFtpIp = StringUtil.null2Str(AaaConstants.conf.getProperty("mmFtpIp"));
AaaConstants.mmFtpPort = StringUtil.nullToLong(AaaConstants.conf.getProperty("mmFtpPort"));
AaaConstants.mmFtpUser = StringUtil.null2Str(AaaConstants.conf.getProperty("mmFtpUser"));
AaaConstants.mmFtpPassword = StringUtil.null2Str(AaaConstants.conf.getProperty("mmFtpPassword"));
AaaConstants.mmFtpSourcePath = StringUtil.null2Str(AaaConstants.conf.getProperty("mmFtpSourcePath"));
AaaConstants.mmFtpMPAll = StringUtil.null2Str(AaaConstants.conf.getProperty("mmFtpMPAll"));
AaaConstants.localFtpFilePath = StringUtil.null2Str(AaaConstants.conf.getProperty("localFtpFilePath"));
AaaConstants.mpAllDownInterval = StringUtil.nullToInteger(AaaConstants.conf.getProperty("mpAllDownInterval"));
AaaConstants.mpAllDealInterval = StringUtil.nullToInteger(AaaConstants.conf.getProperty("mpAllDealInterval"));
AaaConstants.mpAllDealNumber = StringUtil.nullToInteger(AaaConstants.conf.getProperty("mpAllDealNumber"));
AaaConstants.serviceFlag = StringUtil.null2Str(AaaConstants.conf.getProperty("serviceFlag"));
AaaConstants.syncallSqlldrBat = StringUtil.null2Str(AaaConstants.conf.getProperty("syncallSqlldrBat"));
AaaConstants.syncallSqlldrSh = StringUtil.null2Str(AaaConstants.conf.getProperty("syncallSqlldrSh"));
AaaConstants.syncallType = StringUtil.null2Str(AaaConstants.conf.getProperty("syncallType"));
AaaConstants.syncallPathPrefix = StringUtil.null2Str(AaaConstants.conf.getProperty("syncallPathPrefix"));
AaaConstants.syncallPermissionFile = StringUtil.null2Str(AaaConstants.conf.getProperty("syncallPermissionFile"));
AaaConstants.syncallThread = StringUtil.nullToInteger(AaaConstants.conf.getProperty("syncallThread"));
}
public static void callSh(String command) {
try {
Runtime rt = Runtime.getRuntime();
Process pcs = rt.exec(command);
BufferedReader br = new BufferedReader(new InputStreamReader(pcs.getInputStream()));
while (br.readLine() != null) {
}
try {
pcs.waitFor();
} catch (InterruptedException e) {
}
br.close();
pcs.exitValue();
log.info("Sync all white list:perform " + command + " success.");
} catch (Exception e) {
log.info("Sync all white list:perform " + command + " error.");
}
}
}
4.
package com.constants;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import org.hibernate.SessionFactory;
import com.iteye.aaa.job.ImportDataThreadPool;
import com.iteye.aaa.mina.AaaMinaClient;
import com.iteye.aaa.protocol.AaaBaseProtocol;
import com.iteye.aaa.service.AaaInterfaceManager;
import com.iteye.aaa.util.Configuration;
public class AaaConstants {
public static String mmFtpIp;
public static long mmFtpPort;
public static String mmFtpUser;
public static String mmFtpPassword;
public static String mmFtpSourcePath;
public static String mmFtpMPReq;
public static String mmFtpMPRst;
public static String mmFtpMPAll;
public static String mmFtpPMReq;
public static String mmFtpPMRst;
public static String localFtpFilePath;
public static int retryCount = 3;
public static final String mpAll_file_perfix = "MP_wlist_all_sync_";
public static final String mpAll_file_suffix = ".req";
public static final String mpAll_file_insert_sql = "INSERT INTO W_SYNC_FILES(ID, PKGSEQ, FILENAME, STATUS) VALUES (W_SYNC_FILES_SEQ.NEXTVAL,'%s','%s','%s')";
public static final String mpAll_file_update_sql = "UPDATE W_SYNC_FILES SET STATUS='%s' WHERE PKGSEQ='%s' AND FILENAME='%s'";
public static int mpAllDownInterval;
public static int mpAllDealInterval;
public static int mpAllDealNumber;
public static String serviceFlag = "";
public static String syncallPath = "syncall/";
public static String syncallPermissionFile ="";
public static String syncallSqlldrBat = "";
public static String syncallSqlldrSh = "";
public static String syncallType = "";
public static String syncallPathPrefix = "";
public static int syncallThread = 5;
public static List<String> syncallControlFileList = new ArrayList<String>();
public static ImportDataThreadPool importDataThreadPool = null;
static {
syncallControlFileList.add("load data");
syncallControlFileList.add("append into table w_tmp");
syncallControlFileList.add("fields terminated by ',' optionally enclosed by '\"'");
syncallControlFileList.add("trailing nullcols");
syncallControlFileList.add("(");
syncallControlFileList.add(" MOBILE,");
syncallControlFileList.add(" SERVTYPE,");
syncallControlFileList.add(" SERVICEID,");
syncallControlFileList.add(" SERVICECODE,");
syncallControlFileList.add(" ECID,");
syncallControlFileList.add(" PKGSEQ \"to_char(sysdate,'yyyyMMddHH24missSSS') || '01' || W_PKGSEQ.nextval\",");
syncallControlFileList.add(" OPRCODE CONSTANT '01',");
syncallControlFileList.add(" STATUS CONSTANT '1',");
syncallControlFileList.add(" CREATOR \":SERVICECODE\",");
syncallControlFileList.add(" CREATETIME \"SYSTIMESTAMP-1\",");
syncallControlFileList.add(" WID \"w_seq.nextval\"");
syncallControlFileList.add(")");
}
}
5.
ftpIp=192.168.1.133 ftpPort=21 ftpUser=aaa ftpPassword=aaa ftpSourcePath=/aaaservice ftpMPReq=/aa/q ftpMPRst=/aa/r ftpMPAll=/aa/a ftpPMReq=/bb/q ftpPMRst=/bb/r localFtpFilePath=/usr/local/aaaservice/ serviceFtpIp=192.168.1.100 serviceFtpPort=21 serviceFtpUser=aaa serviceFtpPassword=bbb serviceFtpSourcePath=/a/b localServiceFtpFilePath=/a/c retryCount=3 #activeTestInterval=60 activeTestRecivedInterval=60 activeTestNumber=3 mpAllDownInterval=5000 mpAllDealInterval=5000 mpAllDealNumber=10000 serviceFlag=01 syncallThread=5 syncallSqlldrBat=sqlldr userid=admin/admin@oracle data=%s.req control=%scontrol.ctl log=%s.log bad=%s.bad rows=2000 bindsize=6553600 load=%s skip=1 parallel=true syncallSqlldrSh=sqlldr userid=admin/admin@oracle data=%s.req control=%scontrol.ctl log=%s.log bad=%s.bad rows=2000 bindsize=6553600 load=%s skip=1 parallel=true syncallType=bat syncallPathPrefix=e: syncallPermissionFile=syncall.sh
6.Linux修改目录下所有SH文件权限
chmod 755 ${1}/*.sh
exit
7.ctl文件
load data append into table w_tmp fields terminated by ',' optionally enclosed by '"' trailing nullcols ( MOBILE, SERVTYPE, SERVICEID, SERVICECODE, ECID, PKGSEQ "to_char(sysdate,'yyyyMMddHH24missSSS') || '01' || W_PKGSEQ.nextval", OPRCODE CONSTANT '01', STATUS CONSTANT '1', CREATOR ":SERVICECODE", CREATETIME "SYSTIMESTAMP-1", WID "w_seq.nextval" )
相关推荐
Oracle的SQL*Loader（简称SQLLDR）工具就是这样一个强大的数据加载器，它能快速高效地将数据从外部文件批量导入到数据库中。结合批处理（BAT）脚本，我们可以自动化这个过程，提高工作效率。下面我们将详细探讨如何...
同时代码还会生成sqlldr命令需要用到的control文件和bat文件，以及连接Oracle的配置文件，只要将MySQL导出的数据文件合到一起，执行bat文件，即可完成数据导入到Oracle的任务。所有文件都是以导入导出的那张表的表名...
在java代码中调用执行shell脚本，sqlldr导数与使用sqlplus在shell调用执行存储过程。...于是我把oracle的所有环境变量直接复制到shell脚本文件中，在java中调用了一下，然后所有问题迎刃而解！ 具体代码参看文件内容
使用`ora_load.sh`脚本来自动化导出源数据库中的表数据，然后在目标数据库上创建相同的表结构，接着通过FTP传输数据文件到目标数据库，并执行`batch_load_data.sh`脚本来导入数据。导出数据时，利用`spool`功能将...
在“oracle sqlldr多文件批量导入”这个场景下，我们需要掌握如何利用SQL*Loader一次性处理多个数据文件的导入操作。下面我们将详细讲解这个主题。 首先，了解SQL*Loader的基本结构。SQL*Loader通过控制文件（.ctl...
也找不到其他资源，最后去找安装了完整版的同事，借鉴这个下载链接里面的文件，删除了其他没用到的东西(经历了好几个小时，一个一个dll测试过去)，最后sqlplus, sqlldr可用。 如果是遇到sql*loader 284错误的，是...
VS2013 C#.Net开发 C#利用SqlLdr 数据批量导入Oracle程序源码，高效可达万行/秒。 1.可多表同时执行 2.可视导入信息反馈，可查看错误数据及导入日志。 3.内含 批处理 执行方法 及 导入 ldr Demo文件。 4.内含导数据...
用的是11.2.0.4的Instant Client，Oracle没给sqlldr，Tools里面也没有（12C就有）。 网上下载了几个都是11.2.0.1的sqlldr，包括标了版本号的，都需要替换库文件。 这个是自己安装p13390677_112040_Linux-x86-64_4of7...
- 数据文件的内容将根据控制文件的规则，被加载到对应的数据库表中。 3. `MERGE INTO`语句合并更新数据： 当我们需要将新数据集与现有数据库表进行比较并根据条件进行插入或更新时，`MERGE INTO`语句就派上用场了...
在Oracle数据库管理中，SQL*Loader（简称sqlldr）是一个非常强大的工具，用于将外部数据文件中的数据批量加载到Oracle数据库表中。当涉及到处理带有自动增加字段的表时，SQL*Loader提供了一些灵活的方法来确保数据的...
5. **运行SQL*Loader**：在命令行环境中，执行SQL*Loader命令，传入控制文件和数据文件的路径。例如： ``` sqlldr username/password@database control=批量导入.ctl data=批量导入.txt ``` 6. **错误处理**：...
这类问题通常出现在需要将包含较长字符串的数据加载到表中时，而目标列定义的长度不足以容纳实际数据的情况。本文将围绕SQL*Loader字符长度错误的成因、解决方法以及如何优化控制文件（.ctl）进行详细介绍。 #### ...
sqlldr的使用则需要创建一个控制文件，指定数据文件、目标表、字段映射等信息，然后执行sqlldr命令进行数据加载。 总之，"Windows版oracle10g精简客户端（带sqlldr）"是一个为了方便开发者和管理员而定制的工具集，...
oracle 11.2客户端（linux） sqlldr相关文件 如果oracle客户端用的是rpm方式安装： sqlldr命令放在bin 其他文件放在lib 如果oracle客户端用的是直接解压方式： 所有都放在根目录即可
Sql Loader是一种命令行工具，通过读取控制文件（.ctl文件，如提供的`ldr_object.ctl`）来确定如何解析输入数据文件，并将其加载到数据库表中。其主要优点包括速度、灵活性和可配置性。 - **控制文件**：定义了...
SQL*Loader通过读取控制文件（.ctl文件）中的指令，解析数据文件，并将数据插入到指定的数据库表中。控制文件包含了数据文件的格式定义、字段映射、错误处理等信息。sqlldr的优势在于其并行处理能力，可以显著提高...
参数data指定了数据文件的名称和路径，数据文件包含了要加载到Oracle数据库中的数据。注意，data参数只能指定一个数据文件，如果控制文件中也指定了数据文件，那么sqlldr将优先加载data参数指定的数据文件。 ...
7. 可以从磁盘或磁带数据文件中装入数据到表中。 8. 提供装入错误报告。 9. 能将文件中的整型字符串自动转成压缩十进制并装入列表中。 控制文件是SQL*Loader执行数据加载操作的中枢，它是一个用特定语言编写的文本...
最最主要的还是sqlldr，想找个单独可用的sqlldr，在Oracle 10g Client 精简优化安装包不到12M http://download.csdn.net/download/mysky2008/4382509 下载了10g的，但本地测试出现sql*loader 704错误，不知道怎么...
expdp用于将数据库对象和数据导出到一个或多个数据泵转储文件中，而impdp则用于从这些转储文件中将数据和对象恢复到数据库中。这对于数据库备份、迁移和复制场景非常有用。 在“instantclient_11_2”目录下，你会...