C++之管道(Pipeline)模式_c++ pipeline模式

0. 简介

管道模式，不属于23种设计模式之一(是责任链模式的一种变体)，但是在我们实际业务架构中还是有很多场景适用的。尤其是在多传感器的数据流处理中会非常方便。尤其是在大型场景中使用Pipeline将复杂的进程分解成多个子任务。

1. 管道模式示意图

Pipeline模式为管道模式，也称为流水线模式。顾名思义，管道模式就像一条管道把多个对象连接起来，整体看起来就像若干个阀门嵌套在管道中，而处理逻辑就放在阀门上，需要处理的对象进入管道后，分别经过各个阀门，每个阀门都会对进入的对象进行一些逻辑处理，经过一层层的处理后从管道尾出来，此时的对象就是已完成处理的目标对象。

从图中可以看出，整个流水线内数据流转是从上游到下游，上游的输出是下游的输入，按阶段依次执行。

• Source: 表示数据来源，比如：KafkaSource。

• Channel：表示对数据进行处理的组件，比如：JsonChannel，对数据进行json转换和处理。

• Sink：表示数据落地或下沉的地方，比如：KafkaSink，表示数据发送到指定的kafka；DbSInk表示数据落地到DB。

可以看出，Pipeline是由Source（必须有），Channel（不一定需要），Sink（必须有）三种类型的组件自由组合而成的。

2. 示例程序

这里我们借鉴网上的程序，并进行简化

/** 申明整个生命周期 **/
class LifeCycle {
 public:
 	// 纯虚函数
    virtual void init(std::string config) = 0;
	
    virtual void startUp() = 0;

    virtual void shutDown() = 0;
};

/** 模板类，一般对模板参数类型typename和class认为是一样的，并构建多两个纯虚函数
https://blog.csdn.net/function_dou/article/details/84644963 **/
template<typename T>
class Component : public LifeCycle {
 public:
    virtual std::string getName() = 0;

    virtual void execute(T t) = 0;
};

/** 需要把之前的模板一同继承 **/
template<typename T, typename R>
class AbstractComponent : public Component<T> {
 private:
    std::unordered_set<shared_ptr<Component<R>>> down_stream;
 protected:
    const std::unordered_set<shared_ptr<Component<R>>> &getDownStream() {
        return down_stream;
    }

 protected:
    virtual R doExecute(T t) = 0;

 public:
    void addDownStream(shared_ptr<Component<R>> component) {
        down_stream.insert(component);//将函数指针存入数组中
    }

    void init(std::string config) override {
    }
    
    void startUp() override {
        for (auto &&obj : this->getDownStream()) {
            obj->startUp();
        }
        cout << "------------------ " + this->getName() + " is starting ----------------------" << endl;
    }

    void shutDown() override {
        auto downStreams = this->getDownStream();
        for (auto &&obj : downStreams) {
            obj->shutDown();
        }
        cout << "------------------ " + this->getName() + " is starting ----------------------" << endl;
    }

    void execute(T t) override {
        R r = doExecute(t);
        cout << this->getName() + "\treceive\t" << typeid(t).name() << "\t" << t << "\treturn\t" << typeid(r).name()
             << "\t" << r << endl;
        if constexpr (is_same_v<R, void>) {
            return;
        }
        for (auto &&obj : getDownStream()) {//读取数据
            obj->execute(r);//一层层查找
        }
    }
};

/**从这向下才是pipeline的操作，上面的是数据的生命周期**/

template<typename T, typename R>
using Source = AbstractComponent<T, R>;//using来重命名，使用typedef定义的别名和使用using定义的别名在语义上是等效的。 唯一的区别是typedef在模板中有一定的局限性，而using没有。https://blog.csdn.net/qq_35789421/article/details/117591212                                                                                                           

template<typename T, typename R>
using Channel = AbstractComponent<T, R>;

template<typename T, typename R>
using Sink = AbstractComponent<T, R>;

class printSink;// 申明class，以防找不到class申明，如果写在两个文件中，会采用前置声明的方式声明class，这里并没有什么用。https://www.shuzhiduo.com/A/obzb7pZVJE/

class intStringChannel;

class printSink : public Sink<string, int> {
 public:
    string getName() override {
        return "printSink";
    }

 protected:
    int doExecute(string t) override {
        return INT_MIN;
    }
};

class intStringChannel : public Channel<int, string> {
 public:
    string getName() override {
        return "intStringChannel";
    }

    void startUp() override {

    }

 protected:
    string doExecute(int t) override {
        return to_string(t + 100);
    }
};

class IntSource : public Source<int, int> {
 private:
    int val = 0;
 public:
    void init(std::string config) override {
        cout << "--------- " + getName() + " init --------- ";
        val = 1;
    }

    string getName() override {
        return "Int Source";
    }

    void startUp() override {
        this->execute(val);//处理数据
    }

 protected:
    int doExecute(int) override {
        return val + 1;
    }
};

template<typename R, typename T>
class pipeline : public LifeCycle {
 private:
    shared_ptr<Source<R, T>> source;

 public:
    void setSource(shared_ptr<Source<R, T>> component) {
        source = component;//传入类
    }

    void init(std::string config) override {
    }

    void startUp() override {
        assert(source.get() != nullptr);
        source->startUp();//完成类函数的处理
    }

    void shutDown() override {
        source->shutDown();
    }
};

int main() {
    pipeline<int, int> p;
    // source
    auto is = make_shared<IntSource>();

    // channel
    auto isc = make_shared<intStringChannel>();

    // sink
    auto ps = make_shared<printSink>();

    is->addDownStream(isc);
    isc->addDownStream(ps);

    // 设置 source
    p.setSource(is);

    // 启动
    p.startUp();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179

3. bind 回调函数

…详情请参照古月居