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());

}

}

}