For multithreading, the beginning is more confused, do not understand the operation mechanism. In the recent project, we need to solve a problem of slow loading with multi-threading. We hereby write an example for your reference. If you have any suggestions, please give us your advice, hahaha
So, without further ado. First, let’s talk about the requirements: this interface is called by XXX company to achieve the reading and modification of the database. And is full read, not paging read. (Millions of pieces of data) You have to think about if you batch the data, where the running time is relatively long, and what parts of the code are going to be multithreaded.
Note: the most important thing for programmers is not to write code, but to think about the problem and come up with the best solution, and then write code to solve it.
In this case, the full data is processed in batches, and 5000 pieces of data are read in each batch. Code first:
/**
* 查询指定SN的IP值
*
* @return 结果
*/
@RequestMapping("/findIP")
@ResponseBody
public Map<String, Object> getIpFromSn(int flag) {
Map<String, Object> map = new HashMap<String, Object>();
// JSONObject json = null;
try {
//分批进行处理
int total = this.routeInfoService.getRouteFindIPCount();
int onet = 5000;
int begt = 0;
int Wtime = total / onet + 1;
int totalCount = 1;
if (flag == 1) {
KEY_CHECK_DEAL_TYPE = "stop";
} else {
KEY_CHECK_DEAL_TYPE = "deal";
}
for (int j = 1; j <= Wtime; j++) {
logger.info("当前的处理标识为{}"+KEY_CHECK_DEAL_TYPE);
if(KEY_CHECK_DEAL_TYPE.equals("deal")){
begt = (j - 1) * onet;
List<RouteCheckIp> infos = new ArrayList<RouteCheckIp>();
// SimpleDateFormat begint = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
// Date beginti = begint.parse(begint.format(new Date()));
//查询需要新增处理的SN号码
List<RouteInfo> SnList = this.routeInfoService.selectRouteInfoSn(begt, onet);
// 增加线程，处理循环中的每5000条数据
int threadNum = THREAD_NUM;
ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
CountDownLatch countDownLatch = new CountDownLatch(threadNum);
int perSize = SnList.size() / threadNum;
for (int i = 0; i < threadNum; i++) {
MultiThread thread = new MultiThread();
thread.setRouteInfoList(SnList.subList(i * perSize, (i + 1) * perSize));
thread.setRedisReportUtil(redisReportUtil);
thread.setTotalCount(totalCount);
thread.setTotal(total);
thread.setRouteInfoService(routeInfoService);
thread.setJ(j);
thread.setCountDownLatch(countDownLatch);
executorService.submit(thread);
}
countDownLatch.await();
executorService.shutdown();
// 优化前
// for (RouteInfo info : SnList) {
// String dateBySn = redisReportUtil.getString(CommonConstant.ROUTE_INFO_REPORT_REDIS_PRE + info.getRouteSn());
// if (!"".equals(dateBySn) && (dateBySn != null)) {
// RouteCheckIp rci = new RouteCheckIp();
// json = new JSONObject(dateBySn);
// SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
// String str = json.isNull("routerIp") ? "" : json.get("routerIp").toString();
// int inter = 0;
// if(str==null || str.length()<2){
// inter=3;
// }else if (checkIp("192",str) || checkIp("10",str) || checkIp("100",str) || checkIp("172",str)) {
// inter = 0;
// }else{
// inter=1;
// }
// /**
// * 2018/6/27 新增redis中除去deviceList字段的所有字段
// * 开始
// */
// rci.setWanUpSpeed(json.isNull("wanUpSpeed") ? "" : json.get("wanUpSpeed").toString());
// rci.setReport_time_local(json.isNull("report_time_local") ? "" : json.get("report_time_local").toString());
// rci.setWifiUsernum(json.isNull("wifiUsernum") ? "" : json.get("wifiUsernum").toString());
// rci.setSystemUptime(json.isNull("systemUptime") ? "" : json.get("systemUptime").toString());
// rci.setLan1(json.isNull("lan1") ? "" : json.get("lan1").toString());
// rci.setLan2(json.isNull("lan2") ? "" : json.get("lan2").toString());
// rci.setLan3(json.isNull("lan3") ? "" : json.get("lan3").toString());
// rci.setLan4(json.isNull("lan4") ? "" : json.get("lan4").toString());
// rci.setMac(json.isNull("mac") ? "" : json.get("mac").toString());
// rci.setFreeRam(json.isNull("freeRam") ? "" : json.get("freeRam").toString());
// rci.setPriDns(json.isNull("priDns") ? "" : json.get("priDns").toString());
// rci.setGatewayMask(json.isNull("gatewayMask") ? "" : json.get("gatewayMask").toString());
// rci.setWanDownSpeed(json.isNull("wanDownSpeed") ? "" : json.get("wanDownSpeed").toString());
// rci.setUserNumber(json.isNull("userNumber") ? "" : json.get("userNumber").toString());
// rci.setSecDns(json.isNull("secDns") ? "" : json.get("secDns").toString());
// rci.setLinkNumber(json.isNull("linkNumber") ? "" : json.get("linkNumber").toString());
// rci.setWan(json.isNull("wan") ? "" : json.get("wan").toString());
// rci.setCpu(json.isNull("cpu") ? "" : json.get("cpu").toString());
// rci.setOnlineUsernum(json.isNull("onlineUsernum") ? "" : json.get("onlineUsernum").toString());
// rci.setUseRam(json.isNull("useRam") ? "" : json.get("useRam").toString());
// /**
// * 2018/6/27 新增redis中除去deviceList字段的所有字段
// * 结束
// */
// rci.setRouteIp(json.isNull("routerIp") ? "" : json.get("routerIp").toString());
// rci.setInternet(inter);
// rci.setRouteSn(info.getRouteSn());
// rci.setVersion(json.isNull("version") ? "" : json.get("version").toString());
// rci.setGatewayIP(json.isNull("gatewayIP") ? "" : json.get("gatewayIP").toString());
// rci.setAdmuser(json.isNull("admuser") ? "" : json.get("admuser").toString());
// rci.setAdmpass(json.isNull("admpass") ? "" : json.get("admpass").toString());
// rci.setPppoeuser(json.isNull("pppoeuser") ? "" : json.get("pppoeuser").toString());
// rci.setPPPoepass(json.isNull("PPPoepass") ? "" : json.get("PPPoepass").toString());
// rci.setGetIFtime(sdf.format(System.currentTimeMillis()));
// infos.add(rci);
//
// } else {
// RouteCheckIp rci = new RouteCheckIp();
// SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
// rci.setRouteSn(info.getRouteSn().toString());
// rci.setInternet(2);
// rci.setGetIFtime(sdf1.format(System.currentTimeMillis()));
// infos.add(rci);
// }
// totalCount++;
// }
// this.routeInfoService.addRouteIpDate(infos);
//
// SimpleDateFormat endt = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
// Date endti = endt.parse(endt.format(new Date()));
// logger.info("总计数：{}，第{}个5000批处理结束，花费时间为：{}毫秒", totalCount, j, endti.getTime() - beginti.getTime());
}
else{
System.out.println("获取到停止指令！即将停止！");
break;
}
}
map.put("result", "success");
map.put("msg", "操作成功！");
return map;
} catch (Exception e) {
logger.error("数据库操作失败！", e);
map.put("result", "failure");
map.put("msg", "数据库操作失败!");
return map;
}
}
Copy the code1, check the amount of data in the corresponding table in the database, very large, very slow processing, and is nested for loop, running speed needless to say, will certainly be much slower. 2. In this example, the amount of data queried is 3 million, and the data is processed in 600 batches with 5000 pieces in each batch. I am creating 20 threads for processing. Readers can modify the number of threads according to their needs to see if they can be more efficient. 3. Multithreading code
Int threadNum = THREAD_NUM; ExecutorService executorService = Executors.newFixedThreadPool(threadNum); CountDownLatch countDownLatch = new CountDownLatch(threadNum); int perSize = SnList.size() / threadNum; for (int i = 0; i < threadNum; i++) { MultiThread thread = new MultiThread(); thread.setRouteInfoList(SnList.subList(i * perSize, (i + 1) * perSize)); thread.setRedisReportUtil(redisReportUtil); thread.setTotalCount(totalCount); thread.setTotal(total); thread.setRouteInfoService(routeInfoService); thread.setJ(j); thread.setCountDownLatch(countDownLatch); executorService.submit(thread); } countDownLatch.await(); executorService.shutdown();Copy the code4, Write the corresponding Tread class
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.json.JSONObject;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.CountDownLatch;
/**
* @ClassName MultiThread
* @Description 线程类：缩短处理routeInfo的时间
* @Date 2019/7/12 17:32
**/
public class MultiThread extends Thread {
private Logger logger = LoggerFactory.getLogger(getClass());
private RedisReportUtil redisReportUtil;
private RouteInfoService routeInfoService;
private int totalCount;
// 第一层for循环执行次数
int j;
private static final int TOTAL_ACOUNT = 5000;
private List<RouteInfo> routeInfoList;
private CountDownLatch countDownLatch;
// 表数据总数
private int total;
public void setJ(int j) {
this.j = j;
}
public void setRedisReportUtil(RedisReportUtil redisReportUtil) {
this.redisReportUtil = redisReportUtil;
}
public void setRouteInfoList(List<RouteInfo> routeInfoList) {
this.routeInfoList = routeInfoList;
}
public void setCountDownLatch(CountDownLatch countDownLatch) {
this.countDownLatch = countDownLatch;
}
public void setRouteInfoService(RouteInfoService routeInfoService) {
this.routeInfoService = routeInfoService;
}
public void setTotalCount(int totalCount) {
this.totalCount = totalCount;
}
public void setTotal(int total) {
this.total = total;
}
@Override
public void run() {
try {
Thread.sleep(1000);
forRouteInfoList();
} catch (Exception e) {
e.printStackTrace();
} finally {
if (countDownLatch != null) {
countDownLatch.countDown();
}
}
}
/**
* @Description: for循环查询routeInfo并处理
* @return:
* @Date: 2019/7/12
*/
public void forRouteInfoList(){
try{
Map<String, Object> map = new HashMap<String, Object>();
JSONObject json = null;
SimpleDateFormat begint = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date beginti = begint.parse(begint.format(new Date()));
int totalCount = 0;
List<RouteInfo> SnList = this.routeInfoList;
List<RouteCheckIp> infos = new ArrayList<RouteCheckIp>();
for (RouteInfo info : SnList) {
String dateBySn = this.redisReportUtil.getString(CommonConstant.ROUTE_INFO_REPORT_REDIS_PRE + info.getRouteSn());
if (!"".equals(dateBySn) && (dateBySn != null)) {
RouteCheckIp rci = new RouteCheckIp();
json = new JSONObject(dateBySn);
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String str = json.isNull("routerIp") ? "" : json.get("routerIp").toString();
int inter = 0;
if (str == null || str.length() < 2){
inter = 3;
} else if (checkIp("192",str) || checkIp("10",str) || checkIp("100",str) || checkIp("172",str)) {
inter = 0;
} else {
inter = 1;
}
// 2018/6/27 新增redis中除去deviceList字段的所有字段 --- 开始
rci.setWanUpSpeed(json.isNull("wanUpSpeed") ? "" : json.get("wanUpSpeed").toString());
rci.setReport_time_local(json.isNull("report_time_local") ? "" : json.get("report_time_local").toString());
rci.setWifiUsernum(json.isNull("wifiUsernum") ? "" : json.get("wifiUsernum").toString());
rci.setSystemUptime(json.isNull("systemUptime") ? "" : json.get("systemUptime").toString());
rci.setLan1(json.isNull("lan1") ? "" : json.get("lan1").toString());
rci.setLan2(json.isNull("lan2") ? "" : json.get("lan2").toString());
rci.setLan3(json.isNull("lan3") ? "" : json.get("lan3").toString());
rci.setLan4(json.isNull("lan4") ? "" : json.get("lan4").toString());
rci.setMac(json.isNull("mac") ? "" : json.get("mac").toString());
rci.setFreeRam(json.isNull("freeRam") ? "" : json.get("freeRam").toString());
rci.setPriDns(json.isNull("priDns") ? "" : json.get("priDns").toString());
rci.setGatewayMask(json.isNull("gatewayMask") ? "" : json.get("gatewayMask").toString());
rci.setWanDownSpeed(json.isNull("wanDownSpeed") ? "" : json.get("wanDownSpeed").toString());
rci.setUserNumber(json.isNull("userNumber") ? "" : json.get("userNumber").toString());
rci.setSecDns(json.isNull("secDns") ? "" : json.get("secDns").toString());
rci.setLinkNumber(json.isNull("linkNumber") ? "" : json.get("linkNumber").toString());
rci.setWan(json.isNull("wan") ? "" : json.get("wan").toString());
rci.setCpu(json.isNull("cpu") ? "" : json.get("cpu").toString());
rci.setOnlineUsernum(json.isNull("onlineUsernum") ? "" : json.get("onlineUsernum").toString());
rci.setUseRam(json.isNull("useRam") ? "" : json.get("useRam").toString());
// 2018/6/27 新增redis中除去deviceList字段的所有字段 --- 结束
rci.setRouteIp(json.isNull("routerIp") ? "" : json.get("routerIp").toString());
rci.setInternet(inter);
rci.setRouteSn(info.getRouteSn());
rci.setVersion(json.isNull("version") ? "" : json.get("version").toString());
rci.setGatewayIP(json.isNull("gatewayIP") ? "" : json.get("gatewayIP").toString());
rci.setAdmuser(json.isNull("admuser") ? "" : json.get("admuser").toString());
rci.setAdmpass(json.isNull("admpass") ? "" : json.get("admpass").toString());
rci.setPppoeuser(json.isNull("pppoeuser") ? "" : json.get("pppoeuser").toString());
rci.setPPPoepass(json.isNull("PPPoepass") ? "" : json.get("PPPoepass").toString());
rci.setGetIFtime(sdf.format(System.currentTimeMillis()));
infos.add(rci);
} else {
RouteCheckIp rci = new RouteCheckIp();
SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
rci.setRouteSn(info.getRouteSn().toString());
rci.setInternet(2);
rci.setGetIFtime(sdf1.format(System.currentTimeMillis()));
infos.add(rci);
}
this.totalCount++;
}
this.routeInfoService.addRouteIpDate(infos);
SimpleDateFormat endt = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date endti = endt.parse(endt.format(new Date()));
if (TOTAL_ACOUNT * j < this.total) {
if (j == 1){
total = TOTAL_ACOUNT;
} else {
total = TOTAL_ACOUNT * j;
}
} else {
total = SnList.size();
}
logger.info("总计数：{}，第{}个5000批处理结束，花费时间为：{}毫秒" ,total, this.j, endti.getTime() - beginti.getTime());
} catch (Exception e) {
logger.error("数据库操作失败！", e);
}
}
/**
* 校验IP
* @param prefix ip前缀
* @param ipAddr 完整ip地址
* @return
*/
public boolean checkIp(String prefix, String ipAddr){
StringBuilder reg = new StringBuilder(prefix);
reg.append("(\\.(\\d|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])){3}");
return ipAddr.matches(reg.toString());
}
}
Copy the code5, when the creation thread is finished, execute the run() method in the thread, then we need to process the logic code in the run() method.
@Override public void run() { try { Thread.sleep(1000); forRouteInfoList(); } catch (Exception e) {e.printStackTrace();} catch (Exception e) {e.printstackTrace (); } finally { if (countDownLatch ! = null) { countDownLatch.countDown(); }}Copy the code6. In this example, the call interface was reduced from 6 hours to 50 minutes, processing 3 million data and logic.
It can be seen that the high efficiency of multi-threading, of course, for the large amount of data level should consider the use of multi-threading processing, save the running time of the code, to give users the best experience.
Multithreading advantages:
Multithreading, as a multitasking, concurrent way of working, certainly has its advantages: ① Processes cannot share memory before, whereas sharing memory between threads (heap memory) is very simple. ② The system needs to re-allocate system resources for the process when creating the process, the cost of creating threads is much lower, so the realization of multi-task concurrency, multi-threading efficiency is higher. ③ The Java language itself built-in multi-threading function support, rather than simply as the first scheduling mode of the underlying system, thus simplifying multi-threading programming. CS game/single-lane and multi-lane multithreaded download: can be understood as a thread is a file download channel, multithreading is open several download channels at the same time. When the server provides download services, users of download are sharing bandwidth, in the case of the same priority, the master server will distribute the total download threads equally. Understandably, if you have more threads, the faster the download will be. Now popular download software all support multithreading. Multithreading is to complete multiple tasks synchronously, not to provide program efficiency, but to improve the efficiency of the system by improving the efficiency of resource use.
Hope through this example, to solve the current multithreading how to deal with the program ape people a train of thought.
I am the advanced ball, let’s climb the pit in 2019. Feel to share very force words to praise, thank you!! If you have any questions, please leave a message below.