parsealipayrzipsender.java

来自「项目支付宝批量打款,采用httpclient+spring +quarz实现.」· Java 代码 · 共 181 行

package com.szhelper.pay.quartz;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.zip.ZipEntry;

import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.szhelper.pay.ApplicationContextWrapper;
import com.szhelper.pay.Constants;
import com.szhelper.pay.model.DownloadPayBean;
import com.szhelper.pay.quartz.base.BaseSender;
import com.szhelper.pay.quartz.base.Message;
import com.szhelper.pay.quartz.message.ParseAlipayRZipMessage;
import com.szhelper.pay.quartz.message.PayUploadMessage;
import com.szhelper.pay.quartz.util.FileCSVHelper;
import com.szhelper.pay.quartz.util.FileZipHelper;
import com.szhelper.pay.service.IPayProxyService;
import com.szhelper.pay.service.impl.PayProxyServiceImpl;

public class ParseAlipayRZipSender extends BaseSender {
	private static Log logger = LogFactory.getLog(ParseAlipayRZipSender.class);
	private long threadSleepPeriod = 0;
	private int clientConnTimeout;
	private Map downloadPayConfig;
	private IPayProxyService payProxyService;
	private String downloadPath;
	private String seqPreFix, dbErrLog,backupPath;
	private static BaseSender instance = new ParseAlipayRZipSender();

	public static BaseSender getInstance() {		
		return instance;
	}

	private ParseAlipayRZipSender() {
		downloadPayConfig = (Map) ApplicationContextWrapper.getInstance()
				.getBean("downloadPayConfig");
		threadSleepPeriod = Long.parseLong(downloadPayConfig.get("jobSleepOfParseRZip")
				.toString());
		logger.info(threadSleepPeriod);
		clientConnTimeout = Integer.parseInt(downloadPayConfig.get(
				"clientConnTimeout").toString());
		downloadPath = downloadPayConfig.get("downloadPath").toString();
		payProxyService = (PayProxyServiceImpl) ApplicationContextWrapper
				.getInstance().getBean("PayProxyService");
		seqPreFix = ApplicationContextWrapper.getInstance()
		.getBean("seqPreFix").toString();
		dbErrLog = downloadPayConfig.get("dbErrLog").toString();
		backupPath = downloadPayConfig.get("backupPath").toString();	
	}

	public void run() {
		while (!runFlag) {
			synchronized (this) {
				if (!queue.empty()) {
					Message message = (ParseAlipayRZipMessage) queue.pop();
					try {
						if (!send(message)) {
							message.autoIncreaseSentTimes();
							if (message.isAllowSend()) {
								queue.push(message);
							} else {
								message.setStatus(Constants.SENT_RESP_STATUS_E);
								//rollbackLog(message);
							}
						}
					} catch (Exception ex) {
						logger.error(ex);
					}
				}
				try {
					// Thread.sleep(1L * 1000L);
					Thread.sleep(threadSleepPeriod);
				} catch (Exception e) {
					logger.error(e);
				}
			}
		}
	}
	
	public boolean send(Message msg) {
		ParseAlipayRZipMessage message = (ParseAlipayRZipMessage) msg;
		boolean flag = false;
		
		try {			
			logger.info("began to analyze file [" + message.getZipname() +"]");		
			
			Map<String, byte[]> fileMap = FileZipHelper.unZip(message.getZipname());
			Set keySet = fileMap.keySet();
			String files[] = new String[keySet.size()];
			keySet.toArray(files);
			for (int i = 0; i < files.length; i++) {
				if (files[i].indexOf(Constants.POSTFIX_CSV)>0 && !Constants.ALIPAY__RESULT_CSV.equalsIgnoreCase(files[i])) {// deal with
					byte[] b = fileMap.get(files[i]);
					String fileContent = new String(b,0,b.length);					
					logger.info("after analysis,dealing with file [ "
							+ files[i] + " ] with database.");
					logger.info(fileContent);
					analyzeFileContent(fileContent, files[i]);
				}
			}
			FileZipHelper.moveFile(message.getZipname() ,backupPath);
			flag = true;
		} catch (IOException ex) {
			//rollbackLog(message);
			logger.info(ex.getMessage());
		}
		return flag;
	}

	private void analyzeFileContent(String fileContent, String csvname) {	
		String[] lines = fileContent.split("\r\n");
		if (lines != null) {
			logger.info("The csv file[" + csvname
					+ "]  which was just downloaded from Alipay includes "
					+ lines.length + " lines.");

			for (int i = 3; i < lines.length; i++) {
				String[] items = lines[i].split(Constants.ALIPAY_SEPARATOR);
				if (items != null && items.length > 0) {
					DownloadPayBean downloadPayBean = new DownloadPayBean();
					String autoId = items[0];
					if (items[0] != null && items[0].length() > 0
							&& items[0].indexOf(seqPreFix) > -1) {
						autoId = items[0].substring(items[0].indexOf(seqPreFix)
								+ seqPreFix.length());
					}
					downloadPayBean.setAutoId(new Integer(autoId));
					downloadPayBean.setStatus(items[9]);
					DownloadPayBean respBean = payProxyService
							.payResultHandledPM(downloadPayBean);
					if (respBean.getJYDM() == 1301)
						logger
								.info("database deals successfully with this record which autoid is "
										+ autoId
										+ " and status is "
										+ items[9]
										+ " with");
					else {
						logger
								.info("when database deals with this record which autoid is "
										+ autoId
										+ " and status is "
										+ items[9]
										+ " with, error.");
						FileCSVHelper.logDbErr(downloadPayBean, dbErrLog, true,
								"database error, log autoId [" + autoId + "] to " + dbErrLog);
					}
				}
			}
		}
	}
	
	/*private void rollbackLog(Message msg) {
		PayDownloadMessage message = (PayDownloadMessage) msg;
		StringBuffer buf = new StringBuffer();
		try {
			List<String> list = message.getCsvnameList();			
			for(String csv : list){
				buf.append(csv).append("\r\n");
			}
			FileCSVHelper.logUploadedFileName(buf.toString(), message
					.getUploadLog(), true,
					"because of analysis zip file error,push cvs-files-name ["
							+ buf.toString() + "] to "
							+ message.getUploadLog());
		} catch (IOException ex) {
			logger.info("rollback  cvs-files-name [" + buf.toString()
					+ "] to " + message.getUploadLog() + ",Error occurs.");
			logger.error(ex);
		}
	}*/
}