网站的建设与板块,一站式营销型网站建设服务,上海著名网站设计公司,dw网页制作在线编辑本文介绍两种BlockingDeque在多线程任务处理时正确结束的方法 一般最开始简单的多线程处理任务过程
把总任务放入BlockingDeque创建多个线程#xff0c;每个线程内逻辑时#xff0c;判断BlockingDeque任务是否处理完#xff0c;处理完退出#xff0c;还有任务就BlockingDe…本文介绍两种BlockingDeque在多线程任务处理时正确结束的方法 一般最开始简单的多线程处理任务过程
把总任务放入BlockingDeque创建多个线程每个线程内逻辑时判断BlockingDeque任务是否处理完处理完退出还有任务就BlockingDeque.take()取任务处理主线程join等待多线程处理完收尾处理完成任务。
最开始版本代码10个任务3个线程来处理
package org.example;import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;public class Main3 {public static void main(String[] args){System.out.println(start);BlockingDequeInteger task new LinkedBlockingDeque();for (int i 0; i 10; i) {task.add(i);}ListThread workers new ArrayList();for (int i 0; i 3; i) {Thread worker new Thread(()-{while (true) {Integer data null;try {if (task.size()0) {System.out.println(Thread.currentThread().getName() quit);break;}
// Thread.sleep(100); // 默认任务耗时data task.take();} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println(Thread.currentThread().getName() do data);}});workers.add(worker);worker.start();}for (Thread worker: workers) {try {worker.join();} catch (InterruptedException e) {throw new RuntimeException(e);}}System.out.println(job done);}
}运行之后感觉非常好完美实现逻辑
但是当把上面的任务数加到200线程数加到30上面线程sleep的注释打开再次运行就会发现主进程最后会被一直卡着不结束说明多线程没有正确判断任务结束线程不安全
上面的子线程内的size()等于0到下面的BlockingDeque.take()取任务这段之间的代码这段不是线程安全的
让线程正确判断任务结束而且要线程安全的三种方法推荐第二种兼顾效率和兼容正确性
判断任务结束这段代码加synchronized约束起来实现线程安全太慢给总任务task内加入和线程相同数量的停止标志marker使用BlockingDeque.poll(超时时间) 异常数据检查需要检查异常数据
使用synchronized约束
package org.example;import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;public class Main {public static void main(String[] args) {System.out.println(start);BlockingDequeInteger task new LinkedBlockingDeque();for (int i 0; i 20; i) {task.add(i);}ListThread workers new ArrayList();for (int i 0; i 3; i) {Thread worker new Thread(()-{while (true) {Integer data null;synchronized (task) {if (task.size() 0) {System.out.println(Thread.currentThread().getName() quit);break;}try {data task.take();} catch (InterruptedException e) {throw new RuntimeException(e);}}try {Thread.sleep(300);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println(Thread.currentThread().getName() do data);}});workers.add(worker);worker.start();}for (Thread worker: workers) {try {worker.join();} catch (InterruptedException e) {throw new RuntimeException(e);}}System.out.println(job done);}
}总任务添加stop marker停止标志
package org.example;import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;public class Main2 {public static void main(String[] args){System.out.println(start);BlockingDequeInteger task new LinkedBlockingDeque();for (int i 0; i 20; i) {task.add(i);}ListThread workers new ArrayList();for (int i 0; i 3; i) task.add(99);for (int i 0; i 3; i) {Thread worker new Thread(()-{while (true) {Integer data null;try {data task.take();if (data 99) {System.out.println(Thread.currentThread().getName() quit);break;}} catch (InterruptedException e) {throw new RuntimeException(e);}try {Thread.sleep(300);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println(Thread.currentThread().getName() do data);}});workers.add(worker);worker.start();}for (Thread worker: workers) {try {worker.join();} catch (InterruptedException e) {throw new RuntimeException(e);}}System.out.println(job done);}
}使用BlockingDeque.poll(超时时间)避免了take的永久性等待问题但是会取到null值要加判断处理
package org.example;import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;public class Main4 {public static void main(String[] args){System.out.println(start);BlockingDequeInteger task new LinkedBlockingDeque();for (int i 0; i 200; i) {task.add(i);}ListThread workers new ArrayList();for (int i 0; i 30; i) {Thread worker new Thread(()-{while (true) {Integer data null;try {if (task.size()0) {System.out.println(Thread.currentThread().getName() quit);break;}Thread.sleep(100); // 默认任务耗时data task.poll(1000, TimeUnit.MILLISECONDS);if (data null) {System.out.println(Thread.currentThread().getName() get null);continue;}} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println(Thread.currentThread().getName() do data);}});workers.add(worker);worker.start();}for (Thread worker: workers) {try {worker.join();} catch (InterruptedException e) {throw new RuntimeException(e);}}System.out.println(job done);}
}
