释放双眼，带上耳机，听听看~！
1.static 函数使用原因是因为pthread_create中需要使用的是全局函数，声明为静态函数。bufferevent中的回调函数同理。通过传递参数在其中访问非静态成员
2.下列代码中在TcpEventServer::ReadEventCb中为conn中的m_ReadBuf、m_WriteBuf赋值是错误的。若这样操作在新建连接的回调函数也就是ConnectionEvent中调用conn中的读写bufferevent数据将会出错。应该在TcpEventServer::ThreadProcess(应该叫做notifyPipeHandler更恰当一点)中新创建conn之后立马为其赋值。
//将该链接放入队列
Conn *conn = me->connectQueue.InsertConn(confd, me);
conn->m_ReadBuf = bufferevent_get_input(bev);
conn->m_WriteBuf = bufferevent_get_output(bev);
3.可以使用c++11新特性std::function 和std::bind将虚函数注册到server中，不需要进行继承操作
- 按照此方法修改后的代码：https://github.com/kevin-gjm/dragonfly.git
最近在看memcached的源码，觉得它那种libevent+多线程的服务器模型真的很不错，我将这个模型封装成一个C++类，根据我的简单测试，这个模型的效率真的很不错，欢迎大家试用。
这个类的使用方法很简单（缺点是不太灵活），只要派生一个类，根据需要重写以下这几个虚函数就行了：
2
3
4
5
6
7
8
9
10
11
12
2virtual void ConnectionEvent(Conn *conn) { }
3//读取完数据后，会调用该函数
4virtual void ReadEvent(Conn *conn) { }
5//发送完成功后，会调用该函数（因为串包的问题，所以并不是每次发送完数据都会被调用）
6virtual void WriteEvent(Conn *conn) { }
7//断开连接（客户自动断开或异常断开）后，会调用该函数
8virtual void CloseEvent(Conn *conn, short events) { }
9//发生致命错误（如果创建子线程失败等）后，会调用该函数
10//该函数的默认操作是输出错误提示，终止程序
11virtual void ErrorQuit(const char *str);
12
2
2
上代码：
头文件：TcpEventServer.h
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
180
181
182
183
184
185
186
2#ifndef TCPEVENTSERVER_H_
3#define TCPEVENTSERVER_H_
4
5#include <stdio.h>
6#include <stdlib.h>
7#include <unistd.h>
8#include <string.h>
9#include <errno.h>
10#include <signal.h>
11#include <time.h>
12#include <pthread.h>
13#include <fcntl.h>
14
15#include <map>
16using std::map;
17
18#include <event.h>
19#include <event2/bufferevent.h>
20#include <event2/buffer.h>
21#include <event2/listener.h>
22#include <event2/util.h>
23#include <event2/event.h>
24
25class TcpEventServer;
26class Conn;
27class ConnQueue;
28struct LibeventThread;
29
30//这个类一个链表的结点类，结点里存储各个连接的信息，
31//并提供了读写数据的接口
32class Conn
33{
34 //此类只能由TcpBaseServer创建，
35 //并由ConnQueue类管理
36 friend class ConnQueue;
37 friend class TcpEventServer;
38
39private:
40 const int m_fd; //socket的ID
41 evbuffer *m_ReadBuf; //读数据的缓冲区
42 evbuffer *m_WriteBuf; //写数据的缓冲区
43
44 Conn *m_Prev; //前一个结点的指针
45 Conn *m_Next; //后一个结点的指针
46 LibeventThread *m_Thread;
47
48 Conn(int fd=0);
49 ~Conn();
50
51public:
52 LibeventThread *GetThread() { return m_Thread; }
53 int GetFd() { return m_fd; }
54
55 //获取可读数据的长度
56 int GetReadBufferLen()
57 { return evbuffer_get_length(m_ReadBuf); }
58
59 //从读缓冲区中取出len个字节的数据，存入buffer中，若不够，则读出所有数据
60 //返回读出数据的字节数
61 int GetReadBuffer(char *buffer, int len)
62 { return evbuffer_remove(m_ReadBuf, buffer, len); }
63
64 //从读缓冲区中复制出len个字节的数据，存入buffer中，若不够，则复制出所有数据
65 //返回复制出数据的字节数
66 //执行该操作后，数据还会留在缓冲区中，buffer中的数据只是原数据的副本
67 int CopyReadBuffer(char *buffer, int len)
68 { return evbuffer_copyout(m_ReadBuf, buffer, len); }
69
70 //获取可写数据的长度
71 int GetWriteBufferLen()
72 { return evbuffer_get_length(m_WriteBuf); }
73
74 //将数据加入写缓冲区，准备发送
75 int AddToWriteBuffer(char *buffer, int len)
76 { return evbuffer_add(m_WriteBuf, buffer, len); }
77
78 //将读缓冲区中的数据移动到写缓冲区
79 void MoveBufferData()
80 { evbuffer_add_buffer(m_WriteBuf, m_ReadBuf); }
81
82};
83
84//带头尾结点的双链表类，每个结点存储一个连接的数据
85class ConnQueue
86{
87private:
88 Conn *m_head;
89 Conn *m_tail;
90public:
91 ConnQueue();
92 ~ConnQueue();
93 Conn *InsertConn(int fd, LibeventThread *t);
94 void DeleteConn(Conn *c);
95 //void PrintQueue();
96};
97
98//每个子线程的线程信息
99struct LibeventThread
100{
101 pthread_t tid; //线程的ID
102 struct event_base *base; //libevent的事件处理机
103 struct event notifyEvent; //监听管理的事件机
104 int notifyReceiveFd; //管理的接收端
105 int notifySendFd; //管道的发送端
106 ConnQueue connectQueue; //socket连接的链表
107
108 //在libevent的事件处理中要用到很多回调函数，不能使用类隐含的this指针
109 //所以用这样方式将TcpBaseServer的类指针传过去
110 TcpEventServer *tcpConnect; //TcpBaseServer类的指针
111};
112
113class TcpEventServer
114{
115private:
116 int m_ThreadCount; //子线程数
117 int m_Port; //监听的端口
118 LibeventThread *m_MainBase; //主线程的libevent事件处理机
119 LibeventThread *m_Threads; //存储各个子线程信息的数组
120 map<int, event*> m_SignalEvents; //自定义的信号处理
121
122public:
123 static const int EXIT_CODE = -1;
124
125private:
126 //初始化子线程的数据
127 void SetupThread(LibeventThread *thread);
128
129 //子线程的入门函数
130 static void *WorkerLibevent(void *arg);
131 //（主线程收到请求后），对应子线程的处理函数
132 static void ThreadProcess(int fd, short which, void *arg);
133 //被libevent回调的各个静态函数
134 static void ListenerEventCb(evconnlistener *listener, evutil_socket_t fd,
135 sockaddr *sa, int socklen, void *user_data);
136 static void ReadEventCb(struct bufferevent *bev, void *data);
137 static void WriteEventCb(struct bufferevent *bev, void *data);
138 static void CloseEventCb(struct bufferevent *bev, short events, void *data);
139
140protected:
141 //这五个虚函数，一般是要被子类继承，并在其中处理具体业务的
142
143 //新建连接成功后，会调用该函数
144 virtual void ConnectionEvent(Conn *conn) { }
145
146 //读取完数据后，会调用该函数
147 virtual void ReadEvent(Conn *conn) { }
148
149 //发送完成功后，会调用该函数（因为串包的问题，所以并不是每次发送完数据都会被调用）
150 virtual void WriteEvent(Conn *conn) { }
151
152 //断开连接（客户自动断开或异常断开）后，会调用该函数
153 virtual void CloseEvent(Conn *conn, short events) { }
154
155 //发生致命错误（如果创建子线程失败等）后，会调用该函数
156 //该函数的默认操作是输出错误提示，终止程序
157 virtual void ErrorQuit(const char *str);
158
159public:
160 TcpEventServer(int count);
161 ~TcpEventServer();
162
163 //设置监听的端口号，如果不需要监听，请将其设置为EXIT_CODE
164 void SetPort(int port)
165 { m_Port = port; }
166
167 //开始事件循环
168 bool StartRun();
169 //在tv时间里结束事件循环
170 //否tv为空，则立即停止
171 void StopRun(timeval *tv);
172
173 //添加和删除信号处理事件
174 //sig是信号，ptr为要回调的函数
175 bool AddSignalEvent(int sig, void (*ptr)(int, short, void*));
176 bool DeleteSignalEvent(int sig);
177
178 //添加和删除定时事件
179 //ptr为要回调的函数，tv是间隔时间，once决定是否只执行一次
180 event *AddTimerEvent(void(*ptr)(int, short, void*),
181 timeval tv, bool once);
182 bool DeleteTImerEvent(event *ev);
183};
184
185#endif
186
实现文件：TcpEventServer.cpp
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
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
2#include "TcpEventServer.h"
3
4Conn::Conn(int fd) : m_fd(fd)
5{
6 m_Prev = NULL;
7 m_Next = NULL;
8}
9
10Conn::~Conn()
11{
12
13}
14
15ConnQueue::ConnQueue()
16{
17 //建立头尾结点，并调整其指针
18 m_head = new Conn(0);
19 m_tail = new Conn(0);
20 m_head->m_Prev = m_tail->m_Next = NULL;
21 m_head->m_Next = m_tail;
22 m_tail->m_Prev = m_head;
23}
24
25ConnQueue::~ConnQueue()
26{
27 Conn *tcur, *tnext;
28 tcur = m_head;
29 //循环删除链表中的各个结点
30 while( tcur != NULL )
31 {
32 tnext = tcur->m_Next;
33 delete tcur;
34 tcur = tnext;
35 }
36}
37
38Conn *ConnQueue::InsertConn(int fd, LibeventThread *t)
39{
40 Conn *c = new Conn(fd);
41 c->m_Thread = t;
42 Conn *next = m_head->m_Next;
43
44 c->m_Prev = m_head;
45 c->m_Next = m_head->m_Next;
46 m_head->m_Next = c;
47 next->m_Prev = c;
48 return c;
49}
50
51void ConnQueue::DeleteConn(Conn *c)
52{
53 c->m_Prev->m_Next = c->m_Next;
54 c->m_Next->m_Prev = c->m_Prev;
55 delete c;
56}
57
58/*
59void ConnQueue::PrintQueue()
60{
61 Conn *cur = m_head->m_Next;
62 while( cur->m_Next != NULL )
63 {
64 printf("%d ", cur->m_fd);
65 cur = cur->m_Next;
66 }
67 printf("\n");
68}
69*/
70
71TcpEventServer::TcpEventServer(int count)
72{
73 //初始化各项数据
74 m_ThreadCount = count;
75 m_Port = -1;
76 m_MainBase = new LibeventThread;
77 m_Threads = new LibeventThread[m_ThreadCount];
78 m_MainBase->tid = pthread_self();
79 m_MainBase->base = event_base_new();
80
81 //初始化各个子线程的结构体
82 for(int i=0; i<m_ThreadCount; i++)
83 {
84 SetupThread(&m_Threads[i]);
85 }
86
87}
88
89TcpEventServer::~TcpEventServer()
90{
91 //停止事件循环（如果事件循环没开始，则没效果）
92 StopRun(NULL);
93
94 //释放内存
95 event_base_free(m_MainBase->base);
96 for(int i=0; i<m_ThreadCount; i++)
97 event_base_free(m_Threads[i].base);
98
99 delete m_MainBase;
100 delete [] m_Threads;
101}
102
103void TcpEventServer::ErrorQuit(const char *str)
104{
105 //输出错误信息，退出程序
106 fprintf(stderr, "%s", str);
107 if( errno != 0 )
108 fprintf(stderr, " : %s", strerror(errno));
109 fprintf(stderr, "\n");
110 exit(1);
111}
112
113void TcpEventServer::SetupThread(LibeventThread *me)
114{
115 //建立libevent事件处理机制
116 me->tcpConnect = this;
117 me->base = event_base_new();
118 if( NULL == me->base )
119 ErrorQuit("event base new error");
120
121 //在主线程和子线程之间建立管道
122 int fds[2];
123 if( pipe(fds) )
124 ErrorQuit("create pipe error");
125 me->notifyReceiveFd = fds[0];
126 me->notifySendFd = fds[1];
127
128 //让子线程的状态机监听管道
129 event_set( &me->notifyEvent, me->notifyReceiveFd,
130 EV_READ | EV_PERSIST, ThreadProcess, me );
131 event_base_set(me->base, &me->notifyEvent);
132 if ( event_add(&me->notifyEvent, 0) == -1 )
133 ErrorQuit("Can't monitor libevent notify pipe\n");
134}
135
136void *TcpEventServer::WorkerLibevent(void *arg)
137{
138 //开启libevent的事件循环，准备处理业务
139 LibeventThread *me = (LibeventThread*)arg;
140 //printf("thread %u started\n", (unsigned int)me->tid);
141 event_base_dispatch(me->base);
142 //printf("subthread done\n");
143}
144
145bool TcpEventServer::StartRun()
146{
147 evconnlistener *listener;
148
149 //如果端口号不是EXIT_CODE，就监听该端口号
150 if( m_Port != EXIT_CODE )
151 {
152 sockaddr_in sin;
153 memset(&sin, 0, sizeof(sin));
154 sin.sin_family = AF_INET;
155 sin.sin_port = htons(m_Port);
156 listener = evconnlistener_new_bind(m_MainBase->base,
157 ListenerEventCb, (void*)this,
158 LEV_OPT_REUSEABLE|LEV_OPT_CLOSE_ON_FREE, -1,
159 (sockaddr*)&sin, sizeof(sockaddr_in));
160 if( NULL == listener )
161 ErrorQuit("TCP listen error");
162 }
163
164 //开启各个子线程
165 for(int i=0; i<m_ThreadCount; i++)
166 {
167 pthread_create(&m_Threads[i].tid, NULL,
168 WorkerLibevent, (void*)&m_Threads[i]);
169 }
170
171 //开启主线程的事件循环
172 event_base_dispatch(m_MainBase->base);
173
174 //事件循环结果，释放监听者的内存
175 if( m_Port != EXIT_CODE )
176 {
177 //printf("free listen\n");
178 evconnlistener_free(listener);
179 }
180}
181
182void TcpEventServer::StopRun(timeval *tv)
183{
184 int contant = EXIT_CODE;
185 //向各个子线程的管理中写入EXIT_CODE，通知它们退出
186 for(int i=0; i<m_ThreadCount; i++)
187 {
188 write(m_Threads[i].notifySendFd, &contant, sizeof(int));
189 }
190 //结果主线程的事件循环
191 event_base_loopexit(m_MainBase->base, tv);
192}
193
194void TcpEventServer::ListenerEventCb(struct evconnlistener *listener,
195 evutil_socket_t fd,
196 struct sockaddr *sa,
197 int socklen,
198 void *user_data)
199{
200 TcpEventServer *server = (TcpEventServer*)user_data;
201
202 //随机选择一个子线程，通过管道向其传递socket描述符
203 int num = rand() % server->m_ThreadCount;
204 int sendfd = server->m_Threads[num].notifySendFd;
205 write(sendfd, &fd, sizeof(evutil_socket_t));
206}
207
208void TcpEventServer::ThreadProcess(int fd, short which, void *arg)
209{
210 LibeventThread *me = (LibeventThread*)arg;
211
212 //从管道中读取数据（socket的描述符或操作码）
213 int pipefd = me->notifyReceiveFd;
214 evutil_socket_t confd;
215 read(pipefd, &confd, sizeof(evutil_socket_t));
216
217 //如果操作码是EXIT_CODE，则终于事件循环
218 if( EXIT_CODE == confd )
219 {
220 event_base_loopbreak(me->base);
221 return;
222 }
223
224 //新建连接
225 struct bufferevent *bev;
226 bev = bufferevent_socket_new(me->base, confd, BEV_OPT_CLOSE_ON_FREE);
227 if (!bev)
228 {
229 fprintf(stderr, "Error constructing bufferevent!");
230 event_base_loopbreak(me->base);
231 return;
232 }
233
234 //将该链接放入队列
235 Conn *conn = me->connectQueue.InsertConn(confd, me);
236
237
238
2
3
2 conn->m_WriteBuf = bufferevent_get_output(bev);
3
2
2
//conn->m_ReadBuf = bufferevent_get_input(bev);//conn->m_WriteBuf = bufferevent_get_output(bev);//调用用户自定义的读取事件处理函数conn->m_Thread->tcpConnect->ReadEvent(conn);} void TcpEventServer::WriteEventCb(struct bufferevent *bev, void *data){Conn *conn = (Conn*)data;conn->m_ReadBuf = bufferevent_get_input(bev);conn->m_WriteBuf = bufferevent_get_output(bev);//调用用户自定义的写入事件处理函数conn->m_Thread->tcpConnect->WriteEvent(conn);}void TcpEventServer::CloseEventCb(struct bufferevent *bev, short events, void *data){Conn *conn = (Conn*)data;//调用用户自定义的断开事件处理函数conn->m_Thread->tcpConnect->CloseEvent(conn, events);conn->GetThread()->connectQueue.DeleteConn(conn);bufferevent_free(bev);}bool TcpEventServer::AddSignalEvent(int sig, void (*ptr)(int, short, void*)){//新建一个信号事件event *ev = evsignal_new(m_MainBase->base, sig, ptr, (void*)this);if ( !ev || event_add(ev, NULL) < 0 ){event_del(ev);return false;}//删除旧的信号事件（同一个信号只能有一个信号事件）DeleteSignalEvent(sig);m_SignalEvents[sig] = ev;return true;}bool TcpEventServer::DeleteSignalEvent(int sig){map<int, event*>::iterator iter = m_SignalEvents.find(sig);if( iter == m_SignalEvents.end() )return false;event_del(iter->second);m_SignalEvents.erase(iter);return true;}event *TcpEventServer::AddTimerEvent(void (*ptr)(int, short, void *), timeval tv, bool once){int flag = 0;if( !once )flag = EV_PERSIST;//新建定时器信号事件event *ev = new event;event_assign(ev, m_MainBase->base, -1, flag, ptr, (void*)this);if( event_add(ev, &tv) < 0 ){event_del(ev);return NULL;}return ev;}bool TcpEventServer::DeleteTImerEvent(event *ev){int res = event_del(ev);return (0 == res);}
测试文件：test.cpp
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
2这是一个测试用的服务器，只有两个功能：
31：对于每个已连接客户端，每10秒向其发送一句hello, world
42：若客户端向服务器发送数据，服务器收到后，再将数据回发给客户端
5*/
6//test.cpp
7#include "TcpEventServer.h"
8#include <set>
9#include <vector>
10using namespace std;
11
12//测试示例
13class TestServer : public TcpEventServer
14{
15private:
16 vector<Conn*> vec;
17protected:
18 //重载各个处理业务的虚函数
19 void ReadEvent(Conn *conn);
20 void WriteEvent(Conn *conn);
21 void ConnectionEvent(Conn *conn);
22 void CloseEvent(Conn *conn, short events);
23public:
24 TestServer(int count) : TcpEventServer(count) { }
25 ~TestServer() { }
26
27 //退出事件，响应Ctrl+C
28 static void QuitCb(int sig, short events, void *data);
29 //定时器事件，每10秒向所有客户端发一句hello, world
30 static void TimeOutCb(int id, int short events, void *data);
31};
32
33void TestServer::ReadEvent(Conn *conn)
34{
35 conn->MoveBufferData();
36}
37
38void TestServer::WriteEvent(Conn *conn)
39{
40
41}
42
43void TestServer::ConnectionEvent(Conn *conn)
44{
45 TestServer *me = (TestServer*)conn->GetThread()->tcpConnect;
46 printf("new connection: %d\n", conn->GetFd());
47 me->vec.push_back(conn);
48}
49
50void TestServer::CloseEvent(Conn *conn, short events)
51{
52 printf("connection closed: %d\n", conn->GetFd());
53}
54
55void TestServer::QuitCb(int sig, short events, void *data)
56{
57 printf("Catch the SIGINT signal, quit in one second\n");
58 TestServer *me = (TestServer*)data;
59 timeval tv = {1, 0};
60 me->StopRun(&tv);
61}
62
63void TestServer::TimeOutCb(int id, short events, void *data)
64{
65 TestServer *me = (TestServer*)data;
66 char temp[33] = "hello, world\n";
67 for(int i=0; i<me->vec.size(); i++)
68 me->vec[i]->AddToWriteBuffer(temp, strlen(temp));
69}
70
71int main()
72{
73 printf("pid: %d\n", getpid());
74 TestServer server(3);
75 server.AddSignalEvent(SIGINT, TestServer::QuitCb);
76 timeval tv = {10, 0};
77 server.AddTimerEvent(TestServer::TimeOutCb, tv, false);
78 server.SetPort(2111);
79 server.StartRun();
80 printf("done\n");
81
82 return 0;
83}
84
编译与运行命令：
2
3
4
5
6
7
8
2qch@LinuxMint ~/program/ztemp $ ./test
3pid: 20264
4new connection: 22
5connection closed: 22
6^CCatch the SIGINT signal, quit in one second
7done
8
