基于Java線程池讀取數(shù)據(jù)庫中數(shù)據(jù)（學(xué)習(xí)+運(yùn)用）

以下是學(xué)習(xí)內(nèi)容

Main.java

import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit;public class Main {private static final int CORE_POOL_SIZE = 5;//核心線程數(shù)為5private static final int MAX_POOL_SIZE = 15;//最大線程數(shù)15private static final int QUEUE_CAPACITY = 5;//任務(wù)隊(duì)列容量為200private static final Long KEEP_ALIVE_TIME = 200L;//等待時間為1Lpublic static void main(String[] args) {//使用阿里巴巴推薦的創(chuàng)建線程池的方式//通過ThreadPoolExecutor構(gòu)造函數(shù)自定義參數(shù)創(chuàng)建ThreadPoolExecutor executor = new ThreadPoolExecutor(CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(QUEUE_CAPACITY),new ThreadPoolExecutor.CallerRunsPolicy()); //最后是拒絕策略long start = System.currentTimeMillis();System.out.println("activeCountMain1 : " + Thread.activeCount());for (int i = 1; i <= 10; i++) {MySQL mysql = new MySQL(i);executor.execute(mysql);System.out.println("線程池中線程數(shù)目：" + executor.getPoolSize() + "，隊(duì)列中等待執(zhí)行的任務(wù)數(shù)目：" + executor.getQueue().size() + "，已執(zhí)行玩別的任務(wù)數(shù)目：" + executor.getCompletedTaskCount());}executor.shutdown();while (true) {if (executor.getActiveCount() == 0)break;}System.out.println("activeCountMain2 : " + Thread.activeCount());long end = System.currentTimeMillis();System.out.println("平均每秒可輸出: " + 100000 / (end - start) + " 條");} }

MySQL.java

import java.sql.Connection; import java.sql.DriverManager; import java.sql.ResultSet; import java.sql.SQLException;import com.mysql.jdbc.Statement;/** * @author 作者 侯麗娟: * @version 創(chuàng)建時間：2021年10月7日 下午1:37:14 * 類說明 */class MySQL implements Runnable{private Connection con = null;private static String driver = "com.mysql.jdbc.Driver";private static String url = "jdbc:mysql://localhost:3306/loan";private static String username = "root";private static String password = "root";private static Statement NULL = null;private final int taskNum;public MySQL(int taskNum) {this.taskNum = taskNum;}public static void main(String[] args) { // MySQL mySQL = new MySQL(1); // Statement state = mySQL.MysqlOpen(); // System.out.println(state); // mySQL.readMySQL();}public Statement MysqlOpen() {try {Class.forName(driver); //加載驅(qū)動類con = DriverManager.getConnection(url, username, password); //連接數(shù)據(jù)庫if (!con.isClosed())System.out.println("***數(shù)據(jù)庫成功連接***");Statement state = (Statement) con.createStatement();return state;} catch (ClassNotFoundException e) {System.out.println("找不到驅(qū)動程序類，加載驅(qū)動失敗");e.printStackTrace();} catch (SQLException e) {System.out.println("數(shù)據(jù)庫連接失敗");e.printStackTrace();}return NULL;}@Overridepublic void run() {readMySQL();}public void readMySQL() {synchronized(MySQL.class){ResultSet sql = null;Statement state = MysqlOpen();try {sql = state.executeQuery("select * from jd_area_info where AreaID between "+ ((taskNum - 1) * 10) + " and " + (taskNum * 10));System.out.println("---------task " + taskNum + "正在執(zhí)行---------");while (sql.next()) {int id = sql.getInt(1);String areaname = sql.getString(3);String areaname2 = sql.getString(5);System.out.println(id + "\t" + areaname + "\t" + areaname2);}} catch (SQLException e) {e.printStackTrace();} finally {try {sql.close();state.close();con.close();} catch (Exception e) {e.printStackTrace();}}System.out.println("---------task " + taskNum + "執(zhí)行完畢---------");}} }

運(yùn)行結(jié)果如下圖所示：

【參考】https://www.jb51.net/article/168823.htm

以下是實(shí)際運(yùn)用

這是Controller中原本取數(shù)據(jù)的方法：

@RequestMapping(value = "/getSensorData", method = RequestMethod.GET)@ApiOperation(value = "通過sensorid獲取所有數(shù)據(jù)")public Result<datashow> getSensorData(HttpServletRequest request){long startTime = System.currentTimeMillis();String startDate = request.getParameter("startDate");String endDate = request.getParameter("endDate");String sid = request.getParameter("sid");System.out.println(sid+":開始時間加結(jié)束時間："+startDate+" "+endDate+"傳感器編號:"+sid);List<SensorWindPressure> list = iSensorWindPressureService.getSensorData(sid,startDate,endDate);log.info("數(shù)據(jù)量大小"+list.size());SensorInfo sensorInfo = iSensorWindPressureService.getDownUpLimit(sid);String uplimit = sensorInfo.getUpLimit();String downlimit = sensorInfo.getDownLimit();long endTime = System.currentTimeMillis();datashow datashow = new datashow();List<String> xdata = new LinkedList<>();List<String> ydata = new LinkedList<>();for (SensorWindPressure sensorWindPressure : list) {xdata.add(sdf.format(sensorWindPressure.getCreateTime()));ydata.add(sensorWindPressure.getValue());}datashow.setXdata(xdata);datashow.setYdata(ydata);System.out.println("代碼運(yùn)行時間：" + (endTime - startTime) + "ms");datashow.setUplimit(uplimit);datashow.setDownlimit(downlimit);return new ResultUtil<datashow>().setData(datashow); // System.out.println("開始時間加結(jié)束時間："+startDate+" "+endDate+"上限下限:"+uplimit+","+downlimit); // return new ResultUtil<List<SensorWindPressure>>().setData(list,uplimit,downlimit);}

以下是對以上方法的改進(jìn)，使用多線程進(jìn)行數(shù)據(jù)讀取：

private static final int CORE_POOL_SIZE = 5;//核心線程數(shù)為5private static final int MAX_POOL_SIZE = 15;//最大線程數(shù)15private static final int QUEUE_CAPACITY = 5;//任務(wù)隊(duì)列容量為200private static final Long KEEP_ALIVE_TIME = 10L;//等待時間為10s@RequestMapping(value = "/getSensorDataByline", method = RequestMethod.GET)@ApiOperation(value = "通過sensorid獲取所有數(shù)據(jù)")public Result<datashow> getSensorDataByline(HttpServletRequest request) throws Exception{SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd");long startTime = System.currentTimeMillis();String startDate = request.getParameter("startDate");String endDate = request.getParameter("endDate");String sid = request.getParameter("sid");System.out.println(sid+":開始時間加結(jié)束時間："+startDate+" "+endDate+"傳感器編號:"+sid);Date d1=sf.parse(startDate);Date d2=sf.parse(endDate);int len = daysBetween(d1,d2);System.out.println("相差天數(shù)："+len);//使用阿里巴巴推薦的創(chuàng)建線程池的方式//通過ThreadPoolExecutor構(gòu)造函數(shù)自定義參數(shù)創(chuàng)建ThreadPoolExecutor executor = new ThreadPoolExecutor(CORE_POOL_SIZE,MAX_POOL_SIZE,KEEP_ALIVE_TIME,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(QUEUE_CAPACITY),new ThreadPoolExecutor.CallerRunsPolicy()); //最后是拒絕策略ArrayList<Future<List<SensorWindPressure>>> results=new ArrayList<>();for (int i = 1; i <= len; i++) { // SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd");Calendar c = Calendar.getInstance();c.setTime(d1);c.add(c.DATE,i);String endDate2 = sf.format(c.getTime());System.out.println("endDate2："+endDate2+" "+i);c.setTime(d1);c.add(c.DATE,i-1);String startDate2 = sf.format(c.getTime());System.out.println("startDate2："+startDate2+" "+i);SmokeTask test = new SmokeTask(startDate2,endDate2,sid);Future<List<SensorWindPressure>> future = executor.submit(test);try{while (true) {if(future.isDone() && !future.isCancelled()){System.out.println(future);results.add(future);break;}else {Thread.sleep(1000);}}}catch (Exception e) {e.printStackTrace();}/*try {if(future.isDone()){System.out.println("以下是future：");System.out.println(future);System.out.println(future.get());}} catch (Exception e) {e.printStackTrace();}*/System.out.println("線程池中線程數(shù)目：" + executor.getPoolSize() + "，隊(duì)列中等待執(zhí)行的任務(wù)數(shù)目：" + executor.getQueue().size()+ "，已執(zhí)行玩別的任務(wù)數(shù)目：" + executor.getCompletedTaskCount());} // List<SensorWindPressure> list = iSensorWindPressureService.getSensorData(sid,startDate,endDate);List<SensorWindPressure> list = new ArrayList<>();for (Future<List<SensorWindPressure>> future: results) {try {System.out.println(future);System.out.println(future.get());list.addAll(future.get()); // System.out.println(future.get());} catch (Exception e) {e.printStackTrace();}finally{executor.shutdown();}}System.out.println(list.size());list.sort(Comparator.comparing(SensorWindPressure::getCreateTime)); // log.info("數(shù)據(jù)量大小"+list.size());SensorInfo sensorInfo = iSensorWindPressureService.getDownUpLimit(sid);String uplimit = sensorInfo.getUpLimit();String downlimit = sensorInfo.getDownLimit();long endTime = System.currentTimeMillis();datashow datashow = new datashow();List<String> xdata = new LinkedList<>();List<String> ydata = new LinkedList<>();SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");for (SensorWindPressure sensorWindPressure : list) {xdata.add(sdf.format(sensorWindPressure.getCreateTime()));ydata.add(sensorWindPressure.getValue());}datashow.setXdata(xdata);datashow.setYdata(ydata); // System.out.println("代碼運(yùn)行時間：" + (endTime - startTime) + "ms");datashow.setUplimit(uplimit);datashow.setDownlimit(downlimit);return new ResultUtil<datashow>().setData(datashow);/*List<SensorWindPressure> list=new ArrayList<>(); //原來寫的，僅供參考String startDate = request.getParameter("startDate");String endDate = request.getParameter("endDate");String sid = request.getParameter("sid");String daybetween = request.getParameter("daybetween");System.out.println(sid+":開始時間加結(jié)束時間："+startDate+" "+endDate+" "+daybetween);SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd");Date date = sf.parse(startDate);Calendar c = Calendar.getInstance();c.setTime(date);c.add(c.DATE,5);date=c.getTime();String startDate1 = sf.format(date);System.out.println(startDate1);/* BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(10);ThreadPoolExecutor pool = new ThreadPoolExecutor(3, 5, 60, TimeUnit.MICROSECONDS, queue);for(int i=0;i<pool.getPoolSize();i++){Callable task = new SmokeTask(startDate,endDate,sid);Future<List<SensorWindPressure>> f1 = pool.submit(task);System.out.println(f1.get());}pool.shutdown();*/SensorInfo sensorInfo = iSensorWindPressureService.getDownUpLimit(sid);String uplimit = sensorInfo.getUpLimit();String downlimit = sensorInfo.getDownLimit();System.out.println(sid+":開始時間加結(jié)束時間："+startDate+" "+endDate+"上限下限:"+uplimit+","+downlimit);return new ResultUtil<List<SensorWindPressure>>().setData(list,uplimit,downlimit);*/ // return null;}

取數(shù)據(jù)單獨(dú)提取出來了：

package cn.exrick.xboot.modules.your.bim.controller;import cn.exrick.xboot.modules.your.bim.entity.SensorWindPressure; import com.mysql.jdbc.Connection; import com.mysql.jdbc.Statement;import java.sql.DriverManager; import java.sql.ResultSet; import java.sql.SQLException; import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.concurrent.Callable;public class SmokeTask implements Callable<List<SensorWindPressure>> {private Connection con = null;private static String driver = "com.mysql.jdbc.Driver";private static String url = "jdbc:mysql://localhost:3306/shannan";private static String username = "root";private static String password = "root";private static Statement NULL = null;private String startDate;private String endDate;private String sid;public SmokeTask(String startDate,String endDate,String sid){this.startDate = startDate;this.endDate = endDate;this.sid = sid;}// @Autowired // private ISensorWindPressureService iSensorWindPressureService;public Statement MysqlOpen() {try {Class.forName(driver); //加載驅(qū)動類con = (Connection) DriverManager.getConnection(url, username, password); //連接數(shù)據(jù)庫if (!con.isClosed())System.out.println("***數(shù)據(jù)庫成功連接***");Statement state = (Statement) con.createStatement();return state;} catch (ClassNotFoundException e) {System.out.println("找不到驅(qū)動程序類，加載驅(qū)動失敗");e.printStackTrace();} catch (SQLException e) {System.out.println("數(shù)據(jù)庫連接失敗");e.printStackTrace();}return NULL;}@Overridepublic List<SensorWindPressure> call() throws Exception { // List<SensorWindPressure> list = iSensorWindPressureService.getSensorData(sid,startDate,endDate);synchronized (SmokeTask.class){List<SensorWindPressure> list = new ArrayList<>();ResultSet result = null;Statement state = MysqlOpen();try {String sql = "select create_time,value from sensor_wind_pressure where sensor_id = \""+sid+"\" and create_time between \""+startDate+"\" and \""+endDate+"\" order by create_time ASC";result = state.executeQuery(sql);while (result.next()) {Date date = result.getTimestamp(1);String value = result.getString(2); // System.out.println(date + "\t" + value);SensorWindPressure s=new SensorWindPressure();s.setCreateTime(date);s.setValue(value);list.add(s);}} catch (SQLException e) {e.printStackTrace();} finally {try {result.close();state.close();con.close();} catch (Exception e) {e.printStackTrace();}}System.out.println("---------task " + endDate + "執(zhí)行完畢---------");return list;}} }

前端展示如下：

不過這里的代碼還是有些亂，后面可重構(gòu)得簡潔一些。。。
過程中有遇到Java heap space問題，可參考解釋：https://www.cnblogs.com/bolang100/p/6478537.html

獲取返回結(jié)果優(yōu)化：
1、使用ExecutorService的invokeAll函數(shù)
本方法能解決第一個弊端，即并不需要自己去維護(hù)一個存儲返回結(jié)果的容器。當(dāng)我們需要獲取線程池所有的返回結(jié)果時，只需調(diào)用invokeAll函數(shù)即可。
但是，這種方式需要你自己去維護(hù)一個用于存儲任務(wù)的容器。

// 創(chuàng)建一個線程池 ExecutorService executorService = Executors.newFixedThreadPool(10);// 創(chuàng)建存儲任務(wù)的容器 List<Callable<String>> tasks = new ArrayList<Callable<String>>();// 提交10個任務(wù) for ( int i=0; i<10; i++ ) {Callable<String> task = new Callable<String>(){public String call(){int sleepTime = new Random().nextInt(1000);Thread.sleep(sleepTime);return "線程"+i+"睡了"+sleepTime+"秒";}};executorService.submit( task );// 將task添加進(jìn)任務(wù)隊(duì)列tasks.add( task ); }// 獲取10個任務(wù)的返回結(jié)果 List<Future<String>> results = executorService.invokeAll( tasks );// 輸出結(jié)果 for ( int i=0; i<10; i++ ) {// 獲取包含返回結(jié)果的future對象Future<String> future = results.get(i);// 從future中取出執(zhí)行結(jié)果（若尚未返回結(jié)果，則get方法被阻塞，直到結(jié)果被返回為止）String result = future.get();System.out.println(result); }

2、 使用CompletionService
CompletionService內(nèi)部維護(hù)了一個阻塞隊(duì)列，只有執(zhí)行完成的任務(wù)結(jié)果才會被放入該隊(duì)列，這樣就確保執(zhí)行時間較短的任務(wù)率先被存入阻塞隊(duì)列中。

ExecutorService exec = Executors.newFixedThreadPool(10);final BlockingQueue<Future<Integer>> queue = new LinkedBlockingDeque<Future<Integer>>( 10); //實(shí)例化CompletionService final CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>( exec, queue); // 提交10個任務(wù) for ( int i=0; i<10; i++ ) {executorService.submit( new Callable<String>(){public String call(){int sleepTime = new Random().nextInt(1000);Thread.sleep(sleepTime);return "線程"+i+"睡了"+sleepTime+"秒";}} ); }// 輸出結(jié)果 for ( int i=0; i<10; i++ ) {// 獲取包含返回結(jié)果的future對象（若整個阻塞隊(duì)列中還沒有一條線程返回結(jié)果，那么調(diào)用take將會被阻塞，當(dāng)然你可以調(diào)用poll，不會被阻塞，若沒有結(jié)果會返回null，poll和take返回正確的結(jié)果后會將該結(jié)果從隊(duì)列中刪除）Future<String> future = completionService.take();// 從future中取出執(zhí)行結(jié)果，這里存儲的future已經(jīng)擁有執(zhí)行結(jié)果，get不會被阻塞String result = future.get();System.out.println(result); }

【參考】https://blog.csdn.net/lijingjingchn/article/details/104491862

總結(jié)

以上是生活随笔為你收集整理的基于Java线程池读取数据库中数据（学习+运用）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。