Linux下的C++项目开发：聊天室（三）

---

多线程客户端

公用的头文件写在global.h 与global.cpp，代码如下：

global.h:

#ifndef _GLOBAL_H
#define _GLOBAL_H

#include<set>
#include<iostream>
#include<stdlib.h>
#include<unistd.h>
#include<string.h>
#include<vector>
#include<arpa/inet.h>
#include<sys/socket.h>
#include<netinet/in.h>
#include<mysql/mysql.h>
#include<unordered_map>
#include<pthread.h>
#include<sys/epoll.h>
#include<fcntl.h>
#include<unistd.h>
#include<queue>
#include<chrono>
#include<boost/bind.hpp>
#include<boost/asio.hpp>
#include<errno.h>
using namespace std;
using namespace chrono;

extern unordered_map<string,int> name_sock_map;//记录名字和文件描述符
extern unordered_map<int,set<int>> group_map;//记录群号和对应的文件描述符集合
extern unordered_map<string,string> from_to_map;//key:用户名 value:key的用户想私聊的用户
//extern time_point<system_clock> begin_clock;//开始时间，用于压力测试
//extern clock_t begin_clock;//开始时间，用于性能测试，有bug
extern double total_time;//线程池处理任务的总时间
extern int total_handle;//总处理请求数，用于性能测试
extern double top_speed;//记录峰值性能
extern int total_recv_request;//接收到的请求总数，性能测试
extern int Bloom_Filter_bitmap[1000000];//布隆过滤器所用的bitmap
extern queue<int> mission_queue;//任务队列
extern int mission_num;//任务队列中的任务数量
extern pthread_cond_t mission_cond;//线程池所需的条件变量
extern pthread_spinlock_t name_mutex;//互斥锁，锁住需要修改name_sock_map的临界区
extern pthread_spinlock_t from_mutex;//互斥锁，锁住修改from_to_map的临界区
extern pthread_spinlock_t group_mutex;//互斥锁，锁住修改group_map的临界区
extern pthread_mutex_t queue_mutex;//互斥锁，锁住修改任务队列的临界区
extern int epollfd;
extern pthread_spinlock_t count_mutex;

#endif

global.cpp

#include "global.h"

unordered_map<string,int> name_sock_map;
unordered_map<int,set<int>> group_map;
unordered_map<string,string> from_to_map;//key:用户名 value:key的用户想私聊的用户
//time_point<system_clock> begin_clock;
double total_time;//线程池处理任务的总时间
//clock_t begin_clock;//开始时间，用于性能测试
int total_handle;//总处理请求数，用于性能测试
double top_speed;//记录峰值性能
int total_recv_request;//接收到的请求总数，性能测试
int Bloom_Filter_bitmap[1000000];//布隆过滤器所用的bitmap
queue<int> mission_queue;//任务队列
int mission_num;//任务队列中的任务数量
pthread_cond_t mission_cond;//线程池所需的条件变量
pthread_spinlock_t name_mutex;//互斥锁，锁住需要修改name_sock_map的临界区
pthread_spinlock_t from_mutex;//互斥锁，锁住修改from_to_map的临界区
pthread_spinlock_t group_mutex;//互斥锁，锁住修改group_map的临界区
pthread_mutex_t queue_mutex;//互斥锁，锁住修改任务队列的临界区
int epollfd;
pthread_spinlock_t count_mutex;

处理客户端的头文件HandleClient.h：

#ifndef _HANDLECLIENT_H
#define _HANDLECLIENT_H

#include<iostream>
#include<pthread.h>
#include<stdlib.h>
#include<unistd.h>
#include<string.h>
#include<arpa/inet.h>
#include<sys/socket.h>
#include<mysql/mysql.h>
#include<netinet/in.h>
using namespace std;

//线程执行此函数来发送消息
void *handle_send(void *arg);

//线程执行此函数来接收消息
void *handle_recv(void *arg);

#endif

声明两个处理函数，一个发送消息函数，一个接收消息函数。

处理客户端的程序HandleClient.cpp：

#include"HandleClient.h"
using namespace std;

void *handle_recv(void *arg){
    int sock=*(int *)arg;
    while(1){
        char recv_buffer[1000];
        memset(recv_buffer,0,sizeof(recv_buffer));
        int len=recv(sock,recv_buffer,sizeof(recv_buffer),0);
        if(len==0)
            continue;
        if(len==-1)
            break;
        string str(recv_buffer);
        cout<<str<<endl;
    }
}

void *handle_send(void *arg){
    int sock=*(int *)arg;
    while(1){
        string str;
        cin>>str;
        if(str=="exit")
            break;
        if(sock>0){
            str="content:"+str;
            send(sock,str.c_str(),str.length(),0);
        }
        else if(sock<0){
            str="gr_message:"+str;
            send(-sock,str.c_str(),str.length(),0);
        }   
    }   
}

接收消息函数，接收传入的套接字描述符，读取消息到缓冲区recv_buffer，成功读取消息后输出字符串内容。

发送消息函数，传入发送数据的套接字描述符，读取输入的字符串，如果为”exit”，终止输入；如果套接字描述符的值大于0，为私聊消息，在输入的str前加上”content”前缀标记，执行数据发送；如果套接字描述符的值小于0，为群聊消息，在输入的str前加上”gr_message:”标记，执行数据发送。

多线程客户端主程序client.cpp:

头文件导入：

#include <sys/types.h>
#include <sys/socket.h>
#include <stdio.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>
#include <fcntl.h>
#include <sys/shm.h>
#include<fstream>
#include<iostream>
#include "HandleClient.h"
using namespace std;

extern void *hand_recv(void *arg);

extern void *handle_send(void *arg);

主函数内容：

定义一个TCP协议的socket，设置服务器端的IP地址与端口号，客户端套接字描述符sock主动连接服务端，调用connect函数；choice记录用户输入的选项（0:退出，1:登录，2:注册），if_login记录是否成功登陆，如果成功登陆，记录该用户名login_name。

int sock;
struct sockaddr_in serv_addr;
pthread_t snd_thread, rcv_thread;
sock = socket(PF_INET, SOCK_STREAM, 0);

memset(&serv_addr, 0, sizeof(serv_addr));
serv_addr.sin_family = AF_INET;
serv_addr.sin_addr.s_addr = inet_addr("192.168.3.202");
serv_addr.sin_port = htons(8023);

if (connect(sock, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) == -1)
    cout<<"connect() error";

int choice;
string name,pass,pass1;
bool if_login=false;//记录是否登录成功
string login_name;//记录成功登录的用户名

新增模块：发送本地cookie，并接收服务器答复，如果答复通过就不用登录

//先检查是否存在cookie文件
ifstream f("cookie.txt");
string cookie_str;
if(f.good()){
    f>>cookie_str;
    f.close();
    cookie_str="cookie:"+cookie_str;
    //将cookie发送到服务器
    send(sock,cookie_str.c_str(),cookie_str.length()+1,0);
    //接收服务器答复
    char cookie_ans[100];
    memset(cookie_ans,0,sizeof(cookie_ans));
    recv(sock,cookie_ans,sizeof(cookie_ans),0);
    //判断服务器答复是否通过
    string ans_str(cookie_ans);
    if(ans_str!="NULL"){//redis查询到了cookie，通过
        if_login=true;
        login_name=ans_str;
    }
}    

初始未登录时：

if(!if_login){
    cout<<" ------------------\n";
    cout<<"|                  |\n";
    cout<<"| 请输入你要的选项:|\n";
    cout<<"|    0:退出        |\n";
    cout<<"|    1:登录        |\n";
    cout<<"|    2:注册        |\n";
    cout<<"|                  |\n";
    cout<<" ------------------ \n\n";
}

处理登录与注册事务，如果已登录，跳出该事务；输入choice，choice为0跳出循环；choice为1，输入用户账号与密码合并为字符串str，str前加入”login”前缀标记，发送登录消息，定义buffer缓存区接收服务端的接收响应，接收消息为recv_str，如果为”ok”，代表密码正确登录成功，本地建立cookie文件保存sessionid。sessionid的起始位置为recv_str的第三位，登录成功则跳出该事务，失败提示密码或用户名错误，等待输入choice选项。输入choice为2，执行注册事务，输入用户名与密码，密码需要第二次输入确认，字符串str存储用户账号+密码，发送到服务端。

//开始处理各种事务
while(1){
    if(if_login)
        break;
    cin>>choice;
    if(choice==0)
        break;
    //登录
    else if(choice==1&&!if_login){
        while(1){
            cout<<"用户名:";
            cin>>name;
            cout<<"密码:";
            cin>>pass;
            string str="login"+name;
            str+="pass:";
            str+=pass;
            send(sock,str.c_str(),str.length(),0);//发送登录信息
            char buffer[1000];
            memset(buffer,0,sizeof(buffer));
            recv(sock,buffer,sizeof(buffer),0);//接收响应
            string recv_str(buffer);
            if(recv_str.substr(0,2)=="ok"){
                if_login=true;
                login_name=name;

                //新增：本地建立cookie文件保存sessionid
                string tmpstr=recv_str.substr(2);
                tmpstr="cat > cookie.txt <<end \n"+tmpstr+"\nend";
                system(tmpstr.c_str());

                cout<<"登陆成功\n\n";
                break;
            }
            else
                cout<<"密码或用户名错误！\n\n";
        }       
    }
    //注册
    else if(choice==2){
        cout<<"注册的用户名:";
        cin>>name;
        while(1){
            cout<<"密码:";
            cin>>pass;
            cout<<"确认密码:";
            cin>>pass1;
            if(pass==pass1)
                break;
            else
                cout<<"两次密码不一致!\n\n";
        }   
        name="name:"+name;
        pass="pass:"+pass;
        string str=name+pass;
        send(sock,str.c_str(),str.length(),0);
        cout<<"注册成功！\n";
        cout<<"\n继续输入你要的选项:";     
    }
    if(if_login)
        break;
}

登录成功后，输入新的选项值choice（0:退出，1:私聊，2:群聊），建立两个线程分别执行发送与接收任务。choice为0，退出登录；choice为1，执行私聊事务，输入对方的用户名target_name，发送字符串为”target:”+target_name+”from:”+login_name，标记目标用户与源用户，执行发送函数，创建发送与接收线程，执行handle_send和handle_recv；choice为2，执行群聊事务，输入群号，字符串str加上”group:”前缀标记，发送到服务端，将sock套接字描述符的值变为负值，创建发送与接收线程，执行handle_send和handle_recv。

//登陆成功
while(if_login&&1){
    if(if_login){
        system("clear");
        cout<<"        欢迎回来,"<<login_name<<endl;
        cout<<" -------------------------------------------\n";
        cout<<"|                                           |\n";
        cout<<"|          请选择你要的选项：               |\n";
        cout<<"|              0:退出                       |\n";
        cout<<"|              1:发起单独聊天               |\n";
        cout<<"|              2:发起群聊                   |\n";
        cout<<"|                                           |\n";
        cout<<" ------------------------------------------- \n\n";
    }
    cin>>choice;
    pthread_t send_t,recv_t;//线程ID
    void *thread_return;
    if(choice==0)
        break;
    if(choice==1){
        cout<<"请输入对方的用户名:";
        string target_name,content;
        cin>>target_name;
        string sendstr("target:"+target_name+"from:"+login_name);//标识目标用户+源用户
        send(sock,sendstr.c_str(),sendstr.length(),0);//先向服务器发送目标用户、源用户
        cout<<"请输入你想说的话(输入exit退出)：\n";
        auto send_thread=pthread_create(&send_t,NULL,handle_send,(void *)&sock);//创建发送线程
        auto recv_thread=pthread_create(&recv_t,NULL,handle_recv,(void *)&sock);//创建接收线程
        pthread_join(send_t,&thread_return);
        //pthread_join(recv_t,&thread_return);
        pthread_cancel(recv_t);
    }    
    if(choice==2){
        cout<<"请输入群号:";
        int num;
        cin>>num;
        string sendstr("group:"+to_string(num));
        send(sock,sendstr.c_str(),sendstr.length(),0);
        cout<<"请输入你想说的话(输入exit退出)：\n";
        int sock1=-sock;
        auto send_thread=pthread_create(&send_t,NULL,handle_send,(void *)&sock1);//创建发送线程
        auto recv_thread=pthread_create(&recv_t,NULL,handle_recv,(void *)&sock);//创建接收线程
        pthread_join(send_t,&thread_return);
        pthread_cancel(recv_t);
    }
} 
close(sock);

线程池服务器

公用头文件相同，global.h与global.cpp

处理线程池服务器的头文件HandleServerUseThreadPool.h，声明一个线程处理请求函数。

#ifndef _HANDLE_SERVER_H
#define _HANDLE_SERVER_H

#include<iostream>
#include<pthread.h>
#include<stdlib.h>
#include<unistd.h>
#include<string.h>
#include<arpa/inet.h>
#include<sys/socket.h>
#include<mysql/mysql.h>
#include<time.h>
#include<netinet/in.h>
#include<netinet/tcp.h>
#include <hiredis/hiredis.h>
using namespace std;

//线程执行此函数，处理请求
void handle_all_request(int arg);

#endif

处理线程池服务器的文件HandleServerUseThreadPool.cpp，定义该线程处理请求函数。

name_sock_map记录用户名与套接字描述符（哈希表），group_map记录群号与套接字描述符集合（哈希表）。定义并创建两个互斥锁，分别用于修改name_sock_map的临界区和group_map的临界区。

#include"HandleServerUseThreadPool.h"
#include"global.h"

//extern unordered_map<pair<int,string>,pair<int,string>> from_to_map;//记录目标用户、源用户
extern unordered_map<string,int> name_sock_map;//名字和套接字描述符
extern unordered_map<int,set<int>> group_map;//记录群号和套接字描述符集合

void handle_all_request(int arg){
    pthread_mutex_t mutx;//互斥锁，锁住需要修改name_sock_map的临界区
    pthread_mutex_t group_mutx;//互斥锁，锁住修改group_map的临界区
    pthread_mutex_init(&mutx, NULL); //创建互斥锁
    pthread_mutex_init(&group_mutx,NULL);//创建互斥锁
    int conn=arg;
    int target_conn=-1;
    char buffer[1000];
    string name,pass;
    bool if_login=false;//记录当前服务对象是否成功登录
    string login_name;//记录当前服务对象的名字
    string target_name;//记录发送信息时目标用户的名字
    int group_num;//记录群号

    //连接MYSQL数据库
    MYSQL *con=mysql_init(NULL);
    mysql_real_connect(con,"127.0.0.1","root","","test_connect",0,NULL,CLIENT_MULTI_STATEMENTS);

    //连接redis数据库
    redisContext *redis_target = redisConnect("127.0.0.1",6379);
    if(redis_target->err){
        redisFree(redis_target);
        cout<<"连接redis失败"<<endl;     
    }

    //禁用nagle算法，防止粘包
    //int enable = 1;
    //setsockopt(conn, IPPROTO_TCP, TCP_NODELAY, (void*)&enable, sizeof(enable));

    while(1){
        cout<<"-----------------------------\n";
        memset(buffer,0,sizeof(buffer));
        int len = recv(conn, buffer, sizeof(buffer),0);

        //断开了连接或者发生了异常
        if(len==0||len==-1)
            break;

        string str(buffer);
        //cout<<"用户"<<inet_ntoa(clnt_adr.sin_addr)<<"正在连接";

        //新增：先接收cookie看看redis是否保存该用户的登录状态
        if(str.find("cookie:")!=str.npos){
            string cookie=str.substr(7);
            string redis_str="hget "+cookie+" name";
            redisReply *r = (redisReply*)redisCommand(redis_target,redis_str.c_str());
            string send_res;
            if(r->str){
                cout<<"查询redis结果："<<r->str<<endl;
                send_res=r->str;
                //cout<<sizeof(r->str)<<endl;
                // cout<<send_res.length()<<endl;
            }
            else
                send_res="NULL";
            send(conn,send_res.c_str(),send_res.length()+1,0);
            //if(r->str==)
        }

        //登录
        else if(str.find("login")!=str.npos){
            int p1=str.find("login"),p2=str.find("pass:");
            name=str.substr(p1+5,p2-5);
            pass=str.substr(p2+5,str.length()-p2-4);
            string search="SELECT * FROM user WHERE NAME=\"";
            search+=name;
            search+="\";";
            cout<<"sql语句:"<<search<<endl;
            auto search_res=mysql_query(con,search.c_str());
            auto result=mysql_store_result(con);
            int col=mysql_num_fields(result);//获取列数
            int row=mysql_num_rows(result);//获取行数
            //auto info=mysql_fetch_row(result);//获取一行的信息
            if(search_res==0&&row!=0){
                cout<<"查询成功\n";
                //auto result=mysql_store_result(con);
                //int col=mysql_num_fields(result);//获取列数
                //int row=mysql_num_rows(result);//获取行数
                auto info=mysql_fetch_row(result);//获取一行的信息
                cout<<"查询到用户名:"<<info[0]<<" 密码:"<<info[1]<<endl;
                if(info[1]==pass){
                    cout<<"登录密码正确\n";
                    string str1="ok";
                    if_login=true;
                    login_name=name;
                    pthread_mutex_lock(&mutx); //上锁
                    name_sock_map[name]=conn;//记录下名字和文件描述符的对应关系
                    pthread_mutex_unlock(&mutx); //解锁

                    //新添加：随机生成sessionid并发送到客户端
                    srand(time(NULL));//初始化随机数种子
                    for(int i=0;i<10;i++){
                        int type=rand()%3;
                        //type为0代表数字，为1代表小写字母，为2代表大写字母
                        if(type==0)
                            str1+='0'+rand()%9;
                        else if(type==1)
                            str1+='a'+rand()%26;
                        else if(type==2)
                            str1+='A'+rand()%26;
                    }
                    //将sessionid存入redis
                    string redis_str="hset "+str1.substr(2)+" name "+login_name;
                    redisReply *r = (redisReply*)redisCommand(redis_target,redis_str.c_str());
                    //设置生存时间,默认300秒
                    redis_str="expire "+str1.substr(2)+" 300";
                    r=(redisReply*)redisCommand(redis_target,redis_str.c_str());

                    cout<<"随机生成的sessionid为："<<str1.substr(2)<<endl;
                    //cout<<"redis指令:"<<r->str<<endl;

                    send(conn,str1.c_str(),str1.length()+1,0);
                }
                else{
                    cout<<"登录密码错误\n";
                    char str1[100]="wrong";
                    send(conn,str1,strlen(str1),0);
                }   
            }
            else{
                cout<<"查询失败\n";
                char str1[100]="wrong";
                send(conn,str1,strlen(str1),0);
            }
        }

        //注册
        else if(str.find("name:")!=str.npos){
            int p1=str.find("name:"),p2=str.find("pass:");
            name=str.substr(p1+5,p2-5);
            pass=str.substr(p2+5,str.length()-p2-4);
            string search="INSERT INTO user VALUES (\"";
            search+=name;
            search+="\",\"";
            search+=pass;
            search+="\");";
            cout<<endl<<"sql语句:"<<search<<endl;
            mysql_query(con,search.c_str());
        } 

        //设定目标的文件描述符
        else if(str.find("target:")!=str.npos){
            int pos1=str.find("from");
            string target=str.substr(7,pos1-7),from=str.substr(pos1+4);
            //pair<string,int> tmp1(from,name_sock_map[from]);
            //pair<string,int> tmp2(target,name_sock_map[target]);
            //from_to_map[tmp1]=tmp2;
            target_name=target;
            if(name_sock_map.find(target)==name_sock_map.end())
                cout<<"源用户为"<<login_name<<",目标用户"<<target_name<<"仍未登陆，无法发起私聊\n";
            else{
                cout<<"源用户"<<login_name<<"向目标用户"<<target_name<<"发起的私聊即将建立";
                cout<<",目标用户的套接字描述符为"<<name_sock_map[target]<<endl;
                target_conn=name_sock_map[target];
            }       
        }

        //接收到消息，转发
        else if(str.find("content:")!=str.npos){
            if(target_conn==-1){
                cout<<"找不到目标用户"<<target_name<<"的套接字，将尝试重新寻找目标用户的套接字\n";
                if(name_sock_map.find(target_name)!=name_sock_map.end()){
                    target_conn=name_sock_map[target_name];
                    cout<<"重新查找目标用户套接字成功\n";
                }
                else{
                    cout<<"查找仍然失败，转发失败！\n";
                    continue;
                }
            }
            //char recv_buff[1000];
            //memset(recv_buff,0,sizeof(recv_buff));
            //int len=recv(conn,recv_buff,sizeof(recv_buff),0);
            string recv_str(str);
            string send_str=recv_str.substr(8);
            cout<<"用户"<<login_name<<"向"<<target_name<<"发送:"<<send_str<<endl;
            send_str="["+login_name+"]:"+send_str;
            send(target_conn,send_str.c_str(),send_str.length(),0);
        }

        //绑定群聊号
        else if(str.find("group:")!=str.npos){
            string recv_str(str);
            string num_str=recv_str.substr(6);
            group_num=stoi(num_str);
            cout<<"用户"<<login_name<<"绑定群聊号为："<<num_str<<endl;
            pthread_mutex_lock(&group_mutx);//上锁
            //group_map[group_num].push_back(conn);
            group_map[group_num].insert(conn);
            pthread_mutex_unlock(&group_mutx);//解锁
        }

        //广播群聊信息
        else if(str.find("gr_message:")!=str.npos){
            string send_str(str);
            send_str=send_str.substr(11);
            send_str="["+login_name+"]:"+send_str;
            cout<<"群聊信息："<<send_str<<endl;
            for(auto i:group_map[group_num]){
                if(i!=conn)
                    send(i,send_str.c_str(),send_str.length(),0);
            }       
        }
    }  
    mysql_close(con);
    close(conn);
}

线程池服务器主函数serverUseThreadPool.cpp：

#include "HandleServerUseThreadPool.h"
#include "global.h"
using namespace std;

extern void handle_all_request(int);

//unordered_map<pair<int,string>,pair<int,string>> from_to_map;//记录源用户、目的用户
extern unordered_map<string,int> name_sock_map;//记录名字和套接字描述符

void tmptest(int a){
    cout<<a<<endl;    
}

int main(){
    int serv_sock, clnt_sock; //服务器端、客户端套接字描述符
    sockaddr_in serv_adr,clnt_adr; //服务器端、客户端的地址与端口号
    socklen_t clnt_adr_sz; //记录客户端地址长度
    pthread_mutex_t mutx;  //互斥锁

    pthread_mutex_init(&mutx, NULL); //创建互斥锁
    serv_sock = socket(PF_INET, SOCK_STREAM, 0);//PF_INET:tcp/ip协议   SOCK_STREAM: tcp    0:默认协议
    memset(&serv_adr, 0, sizeof(serv_adr));
    serv_adr.sin_family = AF_INET;//使用地址族
    serv_adr.sin_addr.s_addr = inet_addr("172.29.18.134");//ip地址
    serv_adr.sin_port = htons(8023);//端口号

    //绑定套接字与地址
    if (bind(serv_sock, (struct sockaddr *)&serv_adr, sizeof(serv_adr)) == -1)
         cout<<"bind() error";
    //服务器监听
    if (listen(serv_sock, 5) == -1)//连接队列大小为5
         cout<<"listen() error";

    //vector<int> sock_arr;//记录套接字描述符
    clnt_adr_sz = sizeof(clnt_adr);
    int conn; //连接套接字描述符
    pthread_t t_id;//线程id

    //新增boost库线程池
    /* 定义一个5线程的线程池 */
    boost::asio::thread_pool tp(5);

    // 接收连接成功
    while(conn = accept(serv_sock, (struct sockaddr *)&clnt_adr, &clnt_adr_sz)){
        //HandleServer target;
        //HandleServer::clnt_adr=clnt_adr;
        cout<<"用户"<<inet_ntoa(clnt_adr.sin_addr)<<"正在连接:\n";
        //cout<<conn<<endl;
        //sock_arr.push_back((conn);
        //新增boost库线程池
        boost::asio::post(boost::bind(handle_all_request,conn)); // 执行函数

        //pthread_create(&t_id, NULL, handle_all_request, (void *)&conn); 
        //pthread_detach(t_id);  
    }
    tp.join();//释放线程池
}

IO复用+线程池服务器

服务器处理函数头文件HandleServerV2.h

#ifndef _HANDLESERVERV2_H
#define _HANDLESERVERV2_H

#include<iostream>
#include<pthread.h>
#include<stdlib.h>
#include<unistd.h>
#include<string.h>
#include<arpa/inet.h>
#include<sys/socket.h>
#include<mysql/mysql.h>
#include<time.h>
#include<netinet/in.h>
#include<netinet/tcp.h>
#include<sys/epoll.h>
#include <hiredis/hiredis.h>
using namespace std;

//线程执行此函数，处理请求
//void handle_all_request(string,int);
void* handle_all_request(void *arg);

#endif

服务器处理函数文件HandleServerV2.cpp

name_sock_map记录用户名与套接字描述符，group_map记录群号和套接字描述符集合，from_to_map记录用户xx要向用户yy发送信息，mission_queue为任务队列。需要创建4个互斥锁，锁住修改这四个变量的临界区。

#include"HandleServerV2.h"
#include"global.h"

extern unordered_map<string,int> name_sock_map;//名字和套接字描述符
extern unordered_map<int,set<int>> group_map;//记录群号和套接字描述符集合
extern unordered_map<string,string> from_to_map;//记录用户xx要向用户yy发送信息
//extern clock_t begin_clock;//开始时间，用于性能测试
//extern time_point<system_clock> begin_clock;
extern int total_handle;//总处理请求数，用于性能测试
extern int total_recv_request;//接收到的请求总数，性能测试
extern double top_speed;//峰值性能
extern int Bloom_Filter_bitmap[1000000];//布隆过滤器所用的bitmap
extern queue<int> mission_queue;//任务队列
extern int mission_num;//任务队列中的任务数量
extern pthread_cond_t mission_cond;//线程池所需的条件变量
extern pthread_spinlock_t name_mutex;//互斥锁，锁住需要修改name_sock_map的临界区
extern pthread_spinlock_t group_mutex;//互斥锁，锁住修改group_map的临界区
extern pthread_spinlock_t from_mutex;//互斥锁，锁住修改from_to_map的临界区
extern pthread_mutex_t queue_mutex;//互斥锁，锁住修改任务队列的临界区
extern int epollfd;
extern pthread_spinlock_t count_mutex;

void* handle_all_request(void *arg){   //(string epoll_str,int conn_num,int epollfd){
    while(1){
        int conn_num;
        //取出任务
        pthread_mutex_lock(&queue_mutex);
        while(mission_queue.empty()){
            pthread_cond_wait(&mission_cond,&queue_mutex);
        }
        conn_num=mission_queue.front();
        mission_queue.pop();
        pthread_mutex_unlock(&queue_mutex);

        time_point<system_clock> begin_clock= system_clock::now();
        //pthread_spin_init(&mutex, NULL); //创建互斥锁
        //pthread_spin_init(&group_mutex,NULL);//创建互斥锁
        int conn=conn_num;
        int target_conn=-1;
        char buffer[1000];
        string name,pass;
        bool if_login=false;//记录当前服务对象是否成功登录
        string login_name;//记录当前服务对象的名字
        string target_name;//记录发送信息时目标用户的名字
        int group_num;//记录群号
        
        
        //连接MYSQL数据库
        MYSQL *con=mysql_init(NULL);
        mysql_real_connect(con,"127.0.0.1","root","","test_connect",0,NULL,CLIENT_MULTI_STATEMENTS);
        
        //连接redis数据库
        redisContext *redis_target = redisConnect("127.0.0.1",6379);
        if(redis_target->err){
            redisFree(redis_target);
            cout<<"连接redis失败"<<endl;     
        }
        
        //禁用nagle算法，防止粘包
        //int enable = 1;
        //setsockopt(conn, IPPROTO_TCP, TCP_NODELAY, (void*)&enable, sizeof(enable));

        cout<<"-----------------------------\n";
        string recv_str;
        while(1){
            char buf[10];
            memset(buf, 0, sizeof(buf));
            int ret  = recv(conn, buf, sizeof(buf), 0);
            if(ret < 0){
                cout<<"recv返回值小于0"<<endl;
                //对于非阻塞IO，下面的事件成立标识数据已经全部读取完毕
                if((errno == EAGAIN) || (errno == EWOULDBLOCK)){
                      printf("数据读取完毕\n");
                    cout<<"接收到的完整内容为："<<recv_str<<endl;
                    cout<<"开始处理事件"<<endl;
                    break;
                }
                cout<<"errno:"<<errno<<endl;
                close(conn);
                mysql_close(con);
                if(!redis_target->err)
                    redisFree(redis_target);
                //events[i].data.fd=-1;
                return NULL;
            }
            else if(ret == 0){
                cout<<"recv返回值为0"<<endl;
                close(conn);
                mysql_close(con);
                if(!redis_target->err)
                    redisFree(redis_target);
                return NULL;
                //events[i].data.fd=-1;
            }
            else{
                printf("接收到内容如下: %s\n",buf);
                string tmp(buf);
                recv_str+=tmp;
            }
        }
        string str=recv_str;

        //新增：先接收cookie看看redis是否保存该用户的登录状态
        if(str.find("cookie:")!=str.npos){
            string cookie=str.substr(7);
            string redis_str="hget "+cookie+" name";
            redisReply *r = (redisReply*)redisCommand(redis_target,redis_str.c_str());
            string send_res;
            if(r->str){
                cout<<"查询redis结果："<<r->str<<endl;
                send_res=r->str;
                pthread_spin_lock(&name_mutex); //上锁
                name_sock_map[send_res]=conn;//记录下名字和文件描述符的对应关系
                pthread_spin_unlock(&name_mutex); //解锁
                //cout<<sizeof(r->str)<<endl;
                // cout<<send_res.length()<<endl;
            }
            else
                send_res="NULL";
            send(conn,send_res.c_str(),send_res.length()+1,0);
            //if(r->str==)
        }

        //登录
        else if(str.find("login")!=str.npos){
            int p1=str.find("login"),p2=str.find("pass:"),flag=0;
            name=str.substr(p1+5,p2-5);
            pass=str.substr(p2+5,str.length()-p2-4);

            //新增布隆过滤器
            //对字符串使用哈希函数
            int hash=0;
            for(auto ch:name){
                hash=hash*131+ch;
                if(hash>=10000000)
                    hash%=10000000;                            
            }
            int index=hash/32,pos=hash%32;
            if((Bloom_Filter_bitmap[index]&(1<<pos))==0){
                cout<<"布隆过滤器查询为0，登录用户名必然不存在数据库中\n";
                char str1[100]="wrong";
                send(conn,str1,strlen(str1),0);
                flag=1;
            }
            
            //布隆过滤器无法判断才要查数据库
            if(flag==0){
                string search="SELECT * FROM user WHERE NAME=\"";
                search+=name;
                search+="\";";
                cout<<"sql语句:"<<search<<endl;
                auto search_res=mysql_query(con,search.c_str());
                auto result=mysql_store_result(con);
                //int col=mysql_num_fields(result);//获取列数
                int row;
                if(result)
                    row=mysql_num_rows(result);//获取行数
                //auto info=mysql_fetch_row(result);//获取一行的信息
                if(search_res==0&&row!=0){
                    cout<<"查询成功\n";
                    //auto result=mysql_store_result(con);
                    //int col=mysql_num_fields(result);//获取列数
                    //int row=mysql_num_rows(result);//获取行数
                    auto info=mysql_fetch_row(result);//获取一行的信息
                    cout<<"查询到用户名:"<<info[0]<<" 密码:"<<info[1]<<endl;
                    if(info[1]==pass){
                        cout<<"登录密码正确\n";
                        string str1="ok";
                        if_login=true;
                        login_name=name;
                        pthread_spin_lock(&name_mutex); //上锁
                        name_sock_map[name]=conn;//记录下名字和文件描述符的对应关系
                        pthread_spin_unlock(&name_mutex); //解锁

                        //新添加：随机生成sessionid并发送到客户端
                        srand(time(NULL));//初始化随机数种子
                        for(int i=0;i<10;i++){
                            int type=rand()%3;//type为0代表数字，为1代表小写字母，为2代表大写字母
                            if(type==0)
                                str1+='0'+rand()%9;
                            else if(type==1)
                                str1+='a'+rand()%26;
                            else if(type==2)
                                str1+='A'+rand()%26;
                        }
                        //将sessionid存入redis
                        string redis_str="hset "+str1.substr(2)+" name "+login_name;
                        redisReply *r = (redisReply*)redisCommand(redis_target,redis_str.c_str());
                        //设置生存时间,默认300秒
                        redis_str="expire "+str1.substr(2)+" 300";
                        r=(redisReply*)redisCommand(redis_target,redis_str.c_str());

                        cout<<"随机生成的sessionid为："<<str1.substr(2)<<endl;
                        //cout<<"redis指令:"<<r->str<<endl;

                        send(conn,str1.c_str(),str1.length()+1,0);
                    }
                    else{
                        cout<<"登录密码错误\n";
                        char str1[100]="wrong";
                        send(conn,str1,strlen(str1),0);
                    }   
                }
                else{
                    cout<<"查询失败\n";
                    char str1[100]="wrong";
                    send(conn,str1,strlen(str1),0);
                }
            }   
        }

        //注册
        else if(str.find("name:")!=str.npos){
            int p1=str.find("name:"),p2=str.find("pass:");
            name=str.substr(p1+5,p2-5);
            pass=str.substr(p2+5,str.length()-p2-4);
            string search="INSERT INTO user VALUES (\"";
            search+=name;
            search+="\",\"";
            search+=pass;
            search+="\");";
            cout<<endl<<"sql语句:"<<search<<endl;
            mysql_query(con,search.c_str());
        } 
        
        //设定目标的文件描述符
        else if(str.find("target:")!=str.npos){
            int pos1=str.find("from");
            string target=str.substr(7,pos1-7),from=str.substr(pos1+5);
            //pair<string,int> tmp1(from,name_sock_map[from]);
            //pair<string,int> tmp2(target,name_sock_map[target]);
            //from_to_map[tmp1]=tmp2;
            target_name=target;
            if(name_sock_map.find(target)==name_sock_map.end())
                cout<<"源用户为"<<from<<",目标用户"<<target_name<<"仍未登陆，无法发起私聊\n";
            else{
                pthread_spin_lock(&from_mutex);
                from_to_map[from]=target;
                pthread_spin_unlock(&from_mutex);
                login_name=from;
                cout<<"源用户"<<login_name<<"向目标用户"<<target_name<<"发起的私聊即将建立";
                cout<<",目标用户的套接字描述符为"<<name_sock_map[target]<<endl;
                target_conn=name_sock_map[target];
            }       
        }

        //接收到消息，转发
        else if(str.find("content:")!=str.npos){
            //根据两个map找出当前用户和目标用户
            for(auto i:name_sock_map){
                if(i.second==conn){
                    login_name=i.first;
                    target_name=from_to_map[i.first];
                    target_conn=name_sock_map[target_name];
                    break;
                }       
            }
            if(target_conn==-1){
                cout<<"找不到目标用户"<<target_name<<"的套接字，将尝试重新寻找目标用户的套接字\n";
                if(name_sock_map.find(target_name)!=name_sock_map.end()){
                    target_conn=name_sock_map[target_name];
                    cout<<"重新查找目标用户套接字成功\n";
                }
                else{
                    cout<<"查找仍然失败，转发失败！\n";
                }
            }
            //char recv_buff[1000];
            //memset(recv_buff,0,sizeof(recv_buff));
            //int len=recv(conn,recv_buff,sizeof(recv_buff),0);
            string recv_str(str);
            string send_str=recv_str.substr(8);
            cout<<"用户"<<login_name<<"向"<<target_name<<"发送:"<<send_str<<endl;
            send_str="["+login_name+"]:"+send_str;
            send(target_conn,send_str.c_str(),send_str.length(),0);
        }
        
        //绑定群聊号
        else if(str.find("group:")!=str.npos){
            string recv_str(str);
            string num_str=recv_str.substr(6);
            group_num=stoi(num_str);
            //找出当前用户
            for(auto i:name_sock_map)
                if(i.second==conn){
                    login_name=i.first;
                    break;
                }
            cout<<"用户"<<login_name<<"绑定群聊号为："<<num_str<<endl;
            pthread_spin_lock(&group_mutex);//上锁
            //group_map[group_num].push_back(conn);
            group_map[group_num].insert(conn);
            pthread_spin_unlock(&group_mutex);//解锁
        }

        //广播群聊信息
        else if(str.find("gr_message:")!=str.npos){
            //找出当前用户
            for(auto i:name_sock_map)
                if(i.second==conn){
                    login_name=i.first;
                    break;
                }
            //找出群号
            for(auto i:group_map)
                if(i.second.find(conn)!=i.second.end()){
                    group_num=i.first;
                    break;
                }
            string send_str(str);
            send_str=send_str.substr(11);
            send_str="["+login_name+"]:"+send_str;
            cout<<"群聊信息："<<send_str<<endl;
            for(auto i:group_map[group_num]){
                if(i!=conn)
                    send(i,send_str.c_str(),send_str.length(),0);
            }       
        }  
        cout<<"---------------------"<<endl;

        //线程工作完毕后重新注册事件
        epoll_event event;
        event.data.fd=conn;
        event.events=EPOLLIN|EPOLLET|EPOLLONESHOT;
        epoll_ctl(epollfd,EPOLL_CTL_MOD,conn,&event);
        
        mysql_close(con);
        if(!redis_target->err)
            redisFree(redis_target);

        //性能测试
        auto end_clock   = system_clock::now();
        auto duration = duration_cast<microseconds>(end_clock - begin_clock);
        pthread_spin_lock(&count_mutex);
        total_time+=double(duration.count()) * microseconds::period::num / microseconds::period::den; 
        total_handle++;
        pthread_spin_unlock(&count_mutex);
        //double total_time=(double)(end_clock-begin_clock)/CLOCKS_PER_SEC;
        //cout<<begin_clock<<" "<<end_clock<<endl;
        double now_rate=total_handle/total_time;
        if(now_rate>top_speed)
            top_speed=now_rate;
        cout<<"已用时"<<total_time<<"秒,";
        cout<<"共收到"<<total_recv_request<<"个请求,";
        cout<<"已处理"<<total_handle<<"个请求\n";
        cout<<"处理一个请求平均需要"<<total_time/total_handle<<"秒,";
        cout<<"平均一秒处理"<<now_rate<<"个请求\n";
        cout<<"峰值性能为一秒处理"<<top_speed<<"个请求";
        cout<<"---------------------------------\n";
    }   
}

服务器主函数serverV2.cpp

#include "global.h"
#include "HandleServerV2.h"
using namespace std;

//#define MAXLINE 1000  
//#define OPEN_MAX 100

//listen的backlog大小
#define LISTENQ 200
//监听端口号
#define SERV_PORT 8000
#define INFTIM 1000 

//extern void handle_all_request(string epoll_str,int conn_num,int epollfd);
extern void* handle_all_request(void *arg);
extern unordered_map<string,int> name_sock_map;//记录名字和套接字描述符
//extern clock_t begin_clock;//开始时间，用于性能测试，有bug
extern double total_time;//线程池处理任务的总时间
//extern time_point<system_clock> begin_clock;//开始时间，压力测试
extern int total_handle;//总处理请求数，用于性能测试
extern int total_recv_request;//接收到的请求总数，性能测试
extern int Bloom_Filter_bitmap[1000000];//布隆过滤器所用的bitmap
extern queue<int> mission_queue;//任务队列
extern int mission_num;//任务队列中的任务数量
extern pthread_cond_t mission_cond;//线程池所需的条件变量
extern pthread_spinlock_t name_mutex;//互斥锁，锁住需要修改name_sock_map的临界区
extern pthread_spinlock_t from_mutex;//互斥锁，锁住修改from_to_map的临界区
extern pthread_spinlock_t group_mutex;//互斥锁，锁住修改group_map的临界区
extern pthread_mutex_t queue_mutex;//互斥锁，锁住修改任务队列的临界区
extern int epollfd;
extern pthread_spinlock_t count_mutex;

//将参数的文件描述符设为非阻塞
void setnonblocking(int sock)  
{  
    int opts;  
    opts=fcntl(sock,F_GETFL);  
    if(opts<0)  
    {  
        perror("fcntl(sock,GETFL)");  
        exit(1);  
    }  
    opts = opts|O_NONBLOCK;  
    if(fcntl(sock,F_SETFL,opts)<0)  
    {  
        perror("fcntl(sock,SETFL,opts)");  
        exit(1);  
    }      
}  

int main(){
    pthread_spin_init(&name_mutex, 0); //创建互斥锁
    pthread_spin_init(&group_mutex,0);//创建互斥锁
    pthread_mutex_init(&queue_mutex,0);
    pthread_spin_init(&count_mutex,0);
    pthread_spin_init(&from_mutex,0);
    pthread_cond_init(&mission_cond,NULL);//初始化条件变量

    int i, maxi, listenfd, connfd, sockfd,epfd,nfds;  
    ssize_t n;  
    //char line[MAXLINE];  
    socklen_t clilen;  
    //声明epoll_event结构体的变量,ev用于注册事件,数组用于回传要处理的事件  
    struct epoll_event ev,events[10000];  
    //生成用于处理accept的epoll专用的文件描述符  
    epfd=epoll_create(10000);
    epollfd=epfd;
    struct sockaddr_in clientaddr;  
    struct sockaddr_in serveraddr;  
    listenfd = socket(PF_INET, SOCK_STREAM, 0);  
    //把socket设置为非阻塞方式  
    setnonblocking(listenfd);  
    //设置与要处理的事件相关的文件描述符  
    ev.data.fd=listenfd;  
    //设置要处理的事件类型  
    ev.events=EPOLLIN|EPOLLET;  
    //注册epoll事件  
    epoll_ctl(epfd,EPOLL_CTL_ADD,listenfd,&ev); //相当于Select模型的FD_SET  
    //设置serveraddr
    bzero(&serveraddr, sizeof(serveraddr));  
    serveraddr.sin_family = AF_INET;  
    serveraddr.sin_addr.s_addr = inet_addr("192.168.3.202");//此处设为服务器的ip
    serveraddr.sin_port=htons(8023);  
    bind(listenfd,(sockaddr *)&serveraddr, sizeof(serveraddr));  
    listen(listenfd, LISTENQ); 
    clilen=sizeof(clientaddr);
    maxi = 0;   
    
    cout<<"准备连数据库\n";

    //连接MYSQL数据库
    MYSQL *con=mysql_init(NULL);
    mysql_real_connect(con,"127.0.0.1","root","","test_connect",0,NULL,CLIENT_MULTI_STATEMENTS);
    string search="SELECT * FROM user;";
    auto search_res=mysql_query(con,search.c_str());
    auto result=mysql_store_result(con);
    int row;
    if(result)
        row=mysql_num_rows(result);//获取行数

    cout<<"连接数据库成功\n准备初始化布隆过滤器\n";

    //读取数据并完成布隆过滤器初始化
    memset(Bloom_Filter_bitmap,0,sizeof(Bloom_Filter_bitmap));
    for(int i=0;i<row;i++){
        auto info=mysql_fetch_row(result);//获取一行的信息
        string read_name=info[0];
        //对字符串使用哈希函数
        int hash=0;
        //cout<<"字符串："<<read_name;
        for(auto ch:read_name){
            hash=hash*131+ch;
            if(hash>=10000000)
                hash%=10000000;
        }
        hash%=32000000;
        //cout<<",hash值为："<<hash;
        //调整bitmap
        int index=hash/32,pos=hash%32;
        //cout<<index<<" "<<pos<<endl;
        Bloom_Filter_bitmap[index]|=(1<<pos);
        //cout<<",调整后的："<<Bloom_Filter_bitmap[index]<<endl;
    }
    mysql_close(con);
    int one=0,zero=0;
    for(auto i:Bloom_Filter_bitmap){
        int b=1;
        for(int j=1;j<=32;j++){
            if(b&i)
                one++;
            else
                zero++;
            b=(b<<1);
        }
    }
    cout<<"布隆过滤器中共有"<<one<<"位被置为1，其余"<<zero<<"位仍为0"<<endl;

    /* 定义一个10线程的线程池 */
    //boost::asio::thread_pool tp(10);

    //10个线程的线程池
    pthread_t tid[10];
    for(int i=0;i<10;i++)
        pthread_create(&tid[i],NULL,handle_all_request,NULL);

    //压力测试
    total_time=0;
    total_handle=0;
    total_recv_request=0;

    while(1){  
        cout<<"--------------------------"<<endl;
        cout<<"epoll_wait阻塞中"<<endl;
        //等待epoll事件的发生  
        nfds=epoll_wait(epfd,events,10000,-1);//最后一个参数是timeout，0:立即返回，-1:一直阻塞直到有事件，x:等待x毫秒
        cout<<"epoll_wait返回，有事件发生"<<endl;
        //处理所发生的所有事件  
        for(i=0;i<nfds;++i)  
        {  
            //有新客户端连接服务器
            if(events[i].data.fd==listenfd) 
            {  
                connfd = accept(listenfd,(sockaddr *)&clientaddr, &clilen);  
                if(connfd<0){  
                     perror("connfd<0");  
                     exit(1);  
                }  
                else{
                    cout<<"用户"<<inet_ntoa(clientaddr.sin_addr)<<"正在连接\n";
                }
                //设置用于读操作的文件描述符  
                ev.data.fd=connfd;  
                //设置用于注册的读操作事件，采用ET边缘触发，为防止多个线程处理同一socket而使用EPOLLONESHOT  
                ev.events=EPOLLIN|EPOLLET|EPOLLONESHOT;
                //边缘触发要将套接字设为非阻塞
                setnonblocking(connfd);
                //注册ev  
                epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev);  
            }  
            //接收到读事件
            else if(events[i].events&EPOLLIN)  
            {  
                total_recv_request++;
                sockfd = events[i].data.fd;
                events[i].data.fd=-1;
                cout<<"接收到读事件"<<endl;
                
                pthread_mutex_lock(&queue_mutex);
                mission_queue.push(sockfd);//加入任务队列
                pthread_cond_broadcast(&mission_cond);//广播唤醒
                pthread_mutex_unlock(&queue_mutex);

                //string recv_str;
                //boost::asio::post(boost::bind(handle_all_request,recv_str,sockfd,epfd)); //加入任务队列，处理事件
            }  
        } 
    }  
    close(listenfd);
}