基于Java线程池读取数据库中数据（学习+运用）

小鱼儿 2022-09-14 12:28 165阅读 0赞

## 基于Java线程池读取数据库中数据（学习+运用） ##

## 以下是学习内容 ##

**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;//核心线程数为5
        private static final int MAX_POOL_SIZE = 15;//最大线程数15
        private static final int QUEUE_CAPACITY = 5;//任务队列容量为200
        private static final Long KEEP_ALIVE_TIME = 200L;//等待时间为1L
        
     public static void main(String[] args) { 
    
    	     //使用阿里巴巴推荐的创建线程池的方式
    	     //通过ThreadPoolExecutor构造函数自定义参数创建
    		 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("线程池中线程数目：" + executor.getPoolSize() + "，队列中等待执行的任务数目：" + executor.getQueue().size() 
    					  + "，已执行玩别的任务数目：" + 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 创建时间：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); //加载驱动类
    			  con = DriverManager.getConnection(url, username, password); //连接数据库
    			  if (!con.isClosed())
    				  System.out.println("***数据库成功连接***");
    			  Statement state = (Statement) con.createStatement();
    			  return state;
    		 } catch (ClassNotFoundException e) { 
    			 System.out.println("找不到驱动程序类，加载驱动失败");
    			 e.printStackTrace();
    		 } catch (SQLException e) { 
    			 System.out.println("数据库连接失败");
    			 e.printStackTrace();
    		 }
    		 return NULL;
    	 }
    
    	@Override
    	public 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 + "正在执行---------");
    			  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 + "执行完毕---------");
    		}
    	 }
    }

运行结果如下图所示：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAbW9ua2V5aGxq_size_20_color_FFFFFF_t_70_g_se_x_16]  
【参考】[https://www.jb51.net/article/168823.htm][https_www.jb51.net_article_168823.htm]

## 以下是实际运用 ##

这是Controller中原本取数据的方法：

@RequestMapping(value = "/getSensorData", method = RequestMethod.GET)
        @ApiOperation(value = "通过sensorid获取所有数据")
        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+":开始时间加结束时间："+startDate+" "+endDate+"传感器编号:"+sid);
            List<SensorWindPressure> list = iSensorWindPressureService.getSensorData(sid,startDate,endDate);
            log.info("数据量大小"+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("代码运行时间：" + (endTime - startTime) + "ms");
            datashow.setUplimit(uplimit);
            datashow.setDownlimit(downlimit);
            return new ResultUtil<datashow>().setData(datashow);
    // System.out.println("开始时间加结束时间："+startDate+" "+endDate+"上限下限:"+uplimit+","+downlimit);
    // return new ResultUtil<List<SensorWindPressure>>().setData(list,uplimit,downlimit);
        }

**以下是对以上方法的改进，使用多线程进行数据读取：**

private static final int CORE_POOL_SIZE = 5;//核心线程数为5
        private static final int MAX_POOL_SIZE = 15;//最大线程数15
        private static final int QUEUE_CAPACITY = 5;//任务队列容量为200
        private static final Long KEEP_ALIVE_TIME = 10L;//等待时间为10s
    
        @RequestMapping(value = "/getSensorDataByline", method = RequestMethod.GET)
        @ApiOperation(value = "通过sensorid获取所有数据")
        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+":开始时间加结束时间："+startDate+" "+endDate+"传感器编号:"+sid);
            Date d1=sf.parse(startDate);
            Date d2=sf.parse(endDate);
            int len = daysBetween(d1,d2);
            System.out.println("相差天数："+len);
    
            //使用阿里巴巴推荐的创建线程池的方式
            //通过ThreadPoolExecutor构造函数自定义参数创建
            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("线程池中线程数目：" + executor.getPoolSize() + "，队列中等待执行的任务数目：" + executor.getQueue().size()
                        + "，已执行玩别的任务数目：" + 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("数据量大小"+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("代码运行时间：" + (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+":开始时间加结束时间："+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+":开始时间加结束时间："+startDate+" "+endDate+"上限下限:"+uplimit+","+downlimit);
            return new ResultUtil<List<SensorWindPressure>>().setData(list,uplimit,downlimit);*/
    // return null;
        }

**取数据单独提取出来了：**

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); //加载驱动类
                con = (Connection) DriverManager.getConnection(url, username, password); //连接数据库
                if (!con.isClosed())
                    System.out.println("***数据库成功连接***");
                Statement state = (Statement) con.createStatement();
                return state;
            } catch (ClassNotFoundException e) { 
                System.out.println("找不到驱动程序类，加载驱动失败");
                e.printStackTrace();
            } catch (SQLException e) { 
                System.out.println("数据库连接失败");
                e.printStackTrace();
            }
            return NULL;
        }
    
        @Override
        public 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 + "执行完毕---------");
                return list;
            }
        }
    }

前端展示如下：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAbW9ua2V5aGxq_size_20_color_FFFFFF_t_70_g_se_x_16 1]  
不过这里的代码还是有些乱，后面可重构得简洁一些。。。  
过程中有遇到`Java heap space`问题，可参考解释：[https://www.cnblogs.com/bolang100/p/6478537.html][https_www.cnblogs.com_bolang100_p_6478537.html]

**获取返回结果优化：**  
1、使用`ExecutorService`的`invokeAll`函数  
本方法能解决第一个弊端，即并不需要自己去维护一个存储返回结果的容器。当我们需要获取线程池所有的返回结果时，只需调用invokeAll函数即可。  
但是，这种方式需要你自己去维护一个用于存储任务的容器。

// 创建一个线程池
    ExecutorService executorService = Executors.newFixedThreadPool(10);
    
    // 创建存储任务的容器
    List<Callable<String>> tasks = new ArrayList<Callable<String>>();
    
    // 提交10个任务
    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添加进任务队列
        tasks.add( task );
    }
    
    // 获取10个任务的返回结果
    List<Future<String>> results = executorService.invokeAll( tasks );
    
    // 输出结果
    for ( int i=0; i<10; i++ ) { 
        // 获取包含返回结果的future对象
        Future<String> future = results.get(i);
        // 从future中取出执行结果（若尚未返回结果，则get方法被阻塞，直到结果被返回为止）
        String result = future.get();
        System.out.println(result);
    }

2、 使用`CompletionService`  
CompletionService内部维护了一个阻塞队列，只有执行完成的任务结果才会被放入该队列，这样就确保执行时间较短的任务率先被存入阻塞队列中。

ExecutorService exec = Executors.newFixedThreadPool(10);
    
    final BlockingQueue<Future<Integer>> queue = new LinkedBlockingDeque<Future<Integer>>(  
                    10);  
            //实例化CompletionService 
            final CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(  
                    exec, queue); 
    
    // 提交10个任务
    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+"秒";
            }
        } );
    }
    
    // 输出结果
    for ( int i=0; i<10; i++ ) { 
        // 获取包含返回结果的future对象（若整个阻塞队列中还没有一条线程返回结果，那么调用take将会被阻塞，当然你可以调用poll，不会被阻塞，若没有结果会返回null，poll和take返回正确的结果后会将该结果从队列中删除）
        Future<String> future = completionService.take();
        // 从future中取出执行结果，这里存储的future已经拥有执行结果，get不会被阻塞
        String result = future.get();
        System.out.println(result);
    }

【参考】[https://blog.csdn.net/lijingjingchn/article/details/104491862][https_blog.csdn.net_lijingjingchn_article_details_104491862]

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAbW9ua2V5aGxq_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/7155d57557d945a09193ce0f76923c94.png
[https_www.jb51.net_article_168823.htm]: https://www.jb51.net/article/168823.htm
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAbW9ua2V5aGxq_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220828/f8bf46e5d6d74f59b38d54e7a3bfc9ba.png
[https_www.cnblogs.com_bolang100_p_6478537.html]: https://www.cnblogs.com/bolang100/p/6478537.html
[https_blog.csdn.net_lijingjingchn_article_details_104491862]: https://blog.csdn.net/lijingjingchn/article/details/104491862