前言：上一篇多线程系列之 java多线程的个人理解(一) 讲到了线程、进程、多线程的基本概念，以及多线程在java中的基本实现方式，本篇主要接着上一篇继续讲述多线程在实际项目中的应用以及遇到的诸多问题和解决方案
文章结构：
- 多线程在实际项目中应用
- 多线程的优缺点
1.多线程在实际项目中应用
项目分享(一)
背景：重庆移动代维管理系统项目，主要负责对重庆移动各代维公司，分公司，代维人员，以及各类代维业务和资产的统筹管理；其中的装维管理模块，是在代维系统中占有一席之地，主要保障移动宽带装机的线上流程以及业务顺利运行；在装维管理中，装机宽带业务在代维系统被抽象成了的工单的形式，每张工单又对应了唯一的流程实体。在整个流转过程中，有很多关键环节，其中就有一个环节叫竣工校验归档。他要完成的工作是当装机工单流程从完成安装到归档前，应该通过调接口的方式校验工单对应的宽带账号是否有上网记录。
遇到的问题：前期的做法是用单线程异步的方式调用目标系统的接口，返回报文来完成业务。按照这样的设计思路系统平稳的运行了一段时间；突然有一天开始，代维系统的运维人员频繁反映，装机工单有很多单子堵塞在竣工校验归档环节。通过日志分析并定位是单线程导致的执行效率过慢，所以考虑到使用多线程，分批次完成工单的校验任务。
具体解决思路：在业务背景下，由于每张工单在进入校验队列后在后台数据库中都保存了一个执行时间，所以，通过与当前时间比较是否应该去调接口校验，前期由于是单线程，考虑的问题很少。后来利用多线程，将多张进入队列的工单分批次处理(即将执行时间的Long型数据对N取余，假使有N+1个线程在并发执行），这样每个线程所分配到的工单都是不同的，不会存在并发修改等多线程安全问题，当然这里还是以继承Thread类为例：
线程类：
public class SmsSendThread extends Thread {
private DataTable sendDt;
private String SQL_DELETE_RECORD = "delete FROM BS_T_SM_SEND_SMS WHERE PID=?";
@Override
public void run() {//跑一次就结束
if(sendDt!=null && sendDt.length()>0){
DataAdapter dataAdapter = new DataAdapter();
Integer counts = sendDt.length();
DataRow dr = null;
DataTable updateDt = null;
DataTable hisDt = null;
DataRow hisDr = null;
DataRow updateDr = null;
for(int i =0 ; i<counts ; i++){
updateDt = new DataTable("BS_T_SM_SEND_SMS");
updateDt.setPrimaryKey(new String[]{"PID"});
hisDt = new DataTable("BS_T_SM_SEND_SMS_LOG");
dr = sendDt.getDataRow(i);
//调用发送短信接口
JSONObject jsonParam = new JSONObject();
jsonParam.put("moblie", dr.getString("MOBILE"));
jsonParam.put("message", dr.getString("CONTENT"));
jsonParam.put("priority", dr.getString("PRIORITY"));
JSONObject returnJson = SmsRestFullClient.postDataToInterface(SmsNewConstants.URL_SEND_SMS, jsonParam.toString(), null,"POST");
String outResult = returnJson.getString("outResult");
String outPut = returnJson.getString("outPut");
int maxSendNum = dr.getInt("MAX_SEND_NUM");
if(maxSendNum==0)maxSendNum=maxSendNum+1;//最少一次
int sendNum = dr.getInt("MAX_SEND_NUM");
sendNum = sendNum+1;
dr.put("SEND_TIME", TimeUtils.getCurrentTime());
dr.put("SEND_NUM", sendNum);
dr.put("REMATK", returnJson);
hisDr = dr.newRowClone();//历史记录
updateDr = dr.newRowClone();
if(SmsRestFullClient.OUTRESULT_SUCC.equals(outResult) && outPut!=null && outPut.contains("serial")){//成功后等待另一线程查询状态
JSONObject serial = JSONObject.fromObject(outPut);
String serialStr = serial.getString("serial");
updateDr.put("SEND_FLAG", SmsNewConstants.SEND_FLAG_SEARCHWAIT);//等待查询
updateDr.put("PRE_SEND_TIME", TimeUtils.getCurrentTime()+120);//两分钟后
updateDr.put("SERIAL", serialStr);
updateDr.put("SEND_NUM", 0);//重置次数
updateDt.putDataRow(updateDr);
dataAdapter.executeUpdate(updateDt);
}else{
if(sendNum<maxSendNum){//继续发送，有发送次数变化，需保存
updateDt.putDataRow(updateDr);
dataAdapter.executeUpdate(updateDt);
}else{//删除，不再发送短信
dataAdapter.execute(SQL_DELETE_RECORD, new Object[]{dr.getString("PID")});
}
hisDr.put("SEND_FLAG", SmsNewConstants.SEND_FLAG_SENDFAIL);//发送失败
}
//插入历史记录表
hisDt.putDataRow(hisDr);
dataAdapter.executeAdd(hisDt);
}
}
}
public DataTable getSendDt() {
return sendDt;
}
public void setSendDt(DataTable sendDt) {
this.sendDt = sendDt;
}
}
SQL查询：
<sqlquery name="SQL_Thread_Eoms2Prov4OldRadius.query"><!-- Eoms2Prov 接口数据查询 集中处理查询失败的 -->
<select>
<![CDATA[ SELECT /*COUNT*/ * /*COUNT*/ FROM (
SELECT
t.*,
mod(RECORDTIME,10) threadid
FROM BS_T_WF4_FTTH_TOINTERFACE t
WHERE RESULTTYPECODE ='RADIUSFILE'
$customwhere$
) WHERE ROWNUM <100 $customwhere1$
/*ORDERBY*/ ORDER BY /*ORDERBY*/ RECORDTIME ASC ]]>
</select>
<customwhere name="customwhere" prepend="and" >
<field prepend="and" operator="=" colname="flag" value="#flag#"/>
<field prepend="and" operator=">" colname="calls" value="#calls#"/>
<field prepend="and" operator="<" colname="recordtime" value="#recordtime#"/>
</customwhere> <customwhere name="customwhere1" prepend="and" >
<field prepend="threadid" operator="=" colname="threadid" value="#threadid#"/>
</customwhere>
</sqlquery>
这里没有直接在main方法中new 线程，而且采用配置表的方式，通过java反射机制动态的生成线程对象并执行。主要是为了后期的多线程智能化管理
Thread管理类：
主要功能：完成对多种线程的创建与启动
/**
* 线程智能化管理
* @author zhj
*
*/
public class ThreadManageAI { private static String projectInfo = null; static Map<String,Thread> threadMap = new HashMap<String,Thread>();
/**
* 获取线程SQL
*/
String SQL_GET_THREADINFO_INDB = "SELECT * FROM BS_T_SM_THREADMANAGE WHERE SERVERINFO=? AND THREADFLAG=1 ORDER BY THREADNAME"; QueryAdapter queryAdapter = new QueryAdapter();
DataAdapter dataAdapter = new DataAdapter(); /**
* 线程监听
*/
public void threadMonitor(){
if(projectInfo==null || "".equals(projectInfo)){
projectInfo = PropertiesUtils.getProperty("ThreadManageAI.projectInfo");
}
DataTable logdt = new DataTable("BS_T_SM_THREADMANAGE_LOG");
DataRow logdr = null;
if(projectInfo!=null && !"".equals(projectInfo)){
DataTable dt = queryAdapter.executeQuery(SQL_GET_THREADINFO_INDB, projectInfo);
if(dt!=null && dt.length()>0){
DataRow dr = null;
String threadName = null;
String clazz = null;
String paramJsonStr = null;
String logDesc = null;
boolean isAlive = false;
for(int i=0,j=dt.length();i<j;i++){
isAlive = false;
dr = dt.getDataRow(i);
threadName = dr.getString("THREADNAME");
threadName = projectInfo+"."+threadName;
Thread t = threadMap.get(threadName);
if(t!=null){
isAlive = t.isAlive();
}else{
isAlive = false;
} if(!isAlive){//需要重启线程
//重启线程
paramJsonStr = dr.getString("PARAMJSON");//获取线程参数
clazz = dr.getString("CLAZZ");//线程类 Class XXthreadClass = null;
try {
XXthreadClass = Class.forName(clazz);
} catch (ClassNotFoundException e) {
XXthreadClass = null;
e.printStackTrace();
}
if(XXthreadClass!=null){
Object xxthread = null;
try {
xxthread = XXthreadClass.newInstance();
} catch (InstantiationException e) {
xxthread = null;
e.printStackTrace();
} catch (IllegalAccessException e) {
xxthread = null;
e.printStackTrace();
}
if(xxthread!=null){
boolean isCanStart = true;//默认可以启动线程
if(paramJsonStr!=null && !"".equals(paramJsonStr)){
JSONObject paramJson = JSONObject.fromObject(paramJsonStr);
Method method = null;
if(paramJson==null||paramJson.size()==0){
//参数不为空，但是有错误，不能启动
logDesc = "参数不为空，但是有错误，不能启动";
isCanStart = false;
}else{
try { Set<String> keys = paramJson.keySet();
for(String key:keys){
//例子，参数名threadid，方法名：setThreadid(String threadid)
String methodName = "set"+key.substring(0,1).toUpperCase()+key.substring(1); method = XXthreadClass.getMethod(methodName, String.class);
try {
method.invoke(xxthread, paramJson.getString(key)); } catch (IllegalArgumentException e) {
isCanStart = false;
e.printStackTrace();
} catch (IllegalAccessException e) {
isCanStart = false;
e.printStackTrace();
} catch (InvocationTargetException e) {
isCanStart = false;
e.printStackTrace();
}
} } catch (SecurityException e) {
isCanStart = false;
e.printStackTrace();
} catch (NoSuchMethodException e) {
isCanStart = false;
e.printStackTrace();
}
}
}
if(isCanStart){
boolean isStarted = true;
try {
Method startMethod = XXthreadClass.getMethod("start");
try {
startMethod.invoke(xxthread);
} catch (IllegalArgumentException e) {
isStarted = false;
e.printStackTrace();
} catch (IllegalAccessException e) {
isStarted = false;
e.printStackTrace();
} catch (InvocationTargetException e) {
isStarted = false;
e.printStackTrace();
}
} catch (SecurityException e) {
isStarted = false;
e.printStackTrace();
} catch (NoSuchMethodException e) {
isStarted = false;
e.printStackTrace();
}
if(isStarted){
t = (Thread)xxthread;
threadMap.put(threadName,t);
//线程启动成功
logDesc = "线程启动成功";
}else{
//线程启动失败
logDesc = "线程启动失败";
}
}else{
//参数设置失败
logDesc = "参数设置失败";
} }else{
//线程无法实例化
logDesc = "线程无法实例化";
}
}else{
//CLAZZ有误，无法加载类
logDesc = "CLAZZ有误，无法加载类";
} }else{
//线程存在，打印日志
logDesc = "线程还存活";
} HashMap<String,Object> rowMap = dr.getRowHashMap();
logdr = new DataRow();
for(String key:rowMap.keySet()){
Object value = rowMap.get(key);
if(value!=null){
logdr.put(key, value);
}
}
logdr.put("PID", UUIDGenerator.getUUIDoffSpace());
logdr.put("LOGTIME", TimeUtils.getCurrentTime());
logdr.put("LOGDESC", logDesc); logdt.putDataRow(logdr);
}
}else{
logdr = new DataRow();
logdr.put("PID", UUIDGenerator.getUUIDoffSpace());
logdr.put("LOGTIME", TimeUtils.getCurrentTime());
logdr.put("LOGDESC", "没有需要启动的线程");
logdr.put("SERVERINFO", projectInfo);
logdt.putDataRow(logdr);
} }else{
logdr = new DataRow();
logdr.put("PID", UUIDGenerator.getUUIDoffSpace());
logdr.put("LOGTIME", TimeUtils.getCurrentTime());
logdr.put("LOGDESC", "projectInfo为空，没有需要启动的线程");
logdt.putDataRow(logdr);
} dataAdapter.executeAdd(logdt);
}
}
这样，我们只需要在目标表中配置线程ID 、线程的业务类型、线程实现类的全路径，以及入参，就可以通过反射的方式创建并启动多个线程，就这样，竣工归档校验功能第二版改造到一段落。通过运维人员反映，部分解决了工单由于等待太久而阻塞的问题。
然后在系统正常运行大概有一个月后，运维人员又一次发现了一个重大问题，那就是有很多装机工单是一直查不到上网记录，不断的进入队列，导致新工单本来可以归档的工单卡在了竣工校验环节。
通过查询日志并定位，又重新拟定了一套方案，就是分组处理工单，即一部分是处理没校验通过的老工单，另一部分则是处理新流转进来的工单。这样就有效解决了新工单不至于在处理队列中等待太久。
代码略___
项目场景分析(二)
背景：同样是重庆移动代维综合管理系统，web端大致可以分为两部分业务系统和流程引擎，其中业务系统主要处理的是各类工单，各种人员，资源以及代维公司和分公司的管理；但往往完成一个业务场景是需要多人协助参与的，所以就引入了工作流这个概念，但往往工作流底层是比较复杂的；流程引擎恰恰就是基于工作流的思想进行了封装和改造，使得开发人员可以更加简洁的新建、修改、启动一个自定义的流程。而由于是是多人参与，那么就会涉及到人与人，人与系统之间的通信，基于人性化考虑，我们决定采用发短信的方式推送给工单待办人。
遇到的问题：由于代维系统本身不提供发短信的服务，因此必须借助第三方系统，即通过调用第三方接口的方式，将代维系统的业务信息推送给代办人；那么问题来了，既然是调接口，那么肯定涉及到时延的问题，这样就存在主线程必须等待接口返回结果后才能往下执行，而往往发送短信是一个比较耗时的过程。
解决方案: 因此我们想到采用多线程异步的方式调用短信平台接口，这样既不阻塞主线程执行，也给用户带来了良好的体验。代码如下：
package com.ultrapower.mams.smsnew.utils; import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException; import com.ultrapower.eoms.common.core.component.data.DataAdapter;
import com.ultrapower.eoms.common.core.component.data.DataRow;
import com.ultrapower.eoms.common.core.component.data.DataTable;
import com.ultrapower.eoms.common.core.util.TimeUtils;
import com.ultrapower.eoms.common.core.util.UUIDGenerator;
import com.ultrapower.mams.smsnew.constants.SmsNewConstants;
import com.ultrapower.mams.smsnew.thread.SmsSendOnceThread; public class SmsNewUtil { public static DataAdapter dataAdapter = new DataAdapter(); /**
*
* @param smsType 业务分类
* @param relateId 业务关联ID
* @param content 短信内容
* @param mobile 短信接收号码
* @param PRIORITY 优先级,1/2/3,优先级一次降低(对于时效性较高的短信才使用1级,一般短信使用2,无时效性要求的使用3)
* @param maxSendNum 最大发送次数
* @param preSendTime 定时发送时间
* @param remark 备注
* @return
*/
public static int insertSms(String smsType,String relateId,String content,
String mobile,Integer maxSendNum,Integer priority,Long preSendTime,String remark){ DataTable dt = new DataTable("BS_T_SM_SEND_SMS");
DataRow dr = new DataRow();
dr.put("PID", UUIDGenerator.getUUIDoffSpace());
dr.put("SMS_TYPE", smsType);
dr.put("RELATE_ID", relateId);
dr.put("CONTENT", content);
dr.put("MOBILE", mobile);
if(priority==null){
priority = 2;
}
dr.put("PRIORITY", priority);
//dr.put("SEND_FLAG", null);
if(maxSendNum==null){
maxSendNum = 1;
}
dr.put("MAX_SEND_NUM", maxSendNum);
dr.put("INPUT_TIME", TimeUtils.getCurrentTime());
dr.put("PRE_SEND_TIME", preSendTime);
//dr.put("SEND_TIME", null);
dr.put("REMATK", remark);
dt.putDataRow(dr); int re = dataAdapter.executeAdd(dt); return re;
} public static void sendAndRecord(String smsType,String relateId,String content,
String mobile,Integer maxSendNum,Integer priority,Long preSendTime,String remark){
DataRow dr = new DataRow();
dr.put("PID", UUIDGenerator.getUUIDoffSpace());
dr.put("SMS_TYPE", smsType);
dr.put("RELATE_ID", relateId);
dr.put("CONTENT", content);
dr.put("MOBILE", mobile);
if(priority==null){
priority = 2;
}
dr.put("PRIORITY", priority);
if(maxSendNum==null){
maxSendNum = 1;
}
if(preSendTime==null){
dr.put("PRE_SEND_TIME", "");
}else{
dr.put("PRE_SEND_TIME", preSendTime);
}
dr.put("MAX_SEND_NUM", maxSendNum);
dr.put("INPUT_TIME", TimeUtils.getCurrentTime());
dr.put("REMATK", remark); ExecutorService exec = Executors.newCachedThreadPool();
sendSMSByThread(exec,dr); }
PS:代码末尾使用了Executors类，Java通过Executors提供四种线程池，本例中使用了Executors创建了一个可以缓存线程池，应用中存在的线程数可以无限大，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
下面再看下sendSMSByThread(exec,dr)方法的实现细节：
public static void sendSMSByThread(ExecutorService exec,DataRow dr){
SmsSendOnceThread task = new SmsSendOnceThread(dr);
Future<Boolean> future = exec.submit(task);
Boolean taskResult = null;
String failReason = null;
try {
// 等待计算结果，最长等待300秒，300秒后中止任务
taskResult = future.get(300, TimeUnit.SECONDS);
} catch (InterruptedException e) {
failReason = "短信线程在等待结果时被中断！";
exec.shutdownNow();
} catch (ExecutionException e) {
failReason = "短信等待结果，但计算抛出异常！";
exec.shutdownNow();
} catch (TimeoutException e) {
failReason = "短信等待结果超时，因此中断任务线程！";
exec.shutdownNow();
}finally{
if(failReason!=null && !"".equals(failReason)){
dr.put("REMATK", failReason);
dr.put("SEND_TIME", TimeUtils.getCurrentTime());
dr.put("SEND_FLAG", SmsNewConstants.SEND_FLAG_SENDFAIL);
dr.put("SEND_NUM", 1);
DataTable dataTable = new DataTable("BS_T_SM_SEND_SMS_LOG");
dataTable.putDataRow(dr);
//dataAdapter.execute(dataTable);
dataAdapter.executeAdd(dataTable);
}
}
}
PS:可以看到是通过Futrue+Callable的方式来实现具有返回结果的线程啊，上一章节已经降到了。可以看到ExecutorService提交了一个Callable的实现类，我们接下来看下这个实现类(SmsSendOnceThread):
public class SmsSendOnceThread implements Callable<Boolean> {
private DataRow datarow;
public SmsSendOnceThread(){
}
public SmsSendOnceThread(DataRow datarow){
this.datarow = datarow;
}
@Override
public Boolean call() throws Exception {//跑一次就结束
String inserSql = "INSERT INTO BS_T_SM_SEND_SMS_LOG(PID, SMS_TYPE, RELATE_ID, CONTENT, MOBILE, PRIORITY, PRE_SEND_TIME, MAX_SEND_NUM, INPUT_TIME, REMATK, SEND_TIME, SEND_NUM, SEND_FLAG, SERIAL) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
Boolean result = Boolean.FALSE;
if(datarow!=null && datarow.length()>0){
DataAdapter dataAdapter = new DataAdapter();
//调用发送短信接口
JSONObject jsonParam = new JSONObject();
jsonParam.put("moblie", datarow.getString("MOBILE"));
jsonParam.put("message", datarow.getString("CONTENT"));
jsonParam.put("priority", datarow.getString("PRIORITY"));
JSONObject returnJson = SmsRestFullClient.postDataToInterface(SmsNewConstants.URL_SEND_SMS, jsonParam.toString(), null,"POST");
String outResult = returnJson.getString("outResult");
String outPut = returnJson.getString("outPut");
datarow.put("SEND_TIME", TimeUtils.getCurrentTime());
datarow.put("SEND_NUM", 1);
datarow.put("REMATK", returnJson);
if(SmsRestFullClient.OUTRESULT_SUCC.equals(outResult) && outPut!=null && outPut.contains("serial")){//成功后等待另一线程查询状态
JSONObject serial = JSONObject.fromObject(outPut);
String serialStr = serial.getString("serial");
datarow.put("SEND_FLAG", SmsNewConstants.SEND_FLAG_SEARCHWAIT);//等待查询
datarow.put("PRE_SEND_TIME", TimeUtils.getCurrentTime()+120);//两分钟后
datarow.put("SERIAL", serialStr);
}else{
datarow.put("SEND_FLAG", SmsNewConstants.SEND_FLAG_SENDFAIL);//发送失败
}
try {
dataAdapter.execute(inserSql, new Object[]{datarow.getString("PID"),datarow.getString("SMS_TYPE"),datarow.getString("RELATE_ID"),datarow.getString("CONTENT"),
datarow.getString("MOBILE"),datarow.getString("PRIORITY"),datarow.getString("PRE_SEND_TIME"),
datarow.getString("MAX_SEND_NUM"),datarow.getString("INPUT_TIME"),datarow.getString("REMATK"),
datarow.getString("SEND_TIME"),datarow.getString("SEND_NUM"),datarow.getString("SEND_FLAG"),datarow.getString("SERIAL")});
result = Boolean.TRUE;
} catch (Exception e) {
result = Boolean.FALSE;
}
}
return result;
}
public DataRow getDatarow() {
return datarow;
}
public void setDatarow(DataRow datarow) {
this.datarow = datarow;
}
}
我们可以看到这个实现类返回的是一个Boolean类型的变量，至于它具体是怎么调接口的这里暂不做阐述。
二、多线程的优缺点
当然，这里仅仅是与单线程作比较
|
|
单线程 |
多线程 |
|
成本 |
系统内存和IO占用较少 |
系统内存以及IO占用较多 |
|
性能 |
代码顺序执行，容易出现代码阻塞 |
线程间独立运行，能有效地避免代码阻塞，并且提高程序的运行性能 |
|
安全 |
很少出现线程安全的问题 |
如果控制不好会引来线程安全的问题 |
小结：多线程使程序的响应速度更快，因为用户界面可以在进行其他工作的同时一直处于活动状态。
是否需要创建多线程应用程序取决于多个因素。在以下情况下，最适合采用多线程处理：
- 耗时或大量占用处理器的任务阻塞用户界面操作
- 各个任务必须等待外部资源（如远程文件、发送短信或 Internet 连接)
(未完待续......)