流水线并发框架架构
V1版本1：此中定义了一个 Task 类和三个继续自 Thread 类的子类 TA、TB 和 TC。
[*] Task 类：
[*]num 是一个整数变量，用于存储任务的结果。
[*]taskA()、taskB() 和 taskC() 是三个任务方法，分别模仿了一些计算和等候的操作。
[*] TA、TB、TC 类：
[*]这三个类分别表示三个差别的线程，每个线程实行一组任务。
[*]每个线程吸收一个 ArrayList<Task> 类型的列表作为参数，在 run() 方法中，通过迭代列表，对每个 Task 对象调用相应的任务方法。
[*] Main 类：
[*]在 main 方法中，首先创建了一个包罗50个 Task 对象的列表 list。
[*]创建了 TA、TB 和 TC 的实例，并将 list 作为参数通报给它们。
[*]启动了三个线程，分别实行 taskA、taskB 和 taskC。
[*]创建了一个额外的线程，该线程每秒输出部分任务的结果，以便在实行过程中观察任务的完成环境。
[*]使用 join() 方法等候三个线程实行完成。
[*] 任务实行过程：
[*]TA 线程每次迭代调用 taskA()，导致 num 值增加 10。
[*]TB 线程每次迭代调用 taskB()，导致 num 值乘以 20。
[*]TC 线程每次迭代调用 taskC()，导致 num 值乘以自身。
[*] 输出：
[*]在额外的线程中，每秒输出 list 中每个第五个任务的结果。
须要注意的是，由于线程之间并发实行，输出结果可能会交错。此外，对 num 的操作可能导致竞态条件，可能须要使用同步机制来确保线程安全性。
import java.util.ArrayList;
// A-B-C ： 40000 正确完成任务后的结果
public class Task {
int num;
public void taskA() {
try {
Thread.sleep(50);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
num += 10;
}
public void taskB() {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
num *= 20;
}
public void taskC() {
try {
Thread.sleep(650);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
num *= num;
}
}
class TA extends Thread {
ArrayList<Task> list;
public TA(ArrayList<Task> list) {
this.list=list;
}
@Override
public void run() {
for(int i=0;i<list.size();i++){
Task task = list.get(i);
task.taskA();
}
}
}
class TB extends Thread {
ArrayList<Task> list;
public TB(ArrayList<Task> list) {
this.list=list;
}
@Override
public void run() {
for(int i=0;i<list.size();i++){
Task task = list.get(i);
task.taskB();
}
}
}
class TC extends Thread {
ArrayList<Task> list;
public TC(ArrayList<Task> list) {
this.list=list;
}
@Override
public void run() {
for(int i=0;i<list.size();i++){
Task task = list.get(i);
task.taskC();
}
}
}
class Main {
public static void main(String[] args) {
// 1： 定量任务
ArrayList<Task> list = new ArrayList<>();
for (int i = 0; i < 50; i++) {
list.add(new Task());
}
TA ta = new TA(list);
TB tb = new TB(list);
TC tc = new TC(list);
ta.start();
tb.start();
tc.start();
//监听状态线程
new Thread(){
public void run(){
while(true){
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
int size=0;
for(int i=0;i< list.size();i++){
if(i%5==0){
System.out.println(list.get(i).num);
}
}
//System.out.println(list.size());
//System.out.println("已完成"+size+"任务");
}
}
}.start();
try {
ta.join();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
try {
tb.join();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
try {
tc.join();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
} 可见性题目
可见性题目是多线程并发编程中常见的一个挑战，它涉及到一个线程对共享变量的修改是否可以或许立即被其他线程看到。在没有适当同步机制的环境下，由于每个线程有自己的工作内存，可能会导致一个线程对变量的修改对其他线程不可见。
在 Java 中，主要有两个因素导致可见性题目：
[*] 线程工作内存： 每个线程都有自己的工作内存，用于存储主内存中的变量的副本。线程在实行时，操作的是工作内存中的变量，而不是直接操作主内存中的变量。
[*] 指令重排序： 编译器和处理器为了进步实行效率，可能会对指令进行重排序。这就可能导致一些操作的实行顺序与代码中的顺序差别等。
为了办理可见性题目，Java 提供了 volatile 关键字。使用 volatile 关键字修饰的变量具有以下特性：
[*] 可见性： 当一个线程修改了 volatile 变量的值，该变量的新值会立即被写回主内存，而其他线程在读取该变量时会直接从主内存中获取最新的值。
[*] 克制指令重排序： volatile 关键字克制指令重排序，确保了变量的修改按照代码的顺序实行。
例
例子中演示了使用 volatile 关键字办理多线程可见性题目的环境。
[*] volatile 关键字：
[*]volatile 修饰的变量 flag 表示该变量是易变的，并且任何线程对它的修改都会立即反映到其他线程中。这办理了多线程之间的可见性题目，确保一个线程对该变量的修改对其他线程是可见的。
[*] main 方法：
[*]创建了一个名为 t1 的线程，该线程在运行时将 flag 设置为 true。
[*]创建了一个匿名线程，该线程在运行时通过循环检查 flag 的值，一直比及 flag 变为 true 才输出 "T2-end"。
[*]在 main 方法中，通过 Thread.sleep(2000) 使得主线程休眠 2 秒，以确保 t1 线程有足够的时间来设置 flag 的值。
[*]启动了 t1 线程。
[*] 输出：
[*]T1 - start: t1 线程开始实行，将 flag 设置为 true。
[*]T2-start: 另一个线程开始实行，进入循环等候，由于 flag 初始值为 false，因此一直等候。
[*]T1 - end: t1 线程设置 flag 为 true。
[*]T2-end: 循环竣事，输出 "T2-end"。
[*] 关于注释的分析：
[*]在代码中有一行注释 // System.out.print("");，这是一种办理可见性题目的“空循环”技巧。通过参加一个空的循环，可以迫使线程重新读取共享变量，从而办理可见性题目。在这个例子中，由于使用了 volatile 关键字，这行注释实际上不再须要，由于 volatile 自己保证了可见性。
总的来说，这个例子展示了使用 volatile 关键字办理多线程可见性题目的环境，确保一个线程对共享变量的修改可以或许及时被其他线程感知。
public class Test {
volatile static boolean flag;// 每次使用变量时 都会重新拷贝一份副本过来
public static void main(String[] args) {
Thread t1 = new Thread() {
@Override
public void run() {
System.out.println("T1 - start");
flag = true;
System.out.println("T1 - end");
}
};
new Thread() {
@Override
public void run() {
System.out.println("T2-start");
while (!flag) {
// System.out.print("");//可见性问题
}
System.out.println("T2-end");
}
}.start();
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
t1.start();
}
} V2
版本2：展示了一个多线程任务的场景，此中包罗了三个任务（taskA、taskB、taskC），每个任务依次实行，且必须按照一定的顺序。在 Main 类的 main 方法中，创建了一个包罗 100 个 Task 对象的列表，并启动了三个线程 TA、TB、TC 分别实行这些任务。另外，另有一个额外的线程用于监听任务完成状态。
让我们详细讲解这段代码：
[*] Task 类：
[*]每个 Task 对象包罗一个整数变量 num 和三个布尔标志 flagA、flagB、flagC。
[*]taskA、taskB、taskC 方法模仿了三个任务的实行过程，每个任务在实行前会先检查前置任务是否完成（通过相应的标志）。
[*]isEnd 方法用于判定任务是否全部完成。
[*] TA、TB、TC 类：
[*]这三个类分别表示实行任务 A、B、C 的线程，它们负责迭代任务列表，实行相应的任务方法。
[*]每个线程在实行任务的过程中，通过检查任务的标志来确保按照顺序实行。
[*] Main 类：
[*]创建了一个包罗 100 个 Task 对象的列表 list。
[*]创建了 TA、TB 和 TC 的实例，并将 list 作为参数通报给它们。
[*]启动了三个线程，分别实行 taskA、taskB 和 taskC。
[*]创建了一个额外的线程，该线程每秒输出已完成的任务数量和任务列表的总巨细。
[*]使用 join 方法等候三个任务线程实行完成。
[*] 输出：
[*]每个任务线程在实行时，会输出实行的次数和已完成的任务数量。
[*]额外的监听线程每秒输出已完成的任务数量和任务列表的总巨细。
[*] 任务实行过程：
[*]每个任务在实行前都会检查前置任务是否完成，确保按照任务顺序实行。
[*]每个任务实行完成后，会设置相应的标志，表示该任务已完成。
[*]额外的监听线程会每秒输出已完成的任务数量和任务列表的总巨细，直到所有任务完成。
该代码通过使用标志和多线程的方式模仿了任务的顺序实行，确保了每个任务在满足条件时才实行。须要注意的是，在实际生产环境中，更复杂的同步机制可能会更实用。
import java.util.ArrayList;
public class Task {
int num;
boolean flagA = false;
boolean flagB = false;
boolean flagC = false;
public void taskA() {
if (!flagA) {
sleep(40);
num += 10;
flagA = true;
}
}
public void taskB() {
if (!flagB && flagA) {
sleep(30);
num *= 20;
flagB = true;
}
}
public void taskC() {
if (flagB && !flagC) {
sleep(30);
num *= num;
flagC = true;
}
}
public boolean isEnd() {
return flagC;
}
public void sleep(long time) {
try {
Thread.sleep(time);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
class TA extends Thread {
ArrayList<Task> list;
public TA(ArrayList<Task> list) {
this.list = list;
}
@Override
public void run() {
int count = 0;
while (true) {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
System.out.println("TA - run" + count++);
int size = 0;
for (int i = 0; i < list.size(); i++) {
Task task = list.get(i);
task.taskA();
if (task.flagA) {
size++;
}
}
System.out.println("TA - 完成了" + size);
if (size == list.size()) {
break;
}
}
}
}
class TB extends Thread {
ArrayList<Task> list;
public TB(ArrayList<Task> list) {
this.list = list;
}
@Override
public void run() {
int count = 0;
while (true) {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
System.out.println("TB - run" + count++);
int size = 0;
for (int i = 0; i < list.size(); i++) {
Task task = list.get(i);
task.taskB();
if (task.flagB) {
size++;
}
}
System.out.println("TB - 完成了" + size);
if (size == list.size()) {
break;
}
}
}
}
class TC extends Thread {
ArrayList<Task> list;
public TC(ArrayList<Task> list) {
this.list = list;
}
@Override
public void run() {
int count = 0;
while (true) {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
System.out.println("TC - run" + count++);
int size = 0;
for (int i = 0; i < list.size(); i++) {
Task task = list.get(i);
task.taskC();
if (task.flagC) {
size++;
}
}
System.out.println("TC - 完成了" + size);
if (size == list.size()) {
break;
}
}
}
}
class Main {
public static void main(String[] args) {
// 1： 定量任务
ArrayList<Task> list = new ArrayList<>();
for (int i = 0; i < 100; i++) {
list.add(new Task());
}
TA ta = new TA(list);
TB tb = new TB(list);
TC tc = new TC(list);
ta.start();
tb.start();
tc.start();
// 监听状态线程
new Thread() {
@Override
public void run() {
while (true) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
int size = 0;
for (int i = 0; i < list.size(); i++) {
if (list.get(i).num == 40000) {
size++;
}
}
System.out.println(list.size());
System.out.println("已完成：" + size + "任务");
}
}
}.start();
try {
ta.join();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
try {
tb.join();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
try {
tc.join();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
// for (int i = 0; i < list.size(); i++) {
// System.out.println(list.get(i).num);
// }
}
} V3
版本3：代码进一步改进了多线程任务的实现，引入了 volatile 关键字来确保线程之间的可见性。同时，通过在任务的 while 循环中进行阻塞等候，实现了任务的顺序实行。
[*] Task 类：
[*]volatile 关键字被添加到 flagA、flagB、flagC 上，确保了对这些标志的修改立即可见于其他线程。
[*]任务方法 taskA、taskB、taskC 中的 while 循环用于在前置任务完成前阻塞当前任务的实行，确保按照任务顺序实行。
[*] TA、TB、TC 类：
[*]这三个类分别表示实行任务 A、B、C 的线程，它们负责迭代任务列表，实行相应的任务方法。
[*]在任务 B 和任务 C 的实行中，通过 while 循环进行阻塞等候前置任务完成。
[*] Main 类：
[*]创建了一个包罗 500 个 Task 对象的列表 list。
[*]创建了 TA、TB 和 TC 的实例，并将 list 作为参数通报给它们。
[*]启动了三个线程，分别实行 taskA、taskB 和 taskC。
[*]创建了一个额外的线程，该线程每秒输出已完成的任务数量和任务列表的总巨细。
[*]使用 join 方法等候三个任务线程实行完成。
[*]计算任务实行的耗时，并输出每个任务的结果。
[*] 输出：
[*]每个任务线程在实行时，会输出实行的次数和已完成的任务数量。
[*]额外的监听线程每秒输出已完成的任务数量和任务列表的总巨细。
[*]最后输出每个任务的结果和整个任务实行的耗时。
通过引入 volatile 关键字和阻塞等候的方式，确保了任务的按顺序实行，同时在输出中显示了任务实行的次数、已完成的任务数量以及整个任务实行的耗时。这种方式更加妥当地处理了多线程环境下的任务实行顺序题目。
import java.util.ArrayList;
public class Task {
int num;
volatile boolean flagA = false;
volatile boolean flagB = false;
volatile boolean flagC = false;
public void taskA() {
if (!flagA) {
sleep(40);
num += 10;
flagA = true;
}
}
public void taskB() {
while (!flagA) {// 阻塞
}
if (!flagB) {
sleep(30);
num *= 20;
flagB = true;
}
}
public void taskC() {
while (!flagB) {// 阻塞
}
if (!flagC) {
sleep(30);
num *= num;
flagC = true;
}
}
public boolean isEnd() {
return flagC;
}
public void sleep(long time) {
try {
Thread.sleep(time);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
class TA extends Thread {
ArrayList<Task> list;
public TA(ArrayList<Task> list) {
this.list = list;
}
@Override
public void run() {
for (int i = 0; i < list.size(); i++) {
Task task = list.get(i);
task.taskA();
}
}
}
class TB extends Thread {
ArrayList<Task> list;
public TB(ArrayList<Task> list) {
this.list = list;
}
@Override
public void run() {
for (int i = 0; i < list.size(); i++) {
Task task = list.get(i);
task.taskB();
}
}
}
class TC extends Thread {
ArrayList<Task> list;
public TC(ArrayList<Task> list) {
this.list = list;
}
@Override
public void run() {
for (int i = 0; i < list.size(); i++) {
Task task = list.get(i);
task.taskC();
}
}
}
class Main {
public static void main(String[] args) {
// 1： 定量任务
ArrayList<Task> list = new ArrayList<>();
for (int i = 0; i < 500; i++) {
list.add(new Task());
}
TA ta = new TA(list);
TB tb = new TB(list);
TC tc = new TC(list);
ta.start();
tb.start();
tc.start();
// 监听状态线程
new Thread() {
@Override
public void run() {
while (true) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
int size = 0;
for (int i = 0; i < list.size(); i++) {
if (list.get(i).num == 40000) {
size++;
}
}
System.out.println(list.size());
System.out.println("已完成：" + size + "任务");
}
}
}.start();
long start = System.currentTimeMillis();
try {
ta.join();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
try {
tb.join();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
try {
tc.join();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
long end = System.currentTimeMillis();
for (int i = 0; i < list.size(); i++) {
System.out.println(list.get(i).num);
}
System.out.println("耗时：" + (end - start) + " ms");
}
}
V4
版本4：展示了一种通过使用队列（LinkedList）和质料库（taskA、taskB、taskC）的方式来实现任务的生产和消费。每个任务分别在差别的线程中实行，通过队列和质料库的操作来确保任务按照特定的顺序实行。
[*] Task 类：
[*]taskA、taskB、taskC 分别对应三个差别的任务，对 num 进行差别的操作。
[*] TA、TB、TC 类：
[*]这三个类分别表示实行任务 A、B、C 的线程，通过队列和质料库的操作来确保任务按照特定的顺序实行。
[*]run 方法中使用 poll 方法从队列中取出任务，实行相应的任务方法，然后将任务放入下一个阶段的质料库中。
[*] Main 类：
[*]创建了 taskA、taskB、taskC 作为任务的质料库。
[*]创建了 tasks 作为制品库，用于存储已完成的任务。
[*]创建了 TA、TB 和 TC 的实例，并将相应的质料库和制品库通报给它们。
[*]启动了三个线程，分别实行 taskA、taskB 和 taskC。
[*]创建了一个额外的线程，该线程每秒输出已完成的任务数量和任务列表的总巨细。
[*] 输出：
[*]每个任务线程在实行时，会输出实行的次数和已完成的任务数量。
[*]额外的监听线程每秒输出已完成的任务数量和任务列表的总巨细。
通过这种队列和质料库的方式，确保了任务的顺序实行。任务被依次推送到差别的阶段，从而保证了任务的有序性。在这个模型中，每个任务线程负责从上一个阶段的质料库取任务，实行后将任务放入下一个阶段的质料库，以此类推，直至完成。
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.concurrent.ArrayBlockingQueue;
public class Task {
int num;
public void taskA() {
num += 10;
}
public void taskB() {
num *= 20;
}
public void taskC() {
num *= num;
}
}
class TA extends Thread {
LinkedList<Task> taskB;
LinkedList<Task> taskA;
public TA(LinkedList<Task> taskA, LinkedList<Task> taskB) {
this.taskA = taskA;
this.taskB = taskB;
}
@Override
public void run() {
while (true) {
Task task = taskA.poll();// 出列
if (task != null) {
task.taskA();// 执行A任务
taskB.offer(task);// 入队B
}
}
}
}
class TB extends Thread {
LinkedList<Task> taskB;
LinkedList<Task> taskC;
public TB(LinkedList<Task> taskB, LinkedList<Task> taskC) {
this.taskB = taskB;
this.taskC = taskC;
}
@Override
public void run() {
while (true) {
Task task = taskB.poll();// 出列
if (task != null) {
task.taskB();// 执行A任务
taskC.offer(task);// 入队B
}
}
}
}
class TC extends Thread {
ArrayList<Task> tasks;
LinkedList<Task> taskC;
public TC(LinkedList<Task> taskC, ArrayList<Task> tasks) {
this.taskC = taskC;
this.tasks = tasks;
}
@Override
public void run() {
while (true) {
Task task = taskC.poll();// 出列
if (task != null) {
task.taskC();// 执行A任务
tasks.add(task);// 入队B
}
}
}
}
class Main {
public static void main(String[] args) {
// 1： 定量任务
LinkedList<Task> taskA = new LinkedList<>();// A的原料库
LinkedList<Task> taskB = new LinkedList<>();// B的原料库
LinkedList<Task> taskC = new LinkedList<>();// C的原料库
ArrayList<Task> tasks = new ArrayList<>();// 成品库
ArrayBlockingQueue<Task> tasks1 = new ArrayBlockingQueue<>(50);
for (int i = 0; i < 500; i++) {
taskA.offer(new Task());// 入队
}
TA ta = new TA(taskA, taskB);
TB tb = new TB(taskB, taskC);
TC tc = new TC(taskC, tasks);
ta.start();
tb.start();
tc.start();
// 监听状态线程
new Thread() {
@Override
public void run() {
while (true) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
int size = 0;
for (int i = 0; i < tasks.size(); i++) {
System.out.println(tasks.get(i).num);
if (tasks.get(i).num == 40000) {
size++;
}
}
System.out.println(tasks.size());
System.out.println("已完成：" + size + "任务");
}
}
}.start();
}
} V5
版本5：代码是一个改进版本，使用了 ArrayBlockingQueue 作为任务的质料库和制品库，以及通过队列操作来保证任务的有序实行。
[*] Task 类：
[*]taskA、taskB、taskC 方法分别对应三个差别的任务，对 num 进行差别的操作。
[*] TA、TB、TC 类：
[*]这三个类分别表示实行任务 A、B、C 的线程，通过 ArrayBlockingQueue 实现了线程之间的协作，确保任务按照特定的顺序实行。
[*]run 方法中使用 take 方法从队列中取出任务，实行相应的任务方法，然后将任务放入下一个阶段的队列中。
[*] Main 类：
[*]创建了 taskA、taskB、taskC 作为任务的质料库，使用 ArrayBlockingQueue 进行初始化。
[*]创建了 tasks 作为制品库，用于存储已完成的任务。
[*]创建了 TA、TB 和 TC 的实例，并将相应的质料库和制品库通报给它们。
[*]启动了三个线程，分别实行 taskA、taskB 和 taskC。
[*]创建了一个额外的线程，该线程每秒输出已完成的任务数量和任务列表的总巨细。
[*] 输出：
[*]每个任务线程在实行时，会输出实行的次数和已完成的任务数量。
[*]额外的监听线程每秒输出已完成的任务数量和任务列表的总巨细。
通过使用 ArrayBlockingQueue 作为质料库和制品库，以及使用 take 和 put 方法来保证线程之间的协作，这段代码实现了有序实行的任务模型。每个任务线程负责从上一个阶段的队列取任务，实行后将任务放入下一个阶段的队列，从而保证了任务的有序性。
import java.util.ArrayList;
import java.util.concurrent.ArrayBlockingQueue;
public class Task {
int num;
public void taskA() {
num += 10;
}
public void taskB() {
num *= 20;
}
public void taskC() {
num *= num;
}
}
class TA extends Thread {
ArrayBlockingQueue<Task> taskB;
ArrayBlockingQueue<Task> taskA;
public TA(ArrayBlockingQueue<Task> taskA, ArrayBlockingQueue<Task> taskB) {
this.taskA = taskA;
this.taskB = taskB;
}
@Override
public void run() {
while (true) {
try {
Task task = taskA.take(); // 出队
task.taskA(); // 执行A任务
taskB.put(task); // 入队B
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
}
class TB extends Thread {
ArrayBlockingQueue<Task> taskB;
ArrayBlockingQueue<Task> taskC;
public TB(ArrayBlockingQueue<Task> taskB, ArrayBlockingQueue<Task> taskC) {
this.taskB = taskB;
this.taskC = taskC;
}
@Override
public void run() {
while (true) {
try {
Task task = taskB.take(); // 出队
task.taskB(); // 执行B任务
taskC.put(task); // 入队C
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
}
class TC extends Thread {
ArrayList<Task> tasks;
ArrayBlockingQueue<Task> taskC;
public TC(ArrayBlockingQueue<Task> taskC, ArrayList<Task> tasks) {
this.taskC = taskC;
this.tasks = tasks;
}
@Override
public void run() {
while (true) {
try {
Task task = taskC.take(); // 出队
task.taskC(); // 执行C任务
tasks.add(task); // 入队结果列表
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
}
class Main {
public static void main(String[] args) {
// 1：定量任务
ArrayBlockingQueue<Task> taskA = new ArrayBlockingQueue<>(500); // A的原料库
ArrayBlockingQueue<Task> taskB = new ArrayBlockingQueue<>(500); // B的原料库
ArrayBlockingQueue<Task> taskC = new ArrayBlockingQueue<>(500); // C的原料库
ArrayList<Task> tasks = new ArrayList<>(); // 成品库
for (int i = 0; i < 500; i++) {
taskA.offer(new Task()); // 入队
}
TA ta = new TA(taskA, taskB);
TB tb = new TB(taskB, taskC);
TC tc = new TC(taskC, tasks);
ta.start();
tb.start();
tc.start();
// 监听线程状态
new Thread() {
@Override
public void run() {
while (true) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
int size = 0;
for (int i = 0; i < tasks.size(); i++) {
System.out.println(tasks.get(i).num);
if (tasks.get(i).num == 40000) {
size++;
}
}
System.out.println(tasks.size());
System.out.println("已完成：" + size + "任务");
}
}
}.start();
}
} https://i-blog.csdnimg.cn/blog_migrate/7376de369eabc9ef99a22f55ba4bfa7e.png
阻塞队列
ArrayBlockingQueue 是 Java 中 BlockingQueue 接口的一个具体实现，它基于数组实现的有界阻塞队列。以下是对 ArrayBlockingQueue 的详细讲解：
特点和用途：
[*] 有界队列： ArrayBlockingQueue 是一个有界队列，其容量在创建时被指定，不能动态扩展。这意味着队列中的元素数量不能超过指定的容量。
[*] 阻塞操作： 当队列满时，试图将元素放入队列的操作将被阻塞，直到队列有空间。当队列为空时，试图从队列中取出元素的操作将被阻塞，直到队列中有元素。
[*] 线程安全： ArrayBlockingQueue 提供了在多线程环境下安全使用的机制，内部实现使用了锁来掩护队列的操作。
构造方法：
[*] ArrayBlockingQueue(int capacity)： 创建一个指定容量的 ArrayBlockingQueue，默认环境下为公平战略，即等候时间最长的线程将被优先实行。
[*] ArrayBlockingQueue(int capacity, boolean fair)： 创建一个指定容量和公平性战略的 ArrayBlockingQueue。
[*] ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c)： 创建一个包罗指定聚集元素的 ArrayBlockingQueue，容量为指定容量，公平性由 fair 参数决定。
主要方法：
[*] 放入元素：
[*] put(E e)： 将指定元素放入队列，假如队列满，则阻塞直到队列有空间。
[*] offer(E e, long timeout, TimeUnit unit)： 将指定元素放入队列，假如队列满，则等候指定时间，超时后返回 false。
[*] 取出元素：
[*] take()： 取出并删除队列的头部元素，假如队列为空，则阻塞直到队列有元素可取。
[*] poll(long timeout, TimeUnit unit)： 取出并删除队列的头部元素，假如队列为空，则等候指定时间，超时后返回 null。
[*] 其他方法：
[*] remainingCapacity()： 返回队列的剩余容量。
[*] size()： 返回队列中的元素数量。
[*] peek()： 返回队列的头部元素，但不删除。假如队列为空，则返回 null。
[*] contains(Object o)： 判定队列是否包罗指定元素。
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]