logo
搜索
我的图书馆

发文章

发文工具

其他工具

留言交流
搜索
分享

JAVA中用多线程技术实现大数据导入

 aaie_ 2017-01-03
朋友让我帮忙写个程序从文本文档中导入数据到oracle数据库中,技术上没有什么难度,文档的格式都是固定的只要对应数据库中的字段解析就行了,关键在于性能。 
 
数据量很大百万条记录,因此考虑到要用多线程并发执行,在写的过程中又遇到问题,我想统计所有子进程执行完毕总共的耗时,在第一个子进程创建前记录当前时间用System.currentTimeMillis()在最后一个子进程结束后记录当前时间,两次一减得到的时间差即为总共的用时,代码如下

Java代码
long tStart = System.currentTimeMillis();
  1. System.out.println(Thread.currentThread().getName() + "开始");//打印开始标记
  2. for (int ii = 0; ii < threadNum; ii++) {//开threadNum个线程
  3. Runnable r = new Runnable(){
  4. @Override
  5. public void run(){
  6. System.out.println(Thread.currentThread().getName() + "开始");
  7. //做一些事情... ...
  8. System.out.println(Thread.currentThread().getName() + "结束.");
  9. }
  10. }
  11. Thread t = new Thread(r);
  12. t.start();
  13. }
  14. System.out.println(Thread.currentThread().getName() + "结束.");//打印结束标记
  15. long tEnd = System.currentTimeMillis();
  16. System.out.println("总共用时:"+ (tEnd - tStart) + "millions");
  1. long tStart = System.currentTimeMillis();
  2. System.out.println(Thread.currentThread().getName() + "开始");//打印开始标记
  3. for (int ii = 0; ii < threadNum; ii++) {//开threadNum个线程
  4. Runnable r = new Runnable(){
  5. @Override
  6. public void run(){
  7. System.out.println(Thread.currentThread().getName() + "开始");
  8. //做一些事情... ...
  9. System.out.println(Thread.currentThread().getName() + "结束.");
  10. }
  11. }
  12. Thread t = new Thread(r);
  13. t.start();
  14. }
  15. System.out.println(Thread.currentThread().getName() + "结束.");//打印结束标记
  16. long tEnd = System.currentTimeMillis();
  17. System.out.println("总共用时:"+ (tEnd - tStart) + "millions");

结果是几乎在for循环结束的瞬间就执行了主线程打印总共用时的语句,原因是所有的子线程是并发执行的,它们运行时主线程也在运行,这就引出了一个问题即本文标题如何"让主线程等待所有子线程执行完毕"。试过在每个子线程开始后加上t.join(),结果是所有线程都顺序执行,这就失去了并发的意义了,显然不是我想要的。
网上Google了很久也没有找到解决方案,难道就没有人遇到过这种需求吗?还是这个问题太简单了?无耐只得自己想办法了...
最后我的解决办法是,自定义一个ImportThread类继承自java.lang.Thread,重载run()方法,用一个List属性保存所有产生的线程,这样只要判断这个List是否为空就知道还有没有子线程没有执行完了,类代码如下：

Java代码
public class ImportThread extends Thread {
  1. private static List<Thread> runningThreads = new ArrayList<Thread>();
  2. public ImportThread() {
  3. }
  4. @Override
  5. public void run() {
  6. regist(this);//线程开始时注册
  7. System.out.println(Thread.currentThread().getName() + "开始...");//打印开始标记
  8. //做一些事情... ...
  9. unRegist(this);//线程结束时取消注册
  10. System.out.println(Thread.currentThread().getName() + "结束.");//打印结束标记
  11. }
  12. public void regist(Thread t){
  13. synchronized(runningThreads){
  14. runningThreads.add(t);
  15. }
  16. }
  17. public void unRegist(Thread t){
  18. synchronized(runningThreads){
  19. runningThreads.remove(t);
  20. }
  21. }
  22. public static boolean hasThreadRunning() {
  23. return (runningThreads.size() > 0);//通过判断runningThreads是否为空就能知道是否还有线程未执行完
  24. }
  25. }
  1. public class ImportThread extends Thread {
  2. private static List<Thread> runningThreads = new ArrayList<Thread>();
  3. public ImportThread() {
  4. }
  5. @Override
  6. public void run() {
  7. regist(this);//线程开始时注册
  8. System.out.println(Thread.currentThread().getName() + "开始...");//打印开始标记
  9. //做一些事情... ...
  10. unRegist(this);//线程结束时取消注册
  11. System.out.println(Thread.currentThread().getName() + "结束.");//打印结束标记
  12. }
  13. public void regist(Thread t){
  14. synchronized(runningThreads){
  15. runningThreads.add(t);
  16. }
  17. }
  18. public void unRegist(Thread t){
  19. synchronized(runningThreads){
  20. runningThreads.remove(t);
  21. }
  22. }
  23. public static boolean hasThreadRunning() {
  24. return (runningThreads.size() > 0);//通过判断runningThreads是否为空就能知道是否还有线程未执行完
  25. }
  26. }

主线程中代码：
Java代码 
long tStart = System.currentTimeMillis();
  1. System.out.println(Thread.currentThread().getName() + "开始");//打印开始标记
  2. for (int ii = 0; ii < threadNum; ii++) {//开threadNum个线程
  3. Thread t = new ImportThread();
  4. t.start();
  5. }
  6. while(true){//等待所有子线程执行完
  7. if(!ImportThread.hasThreadRunning()){
  8. break;
  9. }
  10. Thread.sleep(500);
  11. }
  12. System.out.println(Thread.currentThread().getName() + "结束.");//打印结束标记
  13. long tEnd = System.currentTimeMillis();
  14. System.out.println("总共用时:"+ (tEnd - tStart) + "millions");
  1. long tStart = System.currentTimeMillis();
  2. System.out.println(Thread.currentThread().getName() + "开始");//打印开始标记
  3. for (int ii = 0; ii < threadNum; ii++) {//开threadNum个线程
  4. Thread t = new ImportThread();
  5. t.start();
  6. }
  7. while(true){//等待所有子线程执行完
  8. if(!ImportThread.hasThreadRunning()){
  9. break;
  10. }
  11. Thread.sleep(500);
  12. }
  13. System.out.println(Thread.currentThread().getName() + "结束.");//打印结束标记
  14. long tEnd = System.currentTimeMillis();
  15. System.out.println("总共用时:"+ (tEnd - tStart) + "millions");

打印的结果是：
main开始
Thread-1开始...
Thread-5开始...
Thread-0开始...
Thread-2开始...
Thread-3开始...
Thread-4开始...
Thread-5结束.
Thread-4结束.
Thread-2结束.
Thread-0结束.
Thread-3结束.
Thread-1结束.
main结束.
总共用时:20860millions
可以看到main线程是等所有子线程全部执行完后才开始执行的。
==================================================以下为第二次编辑===============================================
上面的方法有一个隐患:如果线程1开始并且结束了,而其他线程还没有开始此时runningThreads的size也为0,主线程会以为所有线程都执行完了。解决办法是用一个非简单类型的计数器来取代List型的runningThreads,并且在线程创建之前就应该设定好计数器的值。
MyCountDown类
Java代码 
public class MyCountDown {
  1. private int count;
  2. public MyCountDown(int count){
  3. this.count = count;
  4. }
  5. public synchronized void countDown(){
  6. count--;
  7. }
  8. public synchronized boolean hasNext(){
  9. return (count > 0);
  10. }
  11. public int getCount() {
  12. return count;
  13. }
  14. public void setCount(int count) {
  15. this.count = count;
  16. }
  17. }
  1. public class MyCountDown {
  2. private int count;
  3. public MyCountDown(int count){
  4. this.count = count;
  5. }
  6. public synchronized void countDown(){
  7. count--;
  8. }
  9. public synchronized boolean hasNext(){
  10. return (count > 0);
  11. }
  12. public int getCount() {
  13. return count;
  14. }
  15. public void setCount(int count) {
  16. this.count = count;
  17. }
  18. }

ImportThread类
Java代码
  public class ImportThread extends Thread {
  1. private MyCountDown c;
  2. public ImportThread(MyCountDown c) {
  3. this.c = c;
  4. }
  5. @Override
  6. public void run() {
  7. System.out.println(Thread.currentThread().getName() + "开始...");//打印开始标记
  8. //Do something
  9. c.countDown();//计时器减1
  10. System.out.println(Thread.currentThread().getName() + "结束. 还有" + c.getCount() + " 个线程");//打印结束标记
  11. }
  12. }
  1. public class ImportThread extends Thread {
  2. private MyCountDown c;
  3. public ImportThread(MyCountDown c) {
  4. this.c = c;
  5. }
  6. @Override
  7. public void run() {
  8. System.out.println(Thread.currentThread().getName() + "开始...");//打印开始标记
  9. //Do something
  10. c.countDown();//计时器减1
  11. System.out.println(Thread.currentThread().getName() + "结束. 还有" + c.getCount() + " 个线程");//打印结束标记
  12. }
  13. }

主线程中
Java代码
System.out.println(Thread.currentThread().getName() + "开始");//打印开始标记
  1. MyCountDown c = new MyCountDown(threadNum);//初始化countDown
  2. for (int ii = 0; ii < threadNum; ii++) {//开threadNum个线程
  3. Thread t = new ImportThread(c);
  4. t.start();
  5. }
  6. while(true){//等待所有子线程执行完
  7. if(!c.hasNext()) break;
  8. }
  9. System.out.println(Thread.currentThread().getName() + "结束.");//打印结束标记
  1. System.out.println(Thread.currentThread().getName() + "开始");//打印开始标记
  2. MyCountDown c = new MyCountDown(threadNum);//初始化countDown
  3. for (int ii = 0; ii < threadNum; ii++) {//开threadNum个线程
  4. Thread t = new ImportThread(c);
  5. t.start();
  6. }
  7. while(true){//等待所有子线程执行完
  8. if(!c.hasNext()) break;
  9. }
  10. System.out.println(Thread.currentThread().getName() + "结束.");//打印结束标记

打印结果:
main开始
Thread-2开始...
Thread-1开始...
Thread-0开始...
Thread-3开始...
Thread-5开始...
Thread-4开始...
Thread-5结束. 还有5 个线程
Thread-1结束. 还有4 个线程
Thread-4结束. 还有3 个线程
Thread-2结束. 还有2 个线程
Thread-3结束. 还有1 个线程
Thread-0结束. 还有0 个线程
main结束.

更简单的方法:使用java.util.concurrent.CountDownLatch代替MyCountDown,用await()方法代替while(true){...}
ImportThread类
Java代码
public class ImportThread extends Thread {
  1. private CountDownLatch threadsSignal;
  2. public ImportThread(CountDownLatch threadsSignal) {
  3. this.threadsSignal = threadsSignal;
  4. }
  5. @Override
  6. public void run() {
  7. System.out.println(Thread.currentThread().getName() + "开始...");
  8. //Do somethings
  9. threadsSignal.countDown();//线程结束时计数器减1
  10. System.out.println(Thread.currentThread().getName() + "结束. 还有" + threadsSignal.getCount() + " 个线程");
  11. }
  12. }
  1. public class ImportThread extends Thread {
  2. private CountDownLatch threadsSignal;
  3. public ImportThread(CountDownLatch threadsSignal) {
  4. this.threadsSignal = threadsSignal;
  5. }
  6. @Override
  7. public void run() {
  8. System.out.println(Thread.currentThread().getName() + "开始...");
  9. //Do somethings
  10. threadsSignal.countDown();//线程结束时计数器减1
  11. System.out.println(Thread.currentThread().getName() + "结束. 还有" + threadsSignal.getCount() + " 个线程");
  12. }
  13. }

主线程中
Java代码 
CountDownLatch threadSignal = new CountDownLatch(threadNum);//初始化countDown
  1. for (int ii = 0; ii < threadNum; ii++) {//开threadNum个线程
  2. final Iterator<String> itt = it.get(ii);
  3. Thread t = new ImportThread(itt,sql,threadSignal);
  4. t.start();
  5. }
  6. threadSignal.await();//等待所有子线程执行完
  7. System.out.println(Thread.currentThread().getName() + "结束.");//打印结束标记
  1. CountDownLatch threadSignal = new CountDownLatch(threadNum);//初始化countDown
  2. for (int ii = 0; ii < threadNum; ii++) {//开threadNum个线程
  3. final Iterator<String> itt = it.get(ii);
  4. Thread t = new ImportThread(itt,sql,threadSignal);
  5. t.start();
  6. }
  7. threadSignal.await();//等待所有子线程执行完
  8. System.out.println(Thread.currentThread().getName() + "结束.");//打印结束标记

打印结果:
main开始
Thread-1开始...
Thread-0开始...
Thread-2开始...
Thread-3开始...
Thread-4开始...
Thread-5开始...
Thread-0结束. 还有5 个线程
Thread-1结束. 还有4 个线程
Thread-4结束. 还有3 个线程
Thread-2结束. 还有2 个线程
Thread-5结束. 还有1 个线程
Thread-3结束. 还有0 个线程
main结束. 

    本站是提供个人知识管理的网络存储空间，所有内容均由用户发布，不代表本站观点。请注意甄别内容中的联系方式、诱导购买等信息，谨防诈骗。如发现有害或侵权内容，请点击一键举报。
    转藏 分享 献花（0）

    来自： aaie_ > 《多线程》

    举报/认领

    0条评论

    发表

    请遵守用户 评论公约

    类似文章 更多