如何使用Java多管道流实现双TXT文件读取与数据过滤功能

• 输入输出流列表未初始化：in和out是ArrayList但没有预创建对应数量的PipedInputStream/PipedOutputStream实例，调用get()方法会直接抛出索引越界异常
• 连接方法调用不匹配：你定义的connectOutputTo需要传端口号参数，但LifeCycleManager中调用时没有传参，方法不匹配无法编译
• 端口关闭逻辑不全：现有closePorts只关闭了下标为0的输入输出端口，多端口场景下其余端口未正常关闭会导致流阻塞
• 缺少多输入流读取逻辑：MiddleFilter的specificComputationForFilter没有实现两个输入源的读取、过滤、分流输出的逻辑

1. 修正CommonFilterImpl类 #

补全构造方法用于初始化指定数量的输入输出端口，补全连接方法重载，修复端口关闭逻辑：

package Framework;

import java.io.EOFException;
import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.util.ArrayList;

public abstract class CommonFilterImpl implements CommonFilter {
    protected ArrayList<PipedInputStream> in = new ArrayList<>();
    protected ArrayList<PipedOutputStream> out = new ArrayList<>();

    // 新增构造方法：指定输入端口数、输出端口数，提前初始化流实例
    public CommonFilterImpl(int inputPortCount, int outputPortCount) {
        for (int i = 0; i < inputPortCount; i++) {
            in.add(new PipedInputStream());
        }
        for (int i = 0; i < outputPortCount; i++) {
            out.add(new PipedOutputStream());
        }
    }

    // 原有带端口号的连接方法
    public void connectOutputTo(CommonFilter nextFilter, int portNo) throws IOException {
        out.get(portNo).connect(nextFilter.getPipedInputStream(0));
    }

    // 新增无端口号的连接方法，默认用0号端口，兼容原有调用写法
    public void connectOutputTo(CommonFilter nextFilter) throws IOException {
        connectOutputTo(nextFilter, 0);
    }

    public void connectInputTo(CommonFilter previousFilter, int portNo) throws IOException {
        in.get(portNo).connect(previousFilter.getPipedOutputStream(0));
    }

    public PipedInputStream getPipedInputStream(int portNo) {
        return in.get(portNo);
    }

    public PipedOutputStream getPipedOutputStream(int portNo) {
        return out.get(portNo);
    }

    abstract public boolean specificComputationForFilter() throws IOException;

    public void run() {
        try {
            specificComputationForFilter();
        } catch (IOException e) {
            if (e instanceof EOFException) return;
            else e.printStackTrace();
        } finally {
            closePorts();
        }
    }

    // 修复端口关闭逻辑：遍历所有端口全部关闭
    private void closePorts() {
        try {
            for (PipedOutputStream os : out) {
                os.close();
            }
            for (PipedInputStream is : in) {
                is.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

2. 各Filter类继承示例 #

SourceFilter（源文件读取，0输入1输出） #

package Components.Source;

import Framework.CommonFilterImpl;
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.nio.charset.StandardCharsets;

public class SourceFilter extends CommonFilterImpl {
    private String filePath;

    public SourceFilter(String filePath) {
        // 源过滤器只有1个输出端口，0个输入端口
        super(0, 1);
        this.filePath = filePath;
    }

    @Override
    public boolean specificComputationForFilter() throws IOException {
        BufferedReader br = new BufferedReader(new FileReader(filePath));
        OutputStreamWriter writer = new OutputStreamWriter(out.get(0), StandardCharsets.UTF_8);
        String line;
        while ((line = br.readLine()) != null) {
            writer.write(line + "\n");
            writer.flush();
        }
        br.close();
        return true;
    }
}

MiddleFilter（中间过滤，2输入2输出） #

package Components.Middle;

import Framework.CommonFilterImpl;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

public class MiddleFilter extends CommonFilterImpl {
    public MiddleFilter() {
        // 中间过滤器2个输入端口，2个输出端口
        super(2, 2);
    }

    @Override
    public boolean specificComputationForFilter() throws IOException {
        // 读取第一个输入端口的学生数据
        BufferedReader studentReader = new BufferedReader(new InputStreamReader(in.get(0), StandardCharsets.UTF_8));
        List<String> studentLines = new ArrayList<>();
        String line;
        while ((line = studentReader.readLine()) != null) {
            // 此处添加学生数据自定义过滤逻辑
            studentLines.add(line);
        }

        // 读取第二个输入端口的课程数据
        BufferedReader courseReader = new BufferedReader(new InputStreamReader(in.get(1), StandardCharsets.UTF_8));
        List<String> courseLines = new ArrayList<>();
        while ((line = courseReader.readLine()) != null) {
            // 此处添加课程数据自定义过滤逻辑
            courseLines.add(line);
        }

        // 处理后的学生数据输出到第一个输出端口（对应Output1）
        OutputStreamWriter out1Writer = new OutputStreamWriter(out.get(0), StandardCharsets.UTF_8);
        for (String s : studentLines) {
            out1Writer.write(s + "\n");
            out1Writer.flush();
        }

        // 处理后的课程数据输出到第二个输出端口（对应Output2）
        OutputStreamWriter out2Writer = new OutputStreamWriter(out.get(1), StandardCharsets.UTF_8);
        for (String s : courseLines) {
            out2Writer.write(s + "\n");
            out2Writer.flush();
        }
        return true;
    }
}

SinkFilter（输出文件写入，1输入0输出） #

package Components.Sink;

import Framework.CommonFilterImpl;
import java.io.BufferedReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;

public class SinkFilter extends CommonFilterImpl {
    private String outputPath;

    public SinkFilter(String outputPath) {
        // 输出过滤器只有1个输入端口，0个输出端口
        super(1, 0);
        this.outputPath = outputPath;
    }

    @Override
    public boolean specificComputationForFilter() throws IOException {
        BufferedReader reader = new BufferedReader(new InputStreamReader(in.get(0), StandardCharsets.UTF_8));
        FileWriter writer = new FileWriter(outputPath, StandardCharsets.UTF_8);
        String line;
        while ((line = reader.readLine()) != null) {
            writer.write(line + "\n");
            writer.flush();
        }
        writer.close();
        return true;
    }
}

3. 修正LifeCycleManager连接逻辑 #

package Framework;
import Components.Middle.MiddleFilter;
import Components.Sink.SinkFilter;
import Components.Source.SourceFilter;
public class LifeCycleManager {
    public static void main(String[] args) {
        try {
            CommonFilter srcFilterS = new SourceFilter("Students.txt");
            CommonFilter srcFilterC = new SourceFilter("Courses.txt");
            CommonFilter sinkFilter1 = new SinkFilter("Output1.txt");
            CommonFilter sinkFilter2 = new SinkFilter("Output2.txt");
            CommonFilter middleFilter = new MiddleFilter();

            // 学生源接中间过滤器0号输入端口，课程源接中间过滤器1号输入端口
            srcFilterS.connectOutputTo(middleFilter, 0);
            srcFilterC.connectOutputTo(middleFilter, 1);
            // 中间过滤器0号输出接Output1，1号输出接Output2
            middleFilter.connectOutputTo(sinkFilter1, 0);
            middleFilter.connectOutputTo(sinkFilter2, 1);

            Thread thread1 = new Thread(srcFilterS);
            Thread thread2 = new Thread(srcFilterC);
            Thread thread3 = new Thread(middleFilter);
            Thread thread4 = new Thread(sinkFilter2);
            Thread thread5 = new Thread(sinkFilter1);

            thread1.start();
            thread2.start();
            thread3.start();
            thread4.start();
            thread5.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

• 如果你需要对学生和课程数据做关联过滤（比如匹配学生选的课程），可以调整MiddleFilter里的逻辑，先把两个文件的内容全部读取缓存之后再做关联处理，避免流读取阻塞
• PipedStream的读写是线程安全的，但是不要在同一个线程里同时读写同一个PipedStream，会导致死锁，你当前每个Filter独立开线程的写法符合要求

内容的提问来源于stack exchange，提问作者Honam