[关闭]
@lishuhuakai 2016-11-04T07:41:33.000000Z 字数 5537 阅读 2076

一起来写web server 08 -- 多线程+非阻塞IO+epoll


到了多线程,一些东西就变得耐人寻味了.

这个版本是在前面单线程epoll的基础上引入了线程池,当然不是前面玩具一样的线程池,而是一个通用的组件,生产者消费者队列.

生产者消费者队列

生产者消费者问题是操作系统中一个很经典的同步互斥问题,已经有了很不错的解决方案,将它的解决方案拓展一下,就可以用于我们的实践啦.

我自己写了一个生产者消费者的队列,然后发现muduo中已经内置了这种模型,而且使用起来比我写的更加顺手,所以我就引用它的实现,我这里稍微来讲解一下它的实现,然后我会顺带讲解一下我的思路.

muduo库的生产者消费者模型

这是ThreadPool类的一个声明:

  1. class ThreadPool : noncopyable
  2. {
  3. public:
  4. typedef boost::function<void()> Task; /* 需要执行的任务 */
  5. private:
  6. bool isFull();
  7. Task take();
  8. size_t queueSize();
  9. int threadNum_; /* 线程的数目 */
  10. int maxQueueSize_;
  11. std::list<Task> queue_; /* 工作队列 */
  12. MutexLock mutex_;
  13. Condition notEmpty_;
  14. Condition notFull_;
  15. };

这里的boost::function其实在cpp 11标准中已经加入了,你如果没有安装boost库的话,可以缓存std的版本,效果是一样的.因为boost本来就是cppstd的一个备选库.

为什么要使用boost::function不用我多说,你可以查看这里:http://blog.csdn.net/solstice/article/details/3066268

我们来看一下代码的实现,首先是构造函数:

  1. ThreadPool::ThreadPool(int threadNum, int maxQueueSize)
  2. : threadNum_(threadNum)
  3. , maxQueueSize_(maxQueueSize)
  4. , mutex_()
  5. , notEmpty_(mutex_)
  6. , notFull_(mutex_)
  7. {
  8. assert(threadNum >= 1 && maxQueueSize >= 1);
  9. /* 接下来要构建threadNum个线程 */
  10. pthread_t tid_t;
  11. for (int i = 0; i < threadNum; i++) {
  12. Pthread_create(&tid_t, NULL, startThread, this);
  13. }
  14. }

这里ThreadPool有两个条件变量,一个是notEmpty_,一个是notFull_,构造函数接受两个参数,一个是线程的数目,一个是最大的队列的大小.

接下来是所有的线程都运行的函数startThread:

  1. void* ThreadPool::startThread(void* obj)
  2. { /* 工作者线程 */
  3. Pthread_detach(Pthread_self());
  4. ThreadPool* pool = static_cast<ThreadPool*>(obj);
  5. pool->run();
  6. return pool;
  7. }

它们都开始调用run函数:

  1. void ThreadPool::run()
  2. {
  3. for ( ; ; ) { /* 一直运行下去 */
  4. Task task(take());
  5. if (task) {
  6. //mylog("task run!");
  7. task();
  8. }
  9. //mylog("task over!");
  10. }
  11. }

run函数非常简单,就是不断从队列中取出任务,然后运行任务,没有任务的话,会阻塞在那里.

我们来看take函数:

  1. ThreadPool::Task ThreadPool::take()
  2. {
  3. MutexLockGuard lock(mutex_); /* 加锁 */
  4. while (queue_.empty()) { /* 如果队列为空 */
  5. notEmpty_.wait(); /* 等待 */
  6. }
  7. Task task;
  8. if (!queue_.empty()) {
  9. task = queue_.front();
  10. queue_.pop_front();
  11. if (maxQueueSize_ > 0) { /* 通知生产者队列有空位置了 */
  12. notFull_.notify();
  13. }
  14. }
  15. //mylog("threadpool take 1 task!");
  16. return task;
  17. }

对于生产者而言,有一个非常重要的函数,那就是append:

  1. bool ThreadPool::append(Task&& task)
  2. { /* 使用了右值引用 */
  3. {
  4. MutexLockGuard lock(mutex_); /* 首先加锁 */
  5. while (isFull()) { /* 如果队列已满 */
  6. notFull_.wait(); /* 等待queue有空闲位置 */
  7. }
  8. assert(!isFull());
  9. queue_.push_back(std::move(task)); /* 直接用move语义,提高了效率 */
  10. //mylog("put task onto queue!");
  11. }
  12. notEmpty_.notify(); /* 通知消费者有任务可做了 */
  13. }

生产者消费者队列的代码就是这么简单,但是muduo库写的确实很漂亮.

我的思路

其实代码基本上和前面的类似,不同的是,我压根就没有考虑过使用boost::funcitonboost::bind这对神器,因为我之前也压根就没有这样编过码.

如果不用boost::funcitonboost::bind这两样东西,我们要实现类似的代码的话,可能的一个解决方案是使用模版(template).

队列里面放的是T类型,然后消费者取出一个T类型,调用T类型的一个run或者别的什么不带参数的方法.这样以来,对T类型就有了限制,要求T类型必须实现run之类的方法.

而且代码变得不太容易读.加了模版的玩意总是不容易读,不是吗?所以要积极使用cpp的新特性.

主程序变成了生产者

这一次的代码变得简洁多了,

  1. int main(int argc, char *argv[])
  2. {
  3. int listenfd = Open_listenfd(8080); /* 8080号端口监听 */
  4. epoll_event events[MAXEVENTNUM];
  5. sockaddr clnaddr;
  6. socklen_t clnlen = sizeof(clnaddr);
  7. block_sigpipe(); /* 首先要将SIGPIPE消息阻塞掉 */
  8. int epollfd = Epoll_create(1024); /* 10基本上没有什么用处 */
  9. addfd(epollfd, listenfd, false); /* epollfd要监听listenfd上的可读事件 */
  10. ThreadPool pools(10, 30000); /* 10个线程,300个任务 */
  11. HttpHandle::setEpollfd(epollfd);
  12. HttpHandle handle[2000];
  13. for ( ; ;) {
  14. int eventnum = Epoll_wait(epollfd, events, MAXEVENTNUM, -1);
  15. for (int i = 0; i < eventnum; ++i) {
  16. int sockfd = events[i].data.fd;
  17. if (sockfd == listenfd) { /* 有连接到来 */
  18. //mylog("connection comes!");
  19. for ( ; ; ) {
  20. int connfd = accept(listenfd, &clnaddr, &clnlen);
  21. if (connfd == -1) {
  22. if ((errno == EAGAIN) || (errno == EWOULDBLOCK)) { /* 将连接已经建立完了 */
  23. break;
  24. }
  25. unix_error("accept error");
  26. }
  27. handle[connfd].init(connfd); /* 初始化 */
  28. addfd(epollfd, connfd, false); /* 加入监听 */
  29. }
  30. }
  31. else { /* 有数据可读或者可写 */
  32. pools.append(boost::bind(&HttpHandle::process, &handle[sockfd]));
  33. }
  34. }
  35. }
  36. return 0;
  37. }

注意最后的一句boost::bind(&HttpHandle::process, &handle[sockfd]),直接将对象往函数上一绑定,就往队列里面扔.非常爽.

这一次,我们终于将SIGPIPE消息给忽略掉了,主要是调用下面这个函数:

  1. void block_sigpipe()
  2. {
  3. sigset_t signal_mask;
  4. sigemptyset(&signal_mask);
  5. sigaddset(&signal_mask, SIGPIPE);
  6. int rc = pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);
  7. if (rc != 0) {
  8. printf("block sigpipe error\n");
  9. }
  10. }

shared_ptr并不是线程安全的

正如文章开头所讲的,多线程一来,很多事情就变得莫名奇妙了,比如说shared_ptr,因为这个玩意的线程不安全性,我调了半天bug,才发现原来是Cache的查找函数出了问题,下面是修改过后的线程安全版本的函数:

  1. /* 线程安全版本的getFileAddr */
  2. void getFileAddr(std::string fileName, int fileSize, boost::shared_ptr<FileInfo>& ptr) {
  3. /*-
  4. * shared_ptr并不是线程安全的,对其的读写都需要加锁.
  5. */
  6. MutexLockGuard lock(mutex_);
  7. if (cache_.end() != cache_.find(fileName)) { /* 如果在cache中找到了 */
  8. ptr = cache_[fileName];
  9. return;
  10. }
  11. if (cache_.size() >= MAX_CACHE_SIZE) { /* 文件数目过多,需要删除一个元素 */
  12. cache_.erase(cache_.begin()); /* 直接移除掉最前一个元素 */
  13. }
  14. boost::shared_ptr<FileInfo> fileInfo(new FileInfo(fileName, fileSize));
  15. cache_[fileName] = fileInfo;
  16. ptr = std::move(fileInfo); /* 直接使用move语义 */
  17. }

至于为什么不安全,可以查看这里,写的再好不过了:http://blog.csdn.net/solstice/article/details/8547547

多线程的调试

原谅我到了这接近尾声的时候,才提起多线程的调试,首先要说一句的是,多线程真的不太好调,因为很难重现错误,但是我在这里稍稍介绍一下我的技巧.

打印

打印算是屡试不爽的一种方法,对于我们这个简陋的web server,我封装了一个日志函数mylog:

  1. void mlog(pthread_t tid, const char *fileName, int lineNum, const char *func, const char *log_str, ...)
  2. {
  3. va_list vArgList; //定义一个va_list型的变量,这个变量是指向参数的指针.
  4. char buf[1024];
  5. va_start(vArgList, log_str); //用va_start宏初始化变量,这个宏的第二个参数是第一个可变参数的前一个参数,是一个固定的参数
  6. vsnprintf(buf, 1024, log_str, vArgList); //注意,不要漏掉前面的_
  7. va_end(vArgList); //用va_end宏结束可变参数的获取
  8. printf("%lu:%s:%d:%s --> %s\n", tid, fileName, lineNum, func, buf);
  9. }

然后定义了一个宏,方便使用这个函数:

  1. #define mylog(formatPM, args...)\
  2. mlog(pthread_self(), __FILE__, __LINE__, __FUNCTION__, (formatPM) , ##args)

需要日志的时候,可以像printf函数一样使用:

  1. mylog("My simple web server! %d, %s\n", 1, "hello, workd!");

这个宏展开后会调用mlog函数,打印出行,文件名,函数名等信息,对付我们这个小玩意足够了.

用VS来调试

VS其实也内置了线程的调试,你可以结合Visual Gdb一起来调试linux下的代码.一两个线程问题倒是不大,不过线程多了的话,这个玩意就不好调了,要我说,最好的方法还是分析日志.

总结

这个版本已经算是比较强劲的一个版本了,修复了前面的一些bug,但是引入了新的bug,这个bug我也是折腾了很久才弄出来.

一般在单线程下不可能出现这样的bug,只有在多线程的条件下,这样的代码才变成了bug,正如前面见到的,每个HttpHandle处理一个连接,试想这样一种情形:客户端不知道因为什么原因,第一次发送了这样的数据:

  1. GET /

隔了很短时间才会发送余下的数据.这时,第一次发送的数据正在被另外一个线程处理,在多线程条件下,对于第二次到来的数据,这个HttpHandle会交由另外一个线程处理,也就是说,有两个线程在不加锁地使用同一个HttpHandle,不出问题才怪.

解决方案是有的,那就是EPOLLONESHOT参数.不过那是下一个版本的故事啦.

和之前类似的,代码在这里:https://github.com/lishuhuakai/Spweb

添加新批注
在作者公开此批注前,只有你和作者可见。
回复批注