线程问题，在线等，能解决问题给100分

package com.delochi.dynamic.run;import java.io.FileInputStream;
import java.io.IOException;
import java.sql.CallableStatement;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Timer;import com.delochi.dynamic.db.DbManager;
import com.delochi.dynamic.db.DynamicDB;
import com.delochi.dynamic.db.DynamicDbManager;
import com.delochi.dynamic.model.system.AutoPutVo;
import com.delochi.excelop.web.filter.LogMode;
import com.delochi.vo.config.ConfigVo;
import com.delochi.vo.config.DynicOpVo;
import com.thoughtworks.xstream.XStream;
import com.thoughtworks.xstream.io.xml.DomDriver;
import com.wmb.file.BlandFile;
import com.wmb.file.FileOperation;public class DynamicReportAutoput implements Runnable {
private DbManager db = new DbManager();
private static DynamicReprtSingleton dynamicReprtSingleton = DynamicReprtSingleton
.getInstances();
/**
* 存放各地市名
*/
private volatile static List<String> addrList = null;
/**
* 个地市所对应的储存过程名及所需表
*/
private volatile static Map<String, List<AutoPutVo>> addrProList = null;
private volatile static Map<String, List<AutoPutVo>> runAddrProList = new LinkedHashMap<String, List<AutoPutVo>>();
/**
* 每个地市共有多少个存储过程
*/
private static Map<String, Integer> addrProCount = null;
/**
* 默认允许同时执行21个地市
*/
public static int addrMax = 21;
public static List<Timer> timerList = new ArrayList<Timer>();
/**
* 线程分配计数器
*/
private volatile static int runcount = 0;
/**
* 默认每个地市允许同时执行4个储存过程
*/
public static int addrProMax = 4;
/**
* true时，处于监控状态，false时，停止监控
*/
public static boolean isRun; public int isRuncount;
/**
* 开启线程数量
*/
public static int thread;
/**
* 记录当前日期
*/
private static String curdate; /**
* 保存日志的路径
*/
public static String logPath;
/**
* 配置文件路径
*/
public static String configPath;
/**
* 配置文件路径
*/
protected static String isRunPath; /**
* list初始化锁,解决数组初始化问题
*/
protected volatile static boolean isInit = true; private volatile static int exitSystem = 0; /**
* 初始化路径
*/
public static void init() {
try {
Properties properties = new BlandFile().init("/path.properties");
configPath = properties.getProperty("configPath");
isRunPath = properties.getProperty("isRunPath");
initConfig();
} catch (IOException e) {
e.printStackTrace();
}
} /**
* 初始化配置
*/
public static void initConfig() {
XStream xstream = new XStream(new DomDriver());
DynicOpVo dynicOpVo = new DynicOpVo();
FileInputStream fis = null;
try {
fis = new FileInputStream(configPath);
xstream.alias("dynic", DynicOpVo.class);
xstream.fromXML(fis, dynicOpVo);
addrMax = dynicOpVo.getAddrMax();
addrProMax = dynicOpVo.getAddrProMax();
logPath = dynicOpVo.getLogPath();
thread = dynicOpVo.getThread();
isInit = true;
exitSystem = 0;
} catch (Exception e) {
e.printStackTrace();
}
} /**
* 对isRun进行监听
*/
protected void isRun() {
XStream xstream = new XStream(new DomDriver());
ConfigVo configVo = new ConfigVo();
FileInputStream fis = null;
try {
fis = new FileInputStream(isRunPath);
xstream.alias("config", ConfigVo.class);
xstream.fromXML(fis, configVo);
isRun = Boolean.valueOf(configVo.getStartOrStop());
isRuncount = Integer.valueOf(configVo.getIsRun());
} catch (Exception e) {
e.printStackTrace();
}
} /**
* 初始化容器
*
* @throws SQLException
*/
public static void initAutoPut() {
try {
runcount = 0;
createLog();
String sql = "select addr_info_addr as addr from rpt_addr_info where addr_info_run = 1";
// 初始化addrList
addrList = dynamicReprtSingleton.getAddrList(sql);
sql = "select * from (SELECT rpt_addr_table_pro.addr_pro_id as id,"
+ "rpt_addr_info.addr_info_addr as addr,rpt_addr_table_pro.addr_pro_name as addr_pro,"
+ "rpt_addr_table_pro.addr_pro_need_table as need_table FROM rpt_addr_info ,"
+ "rpt_addr_table_pro WHERE rpt_addr_info.addr_info_id =  rpt_addr_table_pro.addr_pro_aiid AND "
+ "rpt_addr_info.addr_info_run =  '1' and rpt_addr_table_pro.ADDR_PRO_RUN = '1') addr_pro ";
addrProList = dynamicReprtSingleton.getAddrProAndTableList(
addrList, sql);
sql = "select count(*) as num  from (SELECT rpt_addr_table_pro.addr_pro_id as id,"
+ "rpt_addr_info.addr_info_addr as addr,rpt_addr_table_pro.addr_pro_name as addr_pro,"
+ "rpt_addr_table_pro.addr_pro_need_table as need_table FROM rpt_addr_info ,"
+ "rpt_addr_table_pro WHERE rpt_addr_info.addr_info_id =  rpt_addr_table_pro.addr_pro_aiid AND "
+ "rpt_addr_info.addr_info_run =  '1' and rpt_addr_table_pro.ADDR_PRO_RUN = '1') addr_pro ";
addrProCount = dynamicReprtSingleton.getAddrProCount(addrList, sql);
if (addrProList.isEmpty() && addrList.isEmpty()) {
LogMode.log(LogMode.sd.format(System.currentTimeMillis())
+ "  初始化数组addrList、addrProList失败！数组为null。请确认数据库中有数据");
initAutoPut();
} else {
LogMode.log(LogMode.sd.format(System.currentTimeMillis())
+ "  初始化数组addrProList成功！");
}
} catch (Exception e) {
System.out.println("====");
}
} /**
* 创建日志
*/
private static void createLog() {
curdate = LogMode.fileLog.format(System.currentTimeMillis()).toString();
try {
FileOperation.createMulu(logPath);
LogMode.createFile(logPath, curdate + "_autoput");
} catch (Exception e) {
System.out.println("创建日志目录出错");
}
} public void run() {
while (true) {
try {
isRun();
if (isRun) {
initializationList();
printInfo();
LogMode.log(LogMode.sd.format(System.currentTimeMillis())+ "  线程状态："+Thread.currentThread().getState());
if ((addrProList.size() == addrList.size())
&& (addrList.size() == runcount))
runcount = 0;
if (runcount >= addrList.size())
runcount = 0;
addrRun();
} else {
// exitAutoSys();
this.exitSystem += 1;
if (this.exitSystem == thread) {
dynamicReprtSingleton.config(isRunPath);
LogMode.log(LogMode.sd.format(System
.currentTimeMillis())
+ "  退出监控系统！");
}
break;
}
} catch (Exception e) {
LogMode.log(e, LogMode.sd.format(System.currentTimeMillis())
+ "  run最大的异常");
}
try {
Thread.sleep(10 * 1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
/**
* 扫描地市
*/
private void addrRun() {
try {
for (int count = runcount; count < addrList.size(); count++) {
String tempAddr = addrList.get(count);
System.out.println("扫描 :" + tempAddr);
if (addrList.size() == 0) {
continue;
}
try {
runcount++;
if (runcount >= addrList.size())
runcount = 0;
if (checkProPut(tempAddr)) {
addrProRun(tempAddr);
}
} catch (Exception e) {
continue;
}
}
} catch (Exception e) {
LogMode.log(LogMode.sd.format(System.currentTimeMillis())
+ "  数组addrList为空！重新初始化...");
}
} /**
* 扫描地市所对应的存储过程
*
* @param tempAddr
* @throws Exception
*/
private void addrProRun(String tempAddr) throws Exception {
if (addrProList.containsKey(tempAddr)
&& !(addrProList.get(tempAddr)).isEmpty()) {
for (int count = 0; count < addrProList.get(tempAddr).size(); count++) {
if (checkRunAddrPro(tempAddr)
&& checkProTable(addrProList.get(tempAddr).get(count),
count)) {
String temp = addrProList.get(tempAddr).get(count)
.getAddr_pro();
removeAddrPro(tempAddr, addrProList.get(tempAddr), count);// 移除该地市的储存过程
execute(tempAddr, temp);
}
}
} else {
synchronized (this) {
if (addrList.contains(tempAddr)) {
addrList.remove(tempAddr);
System.out.println(tempAddr + "\t为空!");
}
}
}
} /**
* 退出系统
*/
private void exitAutoSys() {
for (int count = 0; count < timerList.size(); count++) {
System.out.println("0000000");
timerList.get(count).cancel();
}
dynamicReprtSingleton.config(isRunPath);
LogMode
.log(LogMode.sd.format(System.currentTimeMillis())
+ "  退出监控系统！");
} /**
* 在控制台输入信息
*/
private void printInfo() {
System.out.print(Thread.currentThread().getName() + "\t");
System.out.print(LogMode.sysdate.format(System.currentTimeMillis())
+ "\t");
System.out.println("addrProList:" + addrProList.size() + "  addrList"
+ addrList.size() + " runcount:" + runcount
+ " runAddrProList：" + runAddrProList.size());
System.out.println("addrList(0):" + addrList.get(0));
} /**
* 初始化
*/
private void initializationList() {
System.out.println("======" + isInit);
if (isInit) {
isInit = false;
try {
if (!(LogMode.fileLog.format(System.currentTimeMillis()))
.toString().equalsIgnoreCase(curdate)) {
try {
addrList.clear();
addrProList.clear();
} catch (Exception e) {
}
/*
* try { System.out.println("等待中......"); Thread.sleep(2 *
* (60 * 60 * 1000) + 30 * 60 * 1000);// 2小时30分 } catch
* (InterruptedException e) { e.printStackTrace(); }
*/
if (checkListNull()) {
// System.gc();
createLog();
initAutoPut();
LogMode.log(LogMode.sd.format(System
.currentTimeMillis())
+ "  日志重新初始化数组成功！");
}
}
if (addrProList.size() == 0 || addrList.size() == 0) {
if (checkListNull()
&& (LogMode.fileLog.format(System
.currentTimeMillis())).toString()
.equalsIgnoreCase(curdate)) {
System.out.println(" 重新初始化数组成功！");
initAutoPut();
}
}
} catch (Exception e) {
e.printStackTrace();
}
isInit = true;
}
} /*------------------------------------------------------------------------------------------*/
/**
* @return 如果addrList大于0，则返回false
*/
private synchronized boolean checkListNull() {
if (addrList.size() > 0)
return false;
return true;
} /**
* 检测没有存储过程的地市或者存储过程已经下发完的地市
*
* @param addr
* @return
*/
private boolean checkProPut(String addr) {
synchronized (this) {
ResultSet rs = db
.executeQuery("select count(*) as num from rpt_put_info_log where trunc(put_info_statedate)=trunc(sysdate) "
+ "and trim(put_info_error_msg) is null and put_info_addr_name='"
+ addr + "'");
int proNum;
try {
if (rs.next()) {
proNum = rs.getInt("num");
if (addrProCount.get(addr) == 0) {
System.out.println("移除存储过程为零的" + addr);
removeNull(addr);
// LogMode.log(LogMode.sd.format(System.currentTimeMillis())+
// "\t移除存储过程为零的" + addr);
return false;
} else if (addrProCount.get(addr) == proNum) {
System.out.println("移除存储过程已经下发完的" + addr);
removeNull(addr);
LogMode.log(LogMode.sd.format(System
.currentTimeMillis())
+ "\t移除存储过程已经下发完的" + addr);
return false;
} else
return true;
}
} catch (Exception e) {
System.out.println("空指针异常" + addr);
removeNull(addr);
} finally {
try {
rs.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
return false;
}
} /**
* @param tempAddr移除没有存储过程的地市或者存储过程已经下发完的地市
*/
private void removeNull(String tempAddr) {
try {
synchronized (this) {
addrProList.remove(tempAddr);
addrList.remove(tempAddr);
runcount = 0;
}
} catch (Exception e) {
e.printStackTrace();
LogMode.log(e, LogMode.sd.format(System.currentTimeMillis())
+ "  removeNull最大的异常" + tempAddr);
}
} /**
* @param autoPutVo
* @return 查看储存过程所需要的表是否到齐，到齐返回true;
*/
private boolean checkProTable(AutoPutVo autoPutVo, int count) {
if (checkLogPro(autoPutVo.getAddr(), autoPutVo.getAddr_pro(), count))
return checkNeedTable(autoPutVo.getPro_need_table(), autoPutVo
.getAddr());
return false;
} /**
* @param needTable
* @return 查看执行改储存过程所需要的表是否已经到齐
*/
private boolean checkNeedTable(String needTable, String addr) {
String sql = "select count(*) as num from (select distinct table_name From dw_Log "
+ "where trunc(start_date) = to_date('"
+ LogMode.date.format(System.currentTimeMillis())
+ "','yyyy-mm-dd') and trunc(end_date) is not null "
+ "and row_num > 1 and trim(state) = '1' and trim(error_depict) is null) a "
+ "where upper(trim(table_name)) in(upper(trim('"
+ needTable.replace("#", "')),upper(trim('") + "')))";
ResultSet rs = null;
try {
rs = new DynamicDbManager().executeQuery(addr, sql);
int num;
if (rs.next()) {
num = rs.getInt("num");
if ((needTable.split("#").length) == num) {
LogMode.log(needTable + "\t"
+ (needTable.split("#").length) + "\tnum" + num);
return true;
} else {
return false;
}
} else
return false;
} catch (Exception e) {
e.printStackTrace();
return false;
} finally {
try {
rs.close();
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
}
/**
* @param addr
* @param pro
* @return 判断该地市所对应的此储存过程是否已经下发,如果下发了就返回false
*/
private boolean checkLogPro(String addr, String pro, int count) {
ResultSet rs = null;
try {
rs = db
.executeQuery("select count(*) as num from rpt_put_info_log where trunc(put_info_statedate)=trunc(sysdate) and trim(put_info_addr_name)=trim('"
+ addr
+ "') and upper(trim(put_info_pro_name))= upper(trim('"
+ pro + "'))");
if (rs.next()) {
if (rs.getInt("num") == 0)
return true;
else
removeAddrPro(addr, addrProList.get(addr), count);// 如果改存储过程已经下发,移除该地市的储存过程
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
rs.close();
} catch (Exception e) {
e.printStackTrace();
}
}
return false;
} /**
* @param addr
* @param list
* @param count
*/
private void removeAddrPro(String addr, List<AutoPutVo> list, int count) {
try {
synchronized (this) {
addrProList.get(addr).remove(count);
System.out.println("移除addrProList" + addr);
LogMode.log(LogMode.sd.format(System.currentTimeMillis())
+ "\t移除addrProList" + addr);
if (addrProList.get(addr).isEmpty()) {
addrProList.remove(addr);
addrList.remove(addr);
runcount = 0;
LogMode.log(LogMode.sd.format(System.currentTimeMillis())
+ "\t移除addrList" + addr);
System.out.println("移除addrList" + addr);
}
}
} catch (Exception e) {
e.printStackTrace();
LogMode.log(e, LogMode.sd.format(System.currentTimeMillis())
+ "  removeAddrPro最大的异常:" + addr);
}
} /**
* @param addr
* @return 检测当前正在执行的地市是否已经满了，满了返回false；检测当前每个地市正在执行的储存过程是否已经满了，满了返回false;
*/
private boolean checkRunAddrPro(String addr) {
synchronized (this) {
try {
if (dynamicReprtSingleton.getAddrCount() < addrMax)
if (dynamicReprtSingleton.getAddrProCount(addr) < addrProMax)
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
return false;
} /**
* @param addr地市
* @param pro地市所对应的储存过程
*/
private void execute(String addr, String pro) {
Connection conn = null;
String sql = "insert into rpt_put_info_log(put_info_id,put_info_pro_name,put_info_addr_name,"
+ "put_info_statedate) values(put_info_idsque.nextval,'"
+ pro
+ "','" + addr + "',sysdate)";
try {
conn = new DynamicDB().getCon(addr);
if (conn != null && checkRunAddrPro(addr)) {
if (dynamicReprtSingleton.updatePutInfo(sql, addr, pro)) {
int id = dynamicReprtSingleton
.getPutInfo("select max(put_info_id) as log_id from rpt_put_info_log");
sql = "{call " + pro + "}";
callPro(conn, sql, addr, pro, id);
}
}
} catch (ClassNotFoundException e) {
LogMode.log(e, LogMode.sd.format(System.currentTimeMillis())
+ "  无法注册 JDBC " + addr + "驱动程序>>" + e);
} catch (Exception e) {
LogMode.log(e, LogMode.sd.format(System.currentTimeMillis())
+ "  获取" + addr + "数据库连接出错>>" + e);
} finally {
try {
if (conn != null)
conn.close();
} catch (Exception e) {
LogMode.log(LogMode.sd.format(System.currentTimeMillis())
+ "关闭" + addr + "连接时出错");
}
}
} /**
* @param conn
* @param sql
* @param addr
* @param pro
* @param id
* @return 执行储存过程，成功返回true，
*/
private void callPro(Connection conn, String sql, String addr, String pro,
int id) {
// boolean isComp = false;
CallableStatement cs = null;
try {
cs = conn.prepareCall(sql);
System.out.println("  开始执行" + addr + "存储过程。存储过程为：" + pro);
LogMode.log(LogMode.sd.format(System.currentTimeMillis())
+ "  开始执行" + addr + "存储过程。存储过程为：" + pro);
cs.execute();
// isComp = true;
System.out.println("  完成" + addr + "存储过程。存储过程为：" + pro);
LogMode.log(LogMode.sd.format(System.currentTimeMillis()) + "  完成"
+ addr + "存储过程。存储过程为：" + pro);
dynamicReprtSingleton.updatePutInfo(
"update rpt_put_info_log set put_info_enddate = sysdate where put_info_id="
+ id, addr, pro);
} catch (Exception e) {
LogMode.log(e, LogMode.sd.format(System.currentTimeMillis())
+ "  执行" + addr + "存储过程时出错。存储过程为：" + pro + ">>" + e);
dynamicReprtSingleton.updatePutInfo(
"update rpt_put_info_log set put_info_error_msg='"
+ LogMode.sd.format(System.currentTimeMillis())
+ "  执行" + addr + "存储过程时出错。存储过程为：" + pro + ">>"
+ dynamicReprtSingleton.replaceStr(e.toString())
+ "' where put_info_id=" + id, addr, pro);
} finally {
try {
cs.close();
} catch (Exception e) {
LogMode.log(LogMode.sd.format(System.currentTimeMillis())
+ "  关闭" + addr + "CallableStatement时出错");
e.printStackTrace();
}
}
// return isComp;
}}

调试易

线程问题，在线等，能解决问题给100分

解决方案 »