并行流水线思想

Posted 2020-09-17 起个po名真费劲

并不是所有的操作都能进行并发计算的，比如有关于数据相关性的操作，如(B＋C) * B,操作之间都会有依赖关系的。

可以使用流水线方式充分利用多核环境。

开多个线程。每一条线程相当于车间，负责执行一小部分工作，执行完传递给下一个线程，线程之间可以通过阻塞队列来进行通信。

具体实现：

其中Div， Multiply， Plus相当于车间，

Msg相当于传输的介质，

PStreamMain为主线程，相当于生产者。

package assemblyline;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

public class Div implements Runnable{
    public static BlockingQueue<Msg> bq = new LinkedBlockingDeque<Msg>();

    @Override
    public void run() {
        while(true){
            try{
                Msg msg = bq.take();
                msg.i = msg.i / 2;
                System.out.println(msg.orgStr + " = " + msg.i);
            }catch(Exception e){
                e.printStackTrace();
            }
        }
    }
    
}

package assemblyline;

public class Msg {
    public double i;
    public double j;
    public String orgStr = null;
}

package assemblyline;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

public class Multiply implements Runnable {
    public static BlockingQueue<Msg> bq = new LinkedBlockingDeque<Msg>();
    @Override
    public void run() {
        while(true){
            try{
                Msg msg = bq.take();
                msg.i = msg.i * msg.j;
                Div.bq.add(msg);
            }catch(Exception e){
                e.printStackTrace();
            }
        }
    }

}

package assemblyline;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

public class Plus implements Runnable{
    public static BlockingQueue<Msg> bq = new LinkedBlockingDeque<Msg>();

    @Override
    public void run() {
        while(true){
            try{
                Msg msg = bq.take();
                msg.j = msg.i + msg.j; 
                Multiply.bq.add(msg);
            }catch(Exception e){
                e.printStackTrace();
            }
        }
    }
    
}

package assemblyline;

public class PStreamMain {
    public static void main(String[] args) {
        new Thread(new Plus()).start();
        new Thread(new Multiply()).start();
        new Thread(new Div()).start();
        for(int i = 1;i <= 1000;i ++){
            for(int j = 1;j <= 1000;j ++){
                Msg msg = new Msg();
                msg.i = i;
                msg.j = j;
                msg.orgStr = "((" + i + "+" + j + ")*" + i +")/2";
                Plus.bq.add(msg);
            }
        }
    }
}