一个消息队列服务器模型

linux下epoll+多线程，使用nosql数据库进行数据持久化存储

#include <sys/socket.h>
#include <sys/epoll.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdio.h>
#include <pthread.h>
#include <errno.h>
#include <string.h>
#include <tcbdb.h>

#define PORT 8888
#define MAXFDS 10000
#define EVENTSIZE 5000
#define THREADNUM 2
#define BUFFERSIZE 4096
#define REPLY_BUFFER "HTTP/1.1 200 OK/r/nConnection: close/r/nContent-Type: text/html/r/n/r/n"
int epfd[THREADNUM];
int THREADCOUNT;
pthread_mutex_t LOCK;
TCBDB *event_db;
struct HTTP{
  char name[ 256 ];
  char charset[ 32 ];
  char opt[ 8 ];
  char data[ 1024 ];
  char pos [ 16 ];
  char num[ 16 ];
  char content[ 1024 ];
};
void *serv_epoll(void *p);
void setnonblocking(int fd)
{
  int opts;
  opts=fcntl(fd, F_GETFL);
  if (opts < 0)
    {
      fprintf(stderr, "fcntl failed/n");
      return;
    }
  opts = opts | O_NONBLOCK;
  if(fcntl(fd, F_SETFL, opts) < 0)
    {
      fprintf(stderr, "fcntl failed/n");
      return;
    }
  return;
}
void get_itemdata( char *buffer, char *itemname, char *item )
{
    int count = strlen( itemname ),length = count + 1;
    char *ptr;
    if(ptr = strstr( buffer, itemname )){
      while( (ptr[ count ] != ' ') && (ptr[ count ] != '&') && (ptr[ count ] != '/r') && (ptr[ count ] != '/0')){
    item[ count - length ] = ptr[ count ];
    count++;
      }
      item[ count - length + 1 ] = '/0';
    }
}
void show_help(void)
{
        char *b = "--------------------------------------------------------------------------------------------------/n"
                  "Message Queue Service" " (April 6, 2010)/n/n"
                  "Author: Kang YuXiang , E-mail: [e-mail]/n"
     "This is free software, and you are welcome to modify and redistribute it under the New BSD License/n"
                  "/n"
     //"-l   interface to listen on, default is 0.0.0.0/n"
     //"-p       TCP port number to listen on (default: 1218)/n"
     //"-x      database directory (example: /opt/httpsqs/data)/n"
     //"-t    timeout for an http request (default: 3)/n"
     //"-s    the interval to sync updated contents to the disk (default: 5)/n"
     //"-c       the maximum number of non-leaf nodes to be cached (default: 1024)/n"
     //"-m      database memory cache size in MB (default: 100)/n"
     //"-h            print this help and exit/n/n"
     // "Use command /"killall httpsqs/", /"pkill httpsqs/" and /"kill `cat /tmp/httpsqs.pid`/" to stop httpsqs./n"
     // "Please note that don't use the command /"pkill -9 httpsqs/" and /"kill -9 PID of httpsqs/"!/n"
                   "/n"
                   "--------------------------------------------------------------------------------------------------/n"
                   "/n";
    fprintf(stderr, b, strlen(b));
}

int main(int argc, char *argv[])
{
    int fd, cfd,opt=1;
    unsigned char i;
    struct epoll_event ev;
    struct sockaddr_in sin, cin;
    socklen_t sin_len = sizeof(struct sockaddr_in);
    pthread_t tid;
    pthread_attr_t attr;
    THREADCOUNT = 0;
    show_help( );
    event_db = tcbdbnew();
    tcbdbtune(event_db, 1024, 2048, 50000000, 8, 10, BDBTLARGE);
    tcbdbsetcache(event_db, 2048, 1024);
    tcbdbsetxmsiz(event_db, 104857600);
    /* open the database */
    if(!tcbdbopen(event_db, "casket.tct", BDBOWRITER | BDBOCREAT)){
      printf( "Attention: Unable to open the database./n/n");
      exit( 1 );
    }

    for( i = 0; i < THREADNUM; i++)
    {
    epfd[ i ] = epoll_create(MAXFDS);
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_DETACHED);
    if (pthread_create(&tid, &attr, serv_epoll, NULL) != 0)
     {
        fprintf(stderr, "pthread_create failed/n");
        return -1;
     }
    }


    if ((fd = socket(AF_INET, SOCK_STREAM, 0)) <= 0)
      {
    fprintf(stderr, "socket failed/n");
    return -1;
      }
    setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (const void*)&opt, sizeof(opt));

    memset(&sin, 0, sizeof(struct sockaddr_in));
    sin.sin_family = AF_INET;
    sin.sin_port = htons((short)(PORT));
    sin.sin_addr.s_addr = INADDR_ANY;
    if (bind(fd, (struct sockaddr *)&sin, sizeof(sin)) != 0)
      {
    fprintf(stderr, "bind failed/n");
    return -1;
      }
    if (listen(fd, 32) != 0)
      {
    fprintf(stderr, "listen failed/n");
    return -1;
      }

    i = 0;
    while ((cfd = accept(fd, (struct sockaddr *)&cin, &sin_len)) > 0)
      {
    setnonblocking(cfd);
    ev.data.fd = cfd;
    ev.events = EPOLLIN | EPOLLET;
    epoll_ctl(epfd[i%THREADNUM], EPOLL_CTL_ADD, cfd, &ev);
    i++;
      }

    if (fd > 0)
      close(fd);
    return 0;
}
void *serv_epoll(void *p)
{
  int i, ret, cfd, nfds,head,tail;
  struct epoll_event ev,events[EVENTSIZE];
  char buffer[BUFFERSIZE];
  const char* ptr;
  int temp;
  struct HTTP http;
  pthread_mutex_lock( &LOCK );
  temp = THREADCOUNT;
  THREADCOUNT++;
  pthread_mutex_unlock( &LOCK );

  while (1)
  {
      nfds = epoll_wait(epfd[temp], events, EVENTSIZE , -1);
      for (i=0; i<nfds; i++)
    {
      memset( http.charset, '/0', 32 );
      memset( http.data, '/0', 1024 );
      memset( http.content, '/0', 1024 );
      memset( http.opt, '/0', 8 );
      memset( http.name, '/0', 256 );
      memset( http.pos, '/0', 16 );
      memset( http.num, '/0', 16 );
     if(events[i].events & EPOLLIN)
     {
          cfd = events[i].data.fd;
          ret = recv(cfd, buffer, sizeof(buffer),0);
     //printf( "%s/n",buffer );
          head = 0;
          tail = 0;
     /*判断读数据是否成功*/
     if( ret <= 0 )
        {
          ev.data.fd = cfd;
          epoll_ctl(epfd[ temp ], EPOLL_CTL_DEL, cfd, &ev);
          close(cfd);
          return NULL;
        }
          get_itemdata( buffer, "name", http.name );//printf( "http.name:%s/n", http.name );
          get_itemdata( buffer, "opt", http.opt );//printf( "http.opt:%s/n", http.opt );
          get_itemdata( buffer, "charset", http.charset );//printf( "http.charset%s/n", http.charset );
          get_itemdata( buffer, "pos", http.pos );//printf( "http.pos:%s/n", http.pos );
          get_itemdata( buffer, "num", http.num );//printf( "http.num:%s/n", http.num );
          get_itemdata( buffer, "data", http.data );//printf( "http.data:%s/n", http.data );

     /*获取post正文信息*/
          ptr = strstr( buffer, "/r/n/r/n" );
     if(strlen( ptr ) > 0 ){
        head = 4;
        while( ptr[ head ] != '/0'){
          http.content[ head-5 ] = ptr[ head ];
          head++;
        }
        http.content[ head ] = '/0';
        //printf( "http.content:%s/n",http.content );
     }
          memset( buffer, '/0', BUFFERSIZE);
          strcpy(buffer, REPLY_BUFFER);
          ret = 0;
     /*根据opt进行相应的入库或者出库操作，并给客户端发送回复信息*/
     if( strcmp( http.opt,"put" ) == 0){
        if( strlen( http.content ) > 0)
            tcbdbput2( event_db, http.name, http.content);
        else
            tcbdbput2( event_db, http.name, http.data);
        strcat( buffer, "OK" );
        ret = send(cfd, buffer, strlen(buffer), 0);
        ev.data.fd = cfd;
        epoll_ctl(epfd[ temp ], EPOLL_CTL_DEL, cfd, &ev);
        close(cfd);
     }
     else if( strcmp(http.opt,"get") == 0 )
        {
          printf( "opt is get/n");
         //printf("http.content:%s/n",tcbdbget2(event_db, http.name));
          ptr = tcbdbget2(event_db, http.name);
         //printf( "ptr:%s/n", ptr );
         if( ptr )
            strcat(buffer ,tcbdbget2(event_db, http.name));
         else
            strcat(buffer ,"no such name");
         if(write( cfd, buffer, sizeof( buffer)) <= 0)
         {
               ev.data.fd = cfd;
               ev.events = EPOLLOUT | EPOLLET;
               epoll_ctl(epfd[ temp ], EPOLL_CTL_MOD, cfd, &ev);
         }
         else{
            ev.data.fd = cfd;
            epoll_ctl(epfd[ temp ], EPOLL_CTL_DEL, cfd, &ev);
            close(cfd);
         }
        }
     }
     else if(events[i].events & EPOLLOUT)
     {
          cfd = events[i].data.fd;
          ret = send(cfd, REPLY_BUFFER, strlen(REPLY_BUFFER), 0);
     //printf("send ret...........= %d/n", ret);

          ev.data.fd = cfd;
          epoll_ctl(epfd[ temp ], EPOLL_CTL_DEL, cfd, &ev);
     //shutdown(cfd, 1);
          close(cfd);
     }
    }
  }
  return NULL;
}