java - 流水线设计模式实现

Question

这是一个关于流水线实现的设计问题。以下是我的幼稚实现。

管道中各个步骤/阶段的接口(interface):

public interface Step<T, U> {
    public U execute(T input);
}

管道中步骤/阶段的具体实现:

public class StepOne implements Step<Integer, Integer> {
    @Override
    public Integer execute(Integer input) {
        return input + 100;
    }
}

public class StepTwo implements Step<Integer, Integer> {
    @Override
    public Integer execute(Integer input) {
        return input + 500;
    }
}

public class StepThree implements Step<Integer, String> {
    @Override
    public String execute(Integer input) {
        return "The final amount is " + input;
    }
}

管道类将保存/注册管道中的步骤并一个接一个地执行它们:

public class Pipeline {
    private List<Step> pipelineSteps = new ArrayList<>();
    private Object firstStepInput = 100;

    public void addStep(Step step) {
        pipelineSteps.add(step);
    }

    public void execute() {
        for (Step step : pipelineSteps) {
            Object out = step.execute(firstStepInput);
            firstStepInput = out;
        }
   }
}

执行管道的Diver程序:

public class Main {
    public static void main(String[] args) {
        Pipeline pipeline = new Pipeline();
        pipeline.addStep(new StepOne());
        pipeline.addStep(new StepTwo());
        pipeline.addStep(new StepThree());

        pipeline.execute();
    } 
}

但是，如您所见，幼稚的实现有很多限制。

其中一个主要的问题是，由于要求每个步骤的输出可以是任何类型，因此幼稚的实现不是类型安全的(Pipeline 类中的 execute 方法)。如果我碰巧错误地连接了管道中的步骤，应用程序将会失败。

谁能帮我设计解决方案，添加到我的代码中，或者指出一个已经存在的模式来解决这个问题？

Answer 1

为什么需要一个额外的 Pipeline 类？我认为您可以删除中间人。这将使您的 api 更简单，例如:

Step<Integer, String> source = Step.of(Object::toString);
Step<Integer, Integer> toHex = source.pipe(it -> Integer.parseInt(it, 16));

toHex.execute(11/*0x11*/);// return 17;

您可以在 java-8 中简单地实现您的管道模式如下:

interface Step<I, O> {

    O execute(I value);

    default <R> Step<I, R> pipe(Step<O, R> source) {
        return value -> source.execute(execute(value));
    }

    static <I, O> Step<I, O> of(Step<I, O> source) {
        return source;
    }
}

在以前的 java 版本中，您可以使用抽象类:

abstract static class Step<I, O> {

    public abstract O execute(I value);

    public <R> Step<I, R> pipe(Step<O, R> source) {
        return new Step<I, R>() {
            @Override
            public R execute(I value) {
                return source.execute(Step.this.execute(value));
            }
        };
    }

    public static <I, O> Step<I, O> of(Step<I, O> source) {
        return source;
    }
}