转载地址:/jiange_zh/article/details/50636536
在这一部分，我们将对上一篇中的master-worker进行拓展，成为一个简单的echo服务器。
这一步我们需要添加两个类：Listener和Connection；
Listener的职责:
1.创建监听套接字;
2.注册监听套接字事件;
3.在监听事件的回调函数中进行accept并创建新连接;
其头文件如下:
/*************************************************************************> File Name: listener.h> Author: Jiange> Mail: jiangezh@ > Created Time: 01月27日 星期三 19时46分34秒************************************************************************/#ifndef _LISTENER_H#define _LISTENER_H#include <string>#include "event2/event.h"#include "event2/util.h"#include "util.h"class Worker;class Listener{public:Listener(const std::string &ip, unsigned short port);~Listener();bool InitListener(Worker *worker);void AddListenEvent();static void ListenEventCallback(evutil_socket_t fd, short event, void *arg);Worker *listen_worker;evutil_socket_t listen_sockfd;struct sockaddr_in listen_addr;struct event *listen_event;uint64_t cnt_connection;};#endif
接下来看看具体实现:
/*************************************************************************> File Name: listener.cpp> Author: Jiange> Mail: jiangezh@ > Created Time: 01月27日 星期三 19时48分56秒************************************************************************/#include "listener.h"#include "worker.h"#include "connection.h"#include <iostream>Listener::Listener(const std::string &ip, unsigned short port){//ipv4listen_addr.sin_family = AF_INET;listen_addr.sin_addr.s_addr = inet_addr(ip.c_str());listen_addr.sin_port = htons(port);listen_event = NULL;cnt_connection = 0;std::cout << "Init listener" << std::endl;}Listener::~Listener(){if (listen_event){event_free(listen_event);close(listen_sockfd);}std::cout<< "Listener closed" << std::endl;}bool Listener::InitListener(Worker *worker){if (-1 == (listen_sockfd = socket(AF_INET, SOCK_STREAM, 0))){return false;}//非阻塞evutil_make_socket_nonblocking(listen_sockfd);int reuse = 1;//重用setsockopt(listen_sockfd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));if (0 != bind(listen_sockfd, (struct sockaddr *)&listen_addr, sizeof(listen_addr))) {return false;}if (0 != listen(listen_sockfd, 5)) {return false;}listen_worker = worker;return true;}/* 这里单独作为一个函数，而不是合并到上面函数中。* 因为InitListener是在fork之前调用的，此时* worker的w_base还未赋值；*/void Listener::AddListenEvent(){//echo先从客户端读取数据，故此处监听读listen_event = event_new(listen_worker->w_base, listen_sockfd, EV_READ | EV_PERSIST, Listener::ListenEventCallback, this);event_add(listen_event, NULL);}void Listener::ListenEventCallback(evutil_socket_t sockfd, short event, void *arg){evutil_socket_t con_fd;struct sockaddr_in con_addr;socklen_t addr_len = sizeof(con_addr);if (-1 == (con_fd = accept(sockfd, (struct sockaddr *)&con_addr, &addr_len))) {return ;}Listener *listener = (Listener *)arg;Connection *con = new Connection();con->con_sockfd = con_fd;pid_t pid = getpid();std::cout << "listen accept: " << con->con_sockfd << " by process " << pid << std::endl;if (!con->InitConnection(listener->listen_worker)){Connection::FreeConnection(con);return ;}con->con_worker->con_map[con->con_sockfd] = con;++listener->cnt_connection;}
接下来看看Connection：
一个Connection实例即代表一个连接，它维护从listener的回调函数那里accept得到的套接字，并在该套接字上监听读写事件，进行request和response。
/*************************************************************************> File Name: connection.h> Author: Jiange> Mail: jiangezh@ > Created Time: 01月27日 星期三 20时10分35秒************************************************************************/#ifndef _CONNECTION_H#define _CONNECTION_H#include <string>#include <queue>#include "event2/event.h"#include "event2/util.h"#include "util.h"class Worker;class Connection{public:Connection();~Connection();bool InitConnection(Worker *worker);static void ConEventCallback(evutil_socket_t fd, short event, void *arg);Worker *con_worker;evutil_socket_t con_sockfd;struct event *read_event;struct event *write_event;std::string con_inbuf;std::string con_intmp;std::string con_outbuf;static void FreeConnection(Connection *con);private:void WantRead();void NotWantRead();void WantWrite();void NotWantWrite();};#endif
这里两个event分别负责读写事件，比一个event效率高一些。
在回调函数的设计中，本来打算使用两个回调函数：一个处理读，一个处理写。
不过其实可以合并到同一个回调函数里，所以还是在一个函数中处理，并增加4个函数，来进行监听事件的切换，这样做更有利于后面状态机的拓展：
void Connection::WantRead(){event_add(read_event, NULL); }void Connection::NotWantRead(){event_del(read_event);}void Connection::WantWrite(){event_add(write_event, NULL);}void Connection::NotWantWrite() {event_del(write_event);}
除以上函数外其他部分的具体实现：
/*************************************************************************> File Name: connection.cpp> Author: Jiange> Mail: jiangezh@ > Created Time: 01月28日 星期四 12时06分22秒************************************************************************/#include "connection.h"#include "worker.h"#include<iostream>Connection::Connection(){con_worker = NULL;read_event = NULL;write_event= NULL;}Connection::~Connection(){if (read_event && write_event){event_free(read_event);event_free(write_event);std::cout << con_sockfd << " closed" << std::endl;close(con_sockfd);}}/* 删除worker中相应的con，并释放该con */void Connection::FreeConnection(Connection *con){Worker *worker = con->con_worker;if (con->read_event && con->write_event){Worker::ConnectionMap::iterator con_iter = worker->con_map.find(con->con_sockfd);worker->con_map.erase(con_iter);}delete con;}bool Connection::InitConnection(Worker *worker){con_worker = worker;try{ //这里不能开太大，会爆内存！//后期可能需要在内存的使用上进行优化～con_intmp.reserve(10 * 1024);con_inbuf.reserve(10 * 1024);con_outbuf.reserve(10 * 1024);evutil_make_socket_nonblocking(con_sockfd);//test：监听读事件，从客户端读，然后回显read_event = event_new(con_worker->w_base, con_sockfd, EV_PERSIST | EV_READ, Connection::ConEventCallback, this);write_event = event_new(con_worker->w_base, con_sockfd, EV_PERSIST | EV_WRITE, Connection::ConEventCallback, this);}catch(std::bad_alloc){std::cout << "InitConnection():bad_alloc" <<std::endl;}WantRead();return true;}/* 循环读写* 注意，在读的时候，此处ret为0时，可能是空字符串之类的* 所以在这里暂不做处理*/void Connection::ConEventCallback(evutil_socket_t sockfd, short event, void *arg){Connection *con = (Connection*)arg;if (event & EV_READ) {int cap = con->con_intmp.capacity();int ret = read(sockfd, &con->con_intmp[0], cap);if (ret == -1){if (errno != EAGAIN && errno != EINTR){FreeConnection(con);return;}}else if (ret == 0){FreeConnection(con); return;}else{con->con_inbuf.clear();con->con_inbuf.append(con->con_intmp.c_str(), ret);}con->con_outbuf = con->con_inbuf;con->NotWantRead();con->WantWrite();}if (event & EV_WRITE){int ret = write(sockfd, con->con_outbuf.c_str(), con->con_outbuf.size());if (ret == -1){if (errno != EAGAIN && errno != EINTR){FreeConnection(con);return;}}con->NotWantWrite();con->WantRead();} }
介绍完listener和connection之后，我们需要相应地调整master和worker的代码：
1.修改两者的构造函数：
Master::Master(const std::string &ip, unsigned short port):worker(ip, port){//……}Worker::Worker(const std::string &ip, unsigned short port):listener(ip, port){//……}
2.在master中开始创建监听套接字：
bool Master::StartMaster(){std::cout << "Start Master" << std::endl;if (!worker.listener.InitListener(&worker)){return false;}//……}
3.在worker中增加listener和connection map成员：
class Master;class Connection;class Worker{public:typedef std::map<evutil_socket_t, Connection*> ConnectionMap;//……Listener listener;ConnectionMap con_map;};
4.worker析构函数释放持有的连接：
Worker::~Worker(){//……if (w_base){ConnectionMap::iterator con_iter = con_map.begin();while (con_iter != con_map.end()){Connection *con = con_iter->second;delete con;++con_iter;}event_base_free(w_base);}}
5.worker增加监听event事件：
void Worker::Run(){w_base = event_base_new();listener.AddListenEvent();//……return;}
6.修改main函数中的master构造；
最后，附上一个使用到的头文件：
/*************************************************************************> File Name: util.h> Author: Jiange> Mail: jiangezh@ > Created Time: 01月28日 星期四 10时39分22秒************************************************************************/#ifndef _UTIL_H#define _UTIL_H#include <signal.h>#include <unistd.h>#include <sys/socket.h>#include <netinet/in.h>#include <arpa/inet.h>#include <errno.h>#include <stdint.h>#endif
至此，我们简单的echo服务器就大功告成啦～
下面给出makefile：
THIRD_LIBS=-leventLIBS=-ldlCFLAGS=-I./includemaster:src/master.o src/worker.o src/listener.o src/connection.o src/main.og++ -g -o $@ src/master.o src/worker.o src/listener.o src/connection.o src/main.o $(THIRD_LIBS) $(LIBS)src/master.o:src/master.cpp include/master.hg++ -g -o $@ -c $< $(CFLAGS)src/worker.o:src/worker.cpp include/worker.h include/util.hg++ -g -o $@ -c $< $(CFLAGS)src/listener.o:src/listener.cpp include/listener.h include/util.hg++ -g -o $@ -c $< $(CFLAGS)src/connection.o:src/connection.cpp include/connection.h include/util.hg++ -g -o $@ -c $< $(CFLAGS)src/main.o:src/main.cpp include/master.hg++ -g -o $@ -c $< $(CFLAGS)clean:rm -f src/*.o master
我用python写了个用于测试的客户端：
#python2.7.6#coding=utf-8import socketif __name__ == "__main__":sockfd = socket.socket(socket.AF_INET, socket.SOCK_STREAM)sockfd.connect(('localhost', 8000))message = ""while 1:message = raw_input("Please input:")sockfd.send(message)message = sockfd.recv(8000)print messagesockfd.close()
另外一个用于测试的：
#python2.7.6#coding=utf-8import socketimport timeimport threadinghttp_request = "POST /test_server HTTP/1.1\r\nHost:test.py\r\nContent-Length:5\r\n\r\nHello"def make_a_request():sockfd = socket.socket(socket.AF_INET, socket.SOCK_STREAM)sockfd.connect(('localhost', 8000))sockfd.sendall(http_request)sockfd.recv(8000)sockfd.close()if __name__ == "__main__":thread_list = []start_time = time.time()for i in range(0, 1000):thread = threading.Thread(target = make_a_request)thread_list.append(thread)thread.start()for thread in thread_list:thread.join()print "Time used for 1000 request: ", time.time() - start_time
另外，由于是多进程，所以需要测试一下并发下的运转情况～
（可以使用webbench～）
本程序在github的源代码
接下来，我们将引入状态机机制，并开始进行一些http的请求与处理！
如果觉得《【slighttpd】基于lighttpd架构的Server项目实战(4)—简单的echo服务器》对你有帮助，请点赞、收藏，并留下你的观点哦！
