java通过ftp方式读取文件，并解析入库

对于自己是个总结，下次使用的时候直接copy，要注意的地方是ftp访问文件的方式，访问的是ftp用户的home目录，而不是从绝对路径访问。

private final static String DELETEUSER = "delete from customermanager_user where cmanagerno in ";
private final static String QUERYUSER = "select cmanagerno from customermanager_user where cmanagerno in ";
private final static String INSERTUSER = "insert into customermanager_user(cmanagerno, cmanagercity, cmanagercontry, cmanagername, cmanagercontact, cmanagerstatus) values(?,?,?,?,?,?)";
private CustomManagerJobDAO managerJobDAO;
public CustomManagerJobDAO getManagerJobDAO() {
return managerJobDAO;
}

public void setManagerJobDAO(CustomManagerJobDAO managerJobDAO) {
this.managerJobDAO = managerJobDAO;
}

/**
* customermanager_group 表增删查
*/
private final static String DELETEGROUP = "delete from customermanager_group where id in ";
private final static String QUERYGROUP = "select id from customermanager_group where id in ";
private final static String INSERTGROUP = "insert into customermanager_group(id, cmgroupno, cmgroupcity, cmgroupcontry, cmgroupcustomername, cmgroupcorno, cmgroupcoradd, cmgroupcustomertype, cmgroupiffirst, cmgroupcustomerlevel, cmgroupcontact, cmgroupslalevel, cmgroupaddress, cmgroupsatus) values(?,?,?,?,?,?,?,?,?,?,?,?,?,?)";

private static Integer file1length = Integer.valueOf(6);
private static Integer file2length = Integer.valueOf(14);

private final static String REMOVEFLAG = "1";
private final static String UPDATEFLAG = "0";

private List removekeylist = new LinkedList();
private List insertkeylist = new LinkedList();
private List insertlist = new LinkedList();

static {
String file1 = StaticMethod.nullObject2String(XmlManage.getFile(
"/config/customanager-ftp-boss.xml").getProperty("file1length"));
if (file1 != null && file1.length() >= 0) {
file1length = Integer.valueOf(file1);
}
String file2 = StaticMethod.nullObject2String(XmlManage.getFile(
"/config/customanager-ftp-boss.xml").getProperty("file2length"));
if (file1 != null && file2.length() >= 0) {
file2length = Integer.valueOf(file2);
}
}

public void parseData(String filepath) {
int table1size = file1length.intValue();
int table2size = file2length.intValue();
int count = 1;
try {
long begintime = System.currentTimeMillis();
URL url = new URL(filepath);
URLConnection urlconn = url.openConnection();
BufferedReader br = new BufferedReader(new InputStreamReader(urlconn.getInputStream()));
String line = null;
String keyflag = null; // 移除和删除标记
while ((line = br.readLine()) != null) {
count++;
String[] fileds = line.split("\\|");
if (table1size == fileds.length) {
keyflag = fileds[table1size - 1];
if (REMOVEFLAG.equals(keyflag)) {
removekeylist.add(fileds[0]);
} else if (UPDATEFLAG.equals(keyflag)) {
insertkeylist.add(fileds[0]);
fileds = changeAreaNametoId(fileds);
insertlist.add(fileds);// 再把所有数组给存起来
}
} else if (table2size == fileds.length) {
keyflag = fileds[table2size - 1];
if (REMOVEFLAG.equals(keyflag)) {
removekeylist.add(fileds[0]);
} else if (UPDATEFLAG.equals(keyflag)) {
insertkeylist.add(fileds[0]);
fileds = changeAreaNametoId(fileds);
insertlist.add(fileds);// 再把所有数组给存起来
}
}
if (count % 5000 == 0) {
queryAndDelete(removekeylist, insertlist); // 先查一把！然后再删
queryAndInsert(insertkeylist, insertlist);// 先查一把！然后再新增
removekeylist.clear();
insertkeylist.clear();
insertlist.clear();
}
queryAndDelete(removekeylist, insertlist); // 先查一把！然后再删
queryAndInsert(insertkeylist, insertlist);// 先查一把！然后再新增
removekeylist.clear();
insertkeylist.clear();
insertlist.clear();
}
br.close();
long endtime = System.currentTimeMillis();
System.out.println((endtime - begintime) / 3600);
} catch (FileNotFoundException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
}

public void getFileByFtp() {
String ftpserver = StaticMethod.nullObject2String(XmlManage.getFile("/config/customanager-ftp-boss.xml").getProperty("ftpserver"));
String userLogin = StaticMethod.nullObject2String(XmlManage.getFile("/config/customanager-ftp-boss.xml").getProperty("ftpuserLogin"));
String pwdLogin = StaticMethod.nullObject2String(XmlManage.getFile("/config/customanager-ftp-boss.xml").getProperty("ftppwdLogin"));
String serverPath = StaticMethod.nullObject2String(XmlManage.getFile("/config/customanager-ftp-boss.xml").getProperty("ftpserverPath"));
try {
// ftp服务器上下载每天的工单流水文件
String retMessage = connectToFtpServer(ftpserver, userLogin,
pwdLogin);
if (!retMessage.equals("connect success")) {
System.out.println("下载时文件时无法和FTP服务器连接");
}
String filename = StaticMethod.getYYYYMMDD(StaticMethod
.getLocalString(-1));
System.out.println("filename===" + filename);

String filepath1 = "ftp://" + userLogin + ":" + pwdLogin + "@"
+ ftpserver + serverPath + filename + "_001.txt";
String filepath2 = "ftp://" + userLogin + ":" + pwdLogin + "@"
+ ftpserver + serverPath + filename + "_002.txt";
System.out.println("path===" + filepath1);
URL url = new URL(filepath1);
URLConnection urlconn = url.openConnection();
BufferedReader br = new BufferedReader(new InputStreamReader(
urlconn.getInputStream()));
String line = null;
while ((line = br.readLine()) != null) {
System.out.println("line======" + line);
}
parseData(filepath1);
parseData(filepath2);
br.close();

closeFtpConnect();
} catch (MalformedURLException e1) {
e1.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
}

/**
* 把地市名称换成id
*
* @param keylist
*/
public String[] changeAreaNametoId(String[] records) {
if (file1length.intValue() == records.length) { // 数组长度为6，为第一张表
records[1] = getAreaid(records[1]);
} else if (file2length.intValue() == records.length) {
records[2] = getAreaid(records[2]);
}
return records;
}

/**
* 对数据进行查询，如果存在则删除
*
* @param keylist
*/
public void queryAndDelete(List keylist, List datalist) {
String sql = null;
String[] records = null;
if (datalist != null && datalist.size() > 0) {
records = (String[]) datalist.get(0);
}
List retlist = null;
try {
if (file1length.intValue() == records.length) { // 数组长度为6，为第一张表
sql = getExecuteSql(QUERYUSER, keylist);
retlist = this.getManagerJobDAO().getKeyList(sql); // 根据key把库中已经存在的记录查出来。
if (retlist != null && retlist.size() > 0) {
sql = getExecuteSql(DELETEUSER, retlist); // 组装成需要的sql
System.out.println("delete id from customermanager_user where flag = 1:===" + retlist);
this.getManagerJobDAO().delteRecord(sql);// 根据key把库中已经存在的删除
}
} else if (file2length.intValue() == records.length) {
sql = getExecuteSql(QUERYGROUP, keylist);
retlist = this.getManagerJobDAO().getKeyList(sql); // 根据key把库中已经存在的记录查出来。
if (retlist != null && retlist.size() > 0) {
sql = getExecuteSql(DELETEGROUP, retlist);
System.out.println("delete id from customermanager_group where flag = 1 ===="+ sql);
this.getManagerJobDAO().delteRecord(sql);
}
}
} catch (SQLException e) {
e.printStackTrace();
}
}

public void queryAndInsert(List keylist, List datalist) {
String sql = null;
String[] records = null;
if (datalist != null && datalist.size() > 0) {
records = (String[]) datalist.get(0);
}
List retlist = new ArrayList();
try {
if (file1length.intValue() == records.length && retlist != null) { // 数组长度为6，为第一张表
sql = getExecuteSql(QUERYUSER, keylist);
retlist = this.getManagerJobDAO().getKeyList(sql);  // 根据key把库中已经存在的记录查出来。
if (retlist.size() == 0) { // 如果没有查到一个，则全部入库
this.getManagerJobDAO().insertTable1(INSERTUSER, datalist); // 把所有字段全部入库
} else {
sql = getExecuteSql(DELETEUSER, retlist); // 如果存在重复，则更新和新增分开操作
System.out.println("delete id from customermanager_user where flag = 1 ===="+ retlist);
this.getManagerJobDAO().delteRecord(sql);// 根据key把库中已经存在的删除
System.out.println("insert * into customermanager_user where flag = 0 ===="+ datalist);
this.getManagerJobDAO().insertTable1(INSERTUSER, datalist); // 把所有字段全部入库
}
} else if (file2length.intValue() == records.length && retlist != null) {
sql = getExecuteSql(QUERYGROUP, keylist);
retlist = this.getManagerJobDAO().getKeyList(sql);
if (retlist.size() == 0) {
this.getManagerJobDAO().insertTable2(INSERTGROUP, datalist);
} else {
sql = getExecuteSql(DELETEGROUP, retlist);
System.out.println("delete id from customermanager_group where flag = 1 ===="+ retlist);
this.getManagerJobDAO().delteRecord(sql);
System.out.println("insert * into customermanager_group where flag = 0 ===="+ datalist);
this.getManagerJobDAO().insertTable2(INSERTGROUP, datalist);
}
}
} catch (SQLException e) {
e.printStackTrace();
}
}

public static String getExecuteSql(final String querySql, List retList) {
String record = null;
String keys = null;
StringBuffer sqlsb = new StringBuffer();
for (int i = 0; i < retList.size(); i++) {
keys = StaticMethod.nullObject2String(retList.get(i));
sqlsb.append("'").append(keys).append("',");
}
record = sqlsb.toString();
String sqlwhere = record.substring(0, record.lastIndexOf(","));
return querySql + "(" + sqlwhere + ")";
}

public Logger logger = Logger.getLogger(this.getClass());
FtpClient ftpClient=null;
/*
* 获取FTP服务
*/
public String connectServer( String ip, int port, String user,
String password, String path) {
String returnResult = "true";
try {
ftpClient = new FtpClient();
ftpClient.openServer(ip, port);
ftpClient.login(user, password);
if (path.length() != 0)
ftpClient.cd(path);
ftpClient.binary();
} catch (IOException ex) {
returnResult = "false";
}
return returnResult;
}

private String connectToFtpServer(String ftpserver, String userLogin,
String pwdLogin) {
FtpClient fc = new FtpClient();
if ((ftpserver == null) || (ftpserver.equals("")))
return "FTP服务器名设置不正确!";
try {
fc.openServer(ftpserver);
fc.login(userLogin, pwdLogin);
fc.binary();
} catch (FtpLoginException e) {
return "没有与FTP服务器连接的权限,或用户名密码设置不正确!";
} catch (IOException e) {
return "与FTP服务器连接失败!";
} catch (SecurityException e) {
return "没有权限与FTP服务器连接";
}
return "connect success";
}

/*
* 关闭FTP
*/
public void closeFtpConnect() {
try {
ftpClient.closeServer();
} catch (IOException ex) {
System.out.println("关闭时出错======："+ex.getMessage());
}
}
}