C++流水线的简易实现_c++ pipeline 例子

流水线本来是工程任务中为了将不同的工序分开的一种处理流程。这种分工，会给工作的流程化和吞吐带来一定的好处。
如果将流水线应用到程序中，会有如下一些好处：
1、不需要锁。线程之间有时序关系，因此从流程设计上就规避了锁的机制；
2、多线程并发。不同线程负责不同的事务处理，可以在一定程度上实现并发；
3、吞吐量高。由于多线程不会像单线程那样阻塞，所以吞吐量会有一定提高。
但流水线也有它的缺点：
1、处理有时序。这就要求对整个处理流程能够正确划分，不要出现处理不均衡的情况；
2、由于无锁的设计，对开发要求比较高。如果用法不当，容易出现多线程的死锁。

总体来说，流水线不管是对性能还是开发人员来说，都是很好的一种方法，能够让程序像人一样思考，分段处理。
下面，实现一个超级简易的样例，共同学习。

#ifndef PIPE_MY_PIPELINE
#define PIPE_MY_PIPELINE

#include <iostream>
#include <vector>
#include <map>
#include "my_message.h"

/**
消息驱动 
**/ 
class IPipeBase
{
	public:
	virtual void process(MyMessage* msg) = 0;
};

class PipeBase : public IPipeBase
{
public:
virtual void forward(MyMessage* msg) 
{
	for (int i = 0; i < next_pipes_.size(); ++i)
	{
		// 将msg丢给所有的下一级流水线处理
		next_pipes_[i]->process(msg); 
	}
}

void insert_next(IPipeBase* pipe)
{
	next_pipes_.push_back(pipe);
}

protected:
std::vector<IPipeBase*> next_pipes_;
};
	
class MyPipeline
{
public:

bool insert_pipeline_seq(PipeBase* source, PipeBase* dest)
{

	source->insert_next(dest);	
}

};

#endif
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

主程序测试样例：
#include
#include “my_pipeline.h”
#include “test_mypipeline.h”
/* run this program using the console pauser or add your own getch, system(“pause”) or input loop */

int main(int argc, char** argv) {
	PipeStart* start = new PipeStart();
	PipeEnd1* end1 = new PipeEnd1();
	PipeEnd2* end2 = new PipeEnd2();
	Pipe1* pipe1 = new Pipe1();
	Pipe2* pipe2 = new Pipe2(); 
	Pipe3* pipe3 = new Pipe3(); 
	Pipe4* pipe4 = new Pipe4(); 
	Pipe5* pipe5 = new Pipe5(); 
	
	MyPipeline* mypipeline = new MyPipeline();
	mypipeline->insert_pipeline_seq(start, pipe1); 
	mypipeline->insert_pipeline_seq(start, pipe2); 
	
	mypipeline->insert_pipeline_seq(pipe1, pipe3); 
	mypipeline->insert_pipeline_seq(pipe1, pipe4); 
	
	mypipeline->insert_pipeline_seq(pipe2, pipe5);
	 
	mypipeline->insert_pipeline_seq(pipe4, pipe5); 
	
	mypipeline->insert_pipeline_seq(pipe3, end1); 
	mypipeline->insert_pipeline_seq(pipe4, end1); 
	
	mypipeline->insert_pipeline_seq(pipe5, end2); 
	
	MyMessage* msg = new MyMessage();
	msg->set_value(0);
	start->process(msg);
	
	return 0;
}
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

该样例的流水线设计如下：

S - 1 - 3 - E1
| | |
2 4 -------->
| |
—>5 ---->E2