1、概述
池化技术，减少了资源创建次数，提高了程序响应性能，特别是在高并发场景下，当程序7*24小时运行，创建资源可能会出现耗时较长和失败等问题，池化技术，主要是程序初始化之前创建多个可用连接，集中管理起来，后续直接使用，使用完并归还。
2、线程池
线程池主要解决问题： 1、解决线程创建和消耗所浪费的资源和时间 2、控制线程数量，减少资源竞争
代码解读
一共由三个接口组成： 1、CreatePthreadPool创建工作线程 初始化线程池结构下的条件变量和互斥锁，创建多个线程通过头插法，插入到线程池结构worker成员中 2、thread_func线程入口函数 进入while循环，第一件事先拿锁，然后再去判断，任务队列中是否有数据，要是没有则进入循环中的条件等待，进入条件等待会先释放锁，所以初始化完成之后所有线程全部都会进入条件等待，然后一旦条件满足则会去拿锁。 3、threadPoolQueue将任务加入到任务链表中 先加锁，然后通过头插法插入到线程池结构中的wait_jobs成员中，通知条件变量，然后解锁 4、threadPoolShutdown回收线程池中线程 将工作线程中的标识位全部置为一，然后通知所有线程，然后等待回收所有线程并释放每个节点
总结
线程池就是一个很典型的生产者消费者模型，不断地向任务队列中投放任务，工作线程不断地去队伍队列中去取任务消费掉。
#include <stdio.h>
#include <unistd.h>
#include <pthread.h>
#include <stdlib.h>
#include <string.h>
#define LL_ADD(item, list) do { \
item->prev = NULL; \
item->next = list; \
list = item; \
} while(0)
#define LL_REMOVE(item,list)do{ \
if(item->prev != NULL) item->prev->next = item->next; \
if(item->next != NULL) item->next->prev = item->prev; \
if(list== item)list= item->next; \
item->prev = item->next = NULL; \
}while(0)
// 工作线程结构体
typedefstruct NWORKER{
pthread_t id;// 线程id
intterminate;// 终止条件
struct NTHREADPOLL *worqueue;
struct NWORKER *prev;// 前向指针
struct NWORKER *next;// 后项指针
}nWorker;
// 任务队列
typedefstruct NJOB{
void(*job_function)(struct NJOB* job);
void* user_data;
struct NJOB *prev;
struct NJOB *next;
}nJob;
typedefstruct NTHREADPOLL{
struct NWORKER *workers;
struct NJOB *wait_jobs;
pthread_cond_t cond;
pthread_mutex_t mtx;
}nThreadpoll;
// 操作线程池函数
#define MAX_PTHREAD 80
#define MAX_JOB 8
// 毁掉函数
voidjob_fun(nJob *job){
if(job ==NULL)return;
int*data =(int*)job->user_data;
printf("user_data:%d-thread_id:%ld,line:%d\n",*data,pthread_self(),__LINE__);
free(job->user_data);
job->user_data =NULL;
free(job);
job =NULL;
return;
}
// 线程入口函数
void*thread_func(void* data){
nWorker *worker =(nWorker*)data;
while(1)
{
pthread_mutex_lock(&(worker->worqueue->mtx));
while(worker->worqueue->wait_jobs ==NULL){
if(worker->terminate)break;
pthread_cond_wait(&(worker->worqueue->cond),&(worker->worqueue->mtx));
}
if(worker->terminate){
pthread_mutex_unlock(&(worker->worqueue->mtx));
break;
}
nJob *job = worker->worqueue->wait_jobs;
if(job !=NULL)
LL_REMOVE(job,worker->worqueue->wait_jobs);
pthread_mutex_unlock(&(worker->worqueue->mtx));
if(job ==NULL)continue;
job->job_function(job);
}
// free(worker);
// worker = NULL;
pthread_exit(0);
}
// 创造线程
intCreatePthreadPool(nThreadpoll* threadpoll,int max_pthread)
{
if(max_pthread <=0) max_pthread = MAX_PTHREAD;
memset(threadpoll,0x00,sizeof(nThreadpoll));
pthread_cond_t blank_cond = PTHREAD_COND_INITIALIZER;
memcpy(&threadpoll->cond,&blank_cond,sizeof(threadpoll->cond));
pthread_mutex_t blank_mutex = PTHREAD_MUTEX_INITIALIZER;
memcpy(&threadpoll->mtx,&blank_mutex,sizeof(threadpoll->mtx));
int i;
for(i=0; i<max_pthread; i++){
nWorker *worker =(nWorker*)malloc(sizeof(nWorker));
if(worker ==NULL){
perror("threadpoll->workers");
return-1;
}
memset(worker,0x00,sizeof(nWorker));
worker->worqueue = threadpoll;
worker->terminate=0;
int ret = pthread_create(&(worker->id),NULL,thread_func,(void*)worker);
if(ret !=0){
perror("pthread_create");
free(worker);
worker =NULL;
return-1;
}
LL_ADD(worker,worker->worqueue->workers);
}
return0;
}
// 向线程中添加任务，唤醒一个线程去处理
voidthreadPoolQueue(nThreadpoll* threadpool,nJob *job){
pthread_mutex_lock(&(threadpool->mtx));
LL_ADD(job,threadpool->wait_jobs);
pthread_cond_signal(&(threadpool->cond));
pthread_mutex_unlock(&(threadpool->mtx));
return;
}
voidthreadPoolShutdown(nThreadpoll *workerqueue){
nWorker *woker =NULL;
for(woker = workerqueue->workers;woker!=NULL;woker = woker->next){
woker->terminate=1;
}
pthread_cond_broadcast(&workerqueue->cond);
struct NWORKER *tmp = workerqueue->workers;
while(tmp)
{
pthread_join(tmp->id,NULL);
if(tmp){
free(tmp);
tmp =NULL;
}
tmp = tmp->next;
}
pthread_cond_destroy(&workerqueue->cond);
pthread_mutex_destroy(&workerqueue->mtx);
workerqueue->workers =NULL;
workerqueue->wait_jobs =NULL;
}
intmain(){
nThreadpoll workerQueue;
if(CreatePthreadPool(&workerQueue,MAX_PTHREAD)!=0){
perror("CreatePthreadPool faile");
return-1;
}
int i;
for(i=0;i<MAX_JOB;i++){
nJob *job =(nJob*)malloc(sizeof(nJob));
if(job ==NULL){
perror("nJob malloc faile");
exit(-1);
}
job->job_function = job_fun;
job->user_data =(int*)malloc(sizeof(int));
*(int*)job->user_data = i;
threadPoolQueue(&workerQueue,job);
}
getchar();
return0;
}
2、异步请求池
当我们向数据库或者DNS服务器发送请求时，需要同步等待他的返回结果，这个过程是比较耗时的，所以衍生出异步发起请求，当发出请求之后，不等待请求的响应，而是通过一个线程去监听，主线程去做其他事情，当请求响应之后如何去做其他事情。
同步向DNS服务器发送请求
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <netdb.h>
#include <arpa/inet.h>
#include <pthread.h>
#define DNS_SVR "114.114.114.114"
#define DNS_HOST 0x01
#define DNS_CNAME 0x05
structdns_header{
unsignedshort id;
unsignedshort flags;
unsignedshort qdcount;
unsignedshort ancount;
unsignedshort nscount;
unsignedshort arcount;
};
structdns_question{
int length;
unsignedshort qtype;
unsignedshort qclass;
char*qname;
};
structdns_item{
char*domain;
char*ip;
};
int dns_create_header(struct dns_header *header) {
if(header ==NULL)return-1;
memset(header,0,sizeof(struct dns_header));
srandom(time(NULL));
header->id =random();
header->flags |=htons(0x0100);
header->qdcount =htons(1);
return0;
}
int dns_create_question(struct dns_question *question, const char *hostname) {
if(question ==NULL)return-1;
memset(question,0,sizeof(struct dns_question));
question->qname =(char*)malloc(strlen(hostname)+2);
if(question->qname ==NULL)return-2;
question->length =strlen(hostname)+2;
question->qtype =htons(1);
question->qclass =htons(1);
constchar delim[2]=".";
char*hostname_dup =strdup(hostname);
char*token =strtok(hostname_dup, delim);
char*qname_p = question->qname;
while(token !=NULL){
size_t len =strlen(token);
*qname_p = len;
qname_p ++;
strncpy(qname_p, token, len+1);
qname_p += len;
token =strtok(NULL, delim);
}
free(hostname_dup);
return0;
}
int dns_build_request(struct dns_header *header, struct dns_question *question, char *request) {
int header_s =sizeof(struct dns_header);
int question_s = question->length +sizeof(question->qtype)+sizeof(question->qclass);
int length = question_s + header_s;
int offset =0;
memcpy(request+offset, header,sizeof(struct dns_header));
offset +=sizeof(struct dns_header);
memcpy(request+offset, question->qname, question->length);
offset += question->length;
memcpy(request+offset,&question->qtype,sizeof(question->qtype));
offset +=sizeof(question->qtype);
memcpy(request+offset,&question->qclass,sizeof(question->qclass));
return length;
}
static int is_pointer(int in) {
return((in &0xC0)==0xC0);
}
static void dns_parse_name(unsigned char *chunk, unsigned char *ptr, char *out, int *len) {
int flag =0, n =0, alen =0;
char*pos = out +(*len);
while(1){
flag =(int)ptr[0];
if(flag ==0)break;
if(is_pointer(flag)){
n =(int)ptr[1];
ptr = chunk + n;
dns_parse_name(chunk, ptr, out, len);
break;
}else{
ptr ++;
memcpy(pos, ptr, flag);
pos += flag;
ptr += flag;
*len += flag;
if((int)ptr[0]!=0){
memcpy(pos,".",1);
pos +=1;
(*len)+=1;
}
}
}
}
static int dns_parse_response(char *buffer, struct dns_item **domains) {
int i =0;
unsignedchar*ptr = buffer;
ptr +=4;
int querys =ntohs(*(unsignedshort*)ptr);
ptr +=2;
int answers =ntohs(*(unsignedshort*)ptr);
ptr +=6;
for(i =0;i < querys;i ++){
while(1){
int flag =(int)ptr[0];
ptr +=(flag +1);
if(flag ==0)break;
}
ptr +=4;
}
char cname[128], aname[128], ip[20], netip[4];
int len, type, ttl, datalen;
int cnt =0;
structdns_item*list=(struct dns_item*)calloc(answers,sizeof(struct dns_item));
if(list==NULL){
return-1;
}
for(i =0;i < answers;i ++){
bzero(aname,sizeof(aname));
len =0;
dns_parse_name(buffer, ptr, aname,&len);
ptr +=2;
type =htons(*(unsignedshort*)ptr);
ptr +=4;
ttl =htons(*(unsignedshort*)ptr);
ptr +=4;
datalen =ntohs(*(unsignedshort*)ptr);
ptr +=2;
if(type == DNS_CNAME){
bzero(cname,sizeof(cname));
len =0;
dns_parse_name(buffer, ptr, cname,&len);
ptr += datalen;
}elseif(type == DNS_HOST){
bzero(ip,sizeof(ip));
if(datalen ==4){
memcpy(netip, ptr, datalen);
inet_ntop(AF_INET , netip , ip ,sizeof(struct sockaddr));
printf("%s has address %s\n", aname, ip);
printf("\tTime to live: %d minutes , %d seconds\n", ttl /60, ttl %60);
list[cnt].domain =(char*)calloc(strlen(aname)+1,1);
memcpy(list[cnt].domain, aname,strlen(aname));
list[cnt].ip =(char*)calloc(strlen(ip)+1,1);
memcpy(list[cnt].ip, ip,strlen(ip));
cnt ++;
}
ptr += datalen;
}
}
*domains =list;
ptr +=2;
return cnt;
}
int dns_client_commit(const char *domain) {
int sockfd =socket(AF_INET, SOCK_DGRAM,0);
if(sockfd <0){
perror("create socket failed\n");
exit(-1);
}
printf("url:%s\n", domain);
structsockaddr_in dest;
bzero(&dest,sizeof(dest));
dest.sin_family = AF_INET;
dest.sin_port =htons(53);
dest.sin_addr.s_addr =inet_addr(DNS_SVR);
int ret =connect(sockfd,(struct sockaddr*)&dest,sizeof(dest));
printf("connect :%d\n", ret);
structdns_header header ={0};
dns_create_header(&header);
structdns_question question ={0};
dns_create_question(&question, domain);
char request[1024]={0};
int req_len =dns_build_request(&header,&question, request);
int slen =sendto(sockfd, request, req_len,0,(struct sockaddr*)&dest,sizeof(struct sockaddr));
char buffer[1024]={0};
structsockaddr_in addr;
size_t addr_len =sizeof(struct sockaddr_in);
int n =recvfrom(sockfd, buffer,sizeof(buffer),0,(struct sockaddr*)&addr,(socklen_t*)&addr_len);
printf("recvfrom n : %d\n", n);
structdns_item*domains =NULL;
dns_parse_response(buffer,&domains);
return0;
}
char*domain[]={
// "www.ntytcp.com",
"bojing.wang",
"",
"tieba.baidu.com",
"news.baidu.com",
"zhidao.baidu.com",
"music.baidu.com",
"image.baidu.com",
"v.baidu.com",
"map.baidu.com",
"baijiahao.baidu.com",
"xueshu.baidu.com",
"cloud.baidu.com",
"",
"open.163.com",
"auto.163.com",
"gov.163.com",
"money.163.com",
"sports.163.com",
"tech.163.com",
"edu.163.com",
"www.taobao.com",
"q.taobao.com",
"sf.taobao.com",
"yun.taobao.com",
"baoxian.taobao.com",
"www.tmall.com",
"suning.tmall.com",
"www.tencent.com",
"www.qq.com",
"",
"www.ctrip.com",
"hotels.ctrip.com",
"hotels.ctrip.com",
"vacations.ctrip.com",
"flights.ctrip.com",
"trains.ctrip.com",
"bus.ctrip.com",
"car.ctrip.com",
"piao.ctrip.com",
"tuan.ctrip.com",
"you.ctrip.com",
"g.ctrip.com",
"lipin.ctrip.com",
"ct.ctrip.com"
};
int main(int argc, char *argv[]) {
int count =sizeof(domain)/sizeof(domain[0]);
int i =0;
for(i =0;i < count;i ++){
dns_client_commit(domain[i]);
}
getchar();
}异步发送请求
1、dns_async_client_init初始化函数 创建epoll、创建线程，用来监控dns的响应消息 2、dns_async_client_commit创建连接，发送请求，将fd加入到epoll中进行监听 3、dns_async_client_proc线程入口函数 通过epoll_wait进行监听，调用取出数据，调用对应的回调函数，epoll_ctl删除节点。
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <netdb.h>
#include <arpa/inet.h>
#include <pthread.h>
#define DNS_SVR "114.114.114.114"
#define DNS_HOST 0x01
#define DNS_CNAME 0x05
#define ASYNC_CLIENT_NUM 1024
structdns_header{
unsignedshort id;
unsignedshort flags;
unsignedshort qdcount;
unsignedshort ancount;
unsignedshort nscount;
unsignedshort arcount;
};
structdns_question{
int length;
unsignedshort qtype;
unsignedshort qclass;
char*qname;
};
structdns_item{
char*domain;
char*ip;
};
typedef void (*async_result_cb)(struct dns_item *list, int count);
structasync_context{
int epfd;
};
structep_arg{
int sockfd;
async_result_cb cb;
};
int dns_create_header(struct dns_header *header) {
if(header ==NULL)return-1;
memset(header,0,sizeof(struct dns_header));
srandom(time(NULL));
header->id =random();
header->flags |=htons(0x0100);
header->qdcount =htons(1);
return0;
}
int dns_create_question(struct dns_question *question, const char *hostname) {
if(question ==NULL)return-1;
memset(question,0,sizeof(struct dns_question));
question->qname =(char*)malloc(strlen(hostname)+2);
if(question->qname ==NULL)return-2;
question->length =strlen(hostname)+2;
question->qtype =htons(1);
question->qclass =htons(1);
constchar delim[2]=".";
char*hostname_dup =strdup(hostname);
char*token =strtok(hostname_dup, delim);
char*qname_p = question->qname;
while(token !=NULL){
size_t len =strlen(token);
*qname_p = len;
qname_p ++;
strncpy(qname_p, token, len+1);
qname_p += len;
token =strtok(NULL, delim);
}
free(hostname_dup);
return0;
}
int dns_build_request(struct dns_header *header, struct dns_question *question, char *request) {
int header_s =sizeof(struct dns_header);
int question_s = question->length +sizeof(question->qtype)+sizeof(question->qclass);
int length = question_s + header_s;
int offset =0;
memcpy(request+offset, header,sizeof(struct dns_header));
offset +=sizeof(struct dns_header);
memcpy(request+offset, question->qname, question->length);
offset += question->length;
memcpy(request+offset,&question->qtype,sizeof(question->qtype));
offset +=sizeof(question->qtype);
memcpy(request+offset,&question->qclass,sizeof(question->qclass));
return length;
}
static int is_pointer(int in) {
return((in &0xC0)==0xC0);
}
static int set_block(int fd, int block) {
int flags =fcntl(fd, F_GETFL,0);
if(flags <0)return flags;
if(block){
flags &=~O_NONBLOCK;
}else{
flags |= O_NONBLOCK;
}
if(fcntl(fd, F_SETFL, flags)<0)return-1;
return0;
}
static void dns_parse_name(unsigned char *chunk, unsigned char *ptr, char *out, int *len) {
int flag =0, n =0, alen =0;
char*pos = out +(*len);
while(1){
flag =(int)ptr[0];
if(flag ==0)break;
if(is_pointer(flag)){
n =(int)ptr[1];
ptr = chunk + n;
dns_parse_name(chunk, ptr, out, len);
break;
}else{
ptr ++;
memcpy(pos, ptr, flag);
pos += flag;
ptr += flag;
*len += flag;
if((int)ptr[0]!=0){
memcpy(pos,".",1);
pos +=1;
(*len)+=1;
}
}
}
}
static int dns_parse_response(char *buffer, struct dns_item **domains) {
int i =0;
unsignedchar*ptr = buffer;
ptr +=4;
int querys =ntohs(*(unsignedshort*)ptr);
ptr +=2;
int answers =ntohs(*(unsignedshort*)ptr);
ptr +=6;
for(i =0;i < querys;i ++){
while(1){
int flag =(int)ptr[0];
ptr +=(flag +1);
if(flag ==0)break;
}
ptr +=4;
}
char cname[128], aname[128], ip[20], netip[4];
int len, type, ttl, datalen;
int cnt =0;
structdns_item*list=(struct dns_item*)calloc(answers,sizeof(struct dns_item));
if(list==NULL){
return-1;
}
for(i =0;i < answers;i ++){
bzero(aname,sizeof(aname));
len =0;
dns_parse_name(buffer, ptr, aname,&len);
ptr +=2;
type =htons(*(unsignedshort*)ptr);
ptr +=4;
ttl =htons(*(unsignedshort*)ptr);
ptr +=4;
datalen =ntohs(*(unsignedshort*)ptr);
ptr +=2;
if(type == DNS_CNAME){
bzero(cname,sizeof(cname));
len =0;
dns_parse_name(buffer, ptr, cname,&len);
ptr += datalen;
}elseif(type == DNS_HOST){
bzero(ip,sizeof(ip));
if(datalen ==4){
memcpy(netip, ptr, datalen);
inet_ntop(AF_INET , netip , ip ,sizeof(struct sockaddr));
printf("%s has address %s\n", aname, ip);
printf("\tTime to live: %d minutes , %d seconds\n", ttl /60, ttl %60);
list[cnt].domain =(char*)calloc(strlen(aname)+1,1);
memcpy(list[cnt].domain, aname,strlen(aname));
list[cnt].ip =(char*)calloc(strlen(ip)+1,1);
memcpy(list[cnt].ip, ip,strlen(ip));
cnt ++;
}
ptr += datalen;
}
}
*domains =list;
ptr +=2;
return cnt;
}
int dns_client_commit(const char *domain) {
int sockfd =socket(AF_INET, SOCK_DGRAM,0);
if(sockfd <0){
perror("create socket failed\n");
exit(-1);
}
printf("url:%s\n", domain);
set_block(sockfd,0);//nonblock
structsockaddr_in dest;
bzero(&dest,sizeof(dest));
dest.sin_family = AF_INET;
dest.sin_port =htons(53);
dest.sin_addr.s_addr =inet_addr(DNS_SVR);
int ret =connect(sockfd,(struct sockaddr*)&dest,sizeof(dest));
//printf("connect :%d\n", ret);
structdns_header header ={0};
dns_create_header(&header);
structdns_question question ={0};
dns_create_question(&question, domain);
char request[1024]={0};
int req_len =dns_build_request(&header,&question, request);
int slen =sendto(sockfd, request, req_len,0,(struct sockaddr*)&dest,sizeof(struct sockaddr));
while(1){
char buffer[1024]={0};
structsockaddr_in addr;
size_t addr_len =sizeof(struct sockaddr_in);
int n =recvfrom(sockfd, buffer,sizeof(buffer),0,(struct sockaddr*)&addr,(socklen_t*)&addr_len);
if(n <=0)continue;
printf("recvfrom n : %d\n", n);
structdns_item*domains =NULL;
dns_parse_response(buffer,&domains);
break;
}
return0;
}
void dns_async_client_free_domains(struct dns_item *list, int count) {
int i =0;
for(i =0;i < count;i ++){
free(list[i].domain);
free(list[i].ip);
}
free(list);
}
//dns_async_client_proc()
//epoll_wait
//result callback
static void* dns_async_client_proc(void *arg) {
structasync_context*ctx =(struct async_context*)arg;
int epfd = ctx->epfd;
while(1){
structepoll_event events[ASYNC_CLIENT_NUM]={0};
int nready =epoll_wait(epfd, events, ASYNC_CLIENT_NUM,-1);
if(nready <0){
if(errno == EINTR || errno == EAGAIN){
continue;
}else{
break;
}
}elseif(nready ==0){
continue;
}
printf("nready:%d\n", nready);
int i =0;
for(i =0;i < nready;i ++){
structep_arg*data =(struct ep_arg*)events[i].data.ptr;
int sockfd = data->sockfd;
char buffer[1024]={0};
structsockaddr_in addr;
size_t addr_len =sizeof(struct sockaddr_in);
int n =recvfrom(sockfd, buffer,sizeof(buffer),0,(struct sockaddr*)&addr,(socklen_t*)&addr_len);
structdns_item*domain_list =NULL;
int count =dns_parse_response(buffer,&domain_list);
data->cb(domain_list, count);//call cb
int ret =epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd,NULL);
//printf("epoll_ctl DEL --> sockfd:%d\n", sockfd);
close(sockfd);/////
dns_async_client_free_domains(domain_list, count);
free(data);
}
}
}
//dns_async_client_init()
//epoll init
//thread init
structasync_context*dns_async_client_init(void){
int epfd =epoll_create(1);//
if(epfd <0)returnNULL;
structasync_context*ctx =calloc(1,sizeof(struct async_context));
if(ctx ==NULL){
close(epfd);
returnNULL;
}
ctx->epfd = epfd;
pthread_t thread_id;
int ret =pthread_create(&thread_id,NULL, dns_async_client_proc, ctx);
if(ret){
perror("pthread_create");
returnNULL;
}
usleep(1);//child go first
return ctx;
}
//dns_async_client_commit(ctx, domain)
//socket init
//dns_request
//sendto dns send
int dns_async_client_commit(struct async_context* ctx, const char *domain, async_result_cb cb) {
int sockfd =socket(AF_INET, SOCK_DGRAM,0);
if(sockfd <0){
perror("create socket failed\n");
exit(-1);
}
printf("url:%s\n", domain);
set_block(sockfd,0);//nonblock
structsockaddr_in dest;
bzero(&dest,sizeof(dest));
dest.sin_family = AF_INET;
dest.sin_port =htons(53);
dest.sin_addr.s_addr =inet_addr(DNS_SVR);
int ret =connect(sockfd,(struct sockaddr*)&dest,sizeof(dest));
//printf("connect :%d\n", ret);
structdns_header header ={0};
dns_create_header(&header);
structdns_question question ={0};
dns_create_question(&question, domain);
char request[1024]={0};
int req_len =dns_build_request(&header,&question, request);
int slen =sendto(sockfd, request, req_len,0,(struct sockaddr*)&dest,sizeof(struct sockaddr));
structep_arg*eparg =(struct ep_arg*)calloc(1,sizeof(struct ep_arg));
if(eparg ==NULL)return-1;
eparg->sockfd = sockfd;
eparg->cb = cb;
structepoll_event ev;
ev.data.ptr = eparg;
ev.events = EPOLLIN;
ret =epoll_ctl(ctx->epfd, EPOLL_CTL_ADD, sockfd,&ev);
//printf(" epoll_ctl ADD: sockfd->%d, ret:%d\n", sockfd, ret);
return ret;
}
char*domain[]={
"www.ntytcp.com",
"bojing.wang",
"",
"tieba.baidu.com",
"news.baidu.com",
"zhidao.baidu.com",
"music.baidu.com",
"image.baidu.com",
"v.baidu.com",
"map.baidu.com",
"baijiahao.baidu.com",
"xueshu.baidu.com",
"cloud.baidu.com",
"",
"open.163.com",
"auto.163.com",
"gov.163.com",
"money.163.com",
"sports.163.com",
"tech.163.com",
"edu.163.com",
"www.taobao.com",
"q.taobao.com",
"sf.taobao.com",
"yun.taobao.com",
"baoxian.taobao.com",
"www.tmall.com",
"suning.tmall.com",
"www.tencent.com",
"www.qq.com",
"",
"www.ctrip.com",
"hotels.ctrip.com",
"hotels.ctrip.com",
"vacations.ctrip.com",
"flights.ctrip.com",
"trains.ctrip.com",
"bus.ctrip.com",
"car.ctrip.com",
"piao.ctrip.com",
"tuan.ctrip.com",
"you.ctrip.com",
"g.ctrip.com",
"lipin.ctrip.com",
"ct.ctrip.com"
};
static void dns_async_client_result_callback(struct dns_item *list, int count) {
int i =0;
for(i =0;i < count;i ++){
printf("name:%s, ip:%s\n",list[i].domain,list[i].ip);
}
}
int main(int argc, char *argv[]) {
#if 0
dns_client_commit(argv[1]);
#else
structasync_context*ctx =dns_async_client_init();
if(ctx ==NULL)return-2;
int count =sizeof(domain)/sizeof(domain[0]);
int i =0;
for(i =0;i < count;i ++){
dns_async_client_commit(ctx, domain[i], dns_async_client_result_callback);
//sleep(2);
}
getchar();
#endif
}
3、内存池
内存池主要是解决，频繁开辟和释放内存，造成的内存碎片问题，大量的内存碎片，会导致我们分配内存失败。 内存池是通过，申请比较大的内存块，后面慢慢的去使用他，避免重复申请和释放内存带来的效率消耗和大量内存碎片
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#define MP_ALIGNMENT 32
#define MP_PAGE_SIZE 4096
#define MP_MAX_ALLOC_FROM_POOL (MP_PAGE_SIZE-1)
#define mp_align(n, alignment) (((n)+(alignment-1)) & ~(alignment-1))
#define mp_align_ptr(p, alignment) (void *)((((size_t)p)+(alignment-1)) & ~(alignment-1))
structmp_large_s{
structmp_large_s*next;
void*alloc;
};
structmp_node_s{
unsignedchar*last;
unsignedchar*end;
structmp_node_s*next;
size_t failed;
};
structmp_pool_s{
size_t max;
structmp_node_s*current;
structmp_large_s*large;
structmp_node_s head[0];
};
structmp_pool_s*mp_create_pool(size_t size);
void mp_destory_pool(struct mp_pool_s *pool);
void *mp_alloc(struct mp_pool_s *pool, size_t size);
void *mp_nalloc(struct mp_pool_s *pool, size_t size);
void *mp_calloc(struct mp_pool_s *pool, size_t size);
void mp_free(struct mp_pool_s *pool, void *p);
structmp_pool_s*mp_create_pool(size_t size){
structmp_pool_s*p;
int ret =posix_memalign((void**)&p, MP_ALIGNMENT, size +sizeof(struct mp_pool_s)+sizeof(struct mp_node_s));
if(ret){
returnNULL;
}
p->max =(size < MP_MAX_ALLOC_FROM_POOL)? size : MP_MAX_ALLOC_FROM_POOL;
p->current = p->head;
p->large =NULL;
p->head->last =(unsignedchar*)p +sizeof(struct mp_pool_s)+sizeof(struct mp_node_s);
p->head->end = p->head->last + size;
p->head->failed =0;
return p;
}
void mp_destory_pool(struct mp_pool_s *pool) {
structmp_node_s*h,*n;
structmp_large_s*l;
for(l = pool->large; l; l = l->next){
if(l->alloc){
free(l->alloc);
}
}
h = pool->head->next;
while(h){
n = h->next;
free(h);
h = n;
}
free(pool);
}
void mp_reset_pool(struct mp_pool_s *pool) {
structmp_node_s*h;
structmp_large_s*l;
for(l = pool->large; l; l = l->next){
if(l->alloc){
free(l->alloc);
}
}
pool->large =NULL;
for(h = pool->head; h; h = h->next){
h->last =(unsignedchar*)h +sizeof(struct mp_node_s);
}
}
static void *mp_alloc_block(struct mp_pool_s *pool, size_t size) {
unsignedchar*m;
structmp_node_s*h = pool->head;
size_t psize =(size_t)(h->end -(unsignedchar*)h);
int ret =posix_memalign((void**)&m, MP_ALIGNMENT, psize);
if(ret)returnNULL;
structmp_node_s*p,*new_node,*current;
new_node =(struct mp_node_s*)m;
new_node->end = m + psize;
new_node->next =NULL;
new_node->failed =0;
m +=sizeof(struct mp_node_s);
m =mp_align_ptr(m, MP_ALIGNMENT);
new_node->last = m + size;
current = pool->current;
for(p = current; p->next; p = p->next){
if(p->failed++>4){
current = p->next;
}
}
p->next = new_node;
pool->current = current ? current : new_node;
return m;
}
static void *mp_alloc_large(struct mp_pool_s *pool, size_t size) {
void*p =malloc(size);
if(p ==NULL)returnNULL;
size_t n =0;
structmp_large_s*large;
for(large = pool->large; large; large = large->next){
if(large->alloc ==NULL){
large->alloc = p;
return p;
}
if(n ++>3)break;
}
large =mp_alloc(pool,sizeof(struct mp_large_s));
if(large ==NULL){
free(p);
returnNULL;
}
large->alloc = p;
large->next = pool->large;
pool->large = large;
return p;
}
void *mp_memalign(struct mp_pool_s *pool, size_t size, size_t alignment) {
void*p;
int ret =posix_memalign(&p, alignment, size);
if(ret){
returnNULL;
}
structmp_large_s*large =mp_alloc(pool,sizeof(struct mp_large_s));
if(large ==NULL){
free(p);
returnNULL;
}
large->alloc = p;
large->next = pool->large;
pool->large = large;
return p;
}
void *mp_alloc(struct mp_pool_s *pool, size_t size) {
unsignedchar*m;
structmp_node_s*p;
if(size <= pool->max){
p = pool->current;
do{
m =mp_align_ptr(p->last, MP_ALIGNMENT);
if((size_t)(p->end - m)>= size){
p->last = m + size;
return m;
}
p = p->next;
}while(p);
returnmp_alloc_block(pool, size);
}
returnmp_alloc_large(pool, size);
}
void *mp_nalloc(struct mp_pool_s *pool, size_t size) {
unsignedchar*m;
structmp_node_s*p;
if(size <= pool->max){
p = pool->current;
do{
m = p->last;
if((size_t)(p->end - m)>= size){
p->last = m+size;
return m;
}
p = p->next;
}while(p);
returnmp_alloc_block(pool, size);
}
returnmp_alloc_large(pool, size);
}
void *mp_calloc(struct mp_pool_s *pool, size_t size) {
void*p =mp_alloc(pool, size);
if(p){
memset(p,0, size);
}
return p;
}
void mp_free(struct mp_pool_s *pool, void *p) {
structmp_large_s*l;
for(l = pool->large; l; l = l->next){
if(p == l->alloc){
free(l->alloc);
l->alloc =NULL;
return;
}
}
}
int main(int argc, char *argv[]) {
int size =1<<12;
structmp_pool_s*p =mp_create_pool(size);
int i =0;
for(i =0;i <10;i ++){
void*mp =mp_alloc(p,512);
// mp_free(mp);
}
//printf("mp_create_pool: %ld\n", p->max);
printf("mp_align(123, 32): %d, mp_align(17, 32): %d\n",mp_align(24,32),mp_align(17,32));
//printf("mp_align_ptr(p->current, 32): %lx, p->current: %lx, mp_align(p->large, 32): %lx, p->large: %lx\n", mp_align_ptr(p->current, 32), p->current, mp_align_ptr(p->large, 32), p->large);
int j =0;
for(i =0;i <5;i ++){
char*pp =mp_calloc(p,32);
for(j =0;j <32;j ++){
if(pp[j]){
printf("calloc wrong\n");
}
printf("calloc success\n");
}
}
//printf("mp_reset_pool\n");
for(i =0;i <5;i ++){
void*l =mp_alloc(p,8192);
mp_free(p, l);
}
mp_reset_pool(p);
//printf("mp_destory_pool\n");
for(i =0;i <58;i ++){
mp_alloc(p,256);
}
mp_destory_pool(p);
return0;
}
4、mysql连接池
池化技术能够减少资源创建和释放次数，提高程序的响应性能，特别是在高并发下这种提高更加明显。 1、对象创建时间长。 2、对象创建需要大量资源。 3、对象创建后可被重复使用
4.1、什么是数据库连接池
数据库连接池，是程序启动建立足够多的数据库连接，并将这些连接组成一个连接池，由程序动态地申请和释放
4.2、 连接池和线程池的关系
线程池是主动执行任务，连接池被动操作，池对象被任务获取，任务执行完后归还。 连接池和线程池设置数量的关系： 1、一般线程池线程数量和连接池连接对象数量一致； 2、一般线程执行任务完毕的时候归还连接对象；
头文件DBPool.h
CDBConn类主要是操作mysql数据库： 1、Init从连接池类中获取mysql信息，进行数据库连接 2、ExecuteCreate执行sql语句 3、ExecuteDrop执行sql语句 4、ExecuteQuery执行查询语句 CDBPool类，主要是管理队列中的CDBConn类： 1、构造函数中初始化变量 2、Init创建最小连接数量CDBConn对象存放到队列中m_free_list 3、GetDBConn从连接池中获取一个连接 4、RelDBConn归还连接到队列中
#ifndef DBPOOL_H_
#define DBPOOL_H_
#include <iostream>
#include <list>
#include <mutex>
#include <condition_variable>
#include <map>
#include <stdint.h>
#include <mysql.h>
#define MAX_ESCAPE_STRING_LEN 10240
usingnamespacestd;
// 返回结果 select的时候用
class CResultSet {
public:
CResultSet(MYSQL_RES* res);
virtual~CResultSet();
boolNext();
intGetInt(const char* key);
char*GetString(const char* key);
private:
int_GetIndex(constchar* key);
MYSQL_RES* m_res;
MYSQL_ROW m_row;
map<string,int> m_key_map;
};
// 插入数据用
class CPrepareStatement {
public:
CPrepareStatement();
virtual~CPrepareStatement();
boolInit(MYSQL* mysql, string& sql);
voidSetParam(uint32_t index, int& value);
voidSetParam(uint32_t index, uint32_t& value);
voidSetParam(uint32_t index, string& value);
voidSetParam(uint32_t index, const string& value);
boolExecuteUpdate();
uint32_tGetInsertId();
private:
MYSQL_STMT* m_stmt;
MYSQL_BIND* m_param_bind;
uint32_t m_param_cnt;
};
class CDBPool;
class CDBConn {
public:
CDBConn(CDBPool* pDBPool);
virtual~CDBConn();
intInit();
// 创建表
boolExecuteCreate(const char* sql_query);
// 删除表
boolExecuteDrop(const char* sql_query);
// 查询
CResultSet*ExecuteQuery(const char* sql_query);
/**
* 执行DB更新，修改
*
* @param sql_query sql
* @param care_affected_rows 是否在意影响的行数，false:不在意；true:在意
*
* @return 成功返回true 失败返回false
*/
boolExecuteUpdate(const char* sql_query, bool care_affected_rows = true);
uint32_tGetInsertId();
// 开启事务
boolStartTransaction();
// 提交事务
boolCommit();
// 回滚事务
boolRollback();
// 获取连接池名
constchar*GetPoolName();
MYSQL*GetMysql(){return m_mysql;}
private:
CDBPool* m_pDBPool;// to get MySQL server information
MYSQL* m_mysql;// 对应一个连接
char m_escape_string[MAX_ESCAPE_STRING_LEN +1];
};
class CDBPool {// 只是负责管理连接CDBConn，真正干活的是CDBConn
public:
CDBPool(){}
CDBPool(constchar* pool_name,constchar* db_server_ip,uint16_t db_server_port,
constchar* username,constchar* password,constchar* db_name,
int max_conn_cnt);
virtual~CDBPool();
intInit();// 连接数据库，创建连接
CDBConn*GetDBConn(const int timeout_ms = 0);// 获取连接资源
voidRelDBConn(CDBConn* pConn);// 归还连接资源
constchar*GetPoolName(){return m_pool_name.c_str();}
constchar*GetDBServerIP(){return m_db_server_ip.c_str();}
uint16_tGetDBServerPort(){return m_db_server_port;}
constchar*GetUsername(){return m_username.c_str();}
constchar*GetPasswrod(){return m_password.c_str();}
constchar*GetDBName(){return m_db_name.c_str();}
private:
string m_pool_name;// 连接池名称
string m_db_server_ip;// 数据库ip
uint16_t m_db_server_port;// 数据库端口
string m_username;// 用户名
string m_password;// 用户密码
string m_db_name;// db名称
int m_db_cur_conn_cnt;// 当前启用的连接数量
int m_db_max_conn_cnt;// 最大连接数量
list<CDBConn*> m_free_list;// 空闲的连接
list<CDBConn*> m_used_list;// 记录已经被请求的连接
std::mutex m_mutex;
std::condition_variable m_cond_var;
bool m_abort_request =false;
// CThreadNotify m_free_notify; // 信号量
};
#endif /* DBPOOL_H_ */
cpp文件
#include "DBPool.h"
#include <string.h>
#define log_error printf
#define log_warn printf
#define log_info printf
#define MIN_DB_CONN_CNT 1
#define MAX_DB_CONN_FAIL_NUM 10
CResultSet::CResultSet(MYSQL_RES *res)
{
m_res = res;
// map table field key to index in the result array
int num_fields = mysql_num_fields(m_res);
MYSQL_FIELD *fields = mysql_fetch_fields(m_res);
for(int i =0; i < num_fields; i++)
{
// 多行
m_key_map.insert(make_pair(fields[i].name, i));
}
}
CResultSet::~CResultSet()
{
if(m_res)
{
mysql_free_result(m_res);
m_res =NULL;
}
}
boolCResultSet::Next()
{
m_row = mysql_fetch_row(m_res);
if(m_row)
{
returntrue;
}
else
{
returnfalse;
}
}
intCResultSet::_GetIndex(constchar*key)
{
map<string,int>::iterator it = m_key_map.find(key);
if(it == m_key_map.end())
{
return-1;
}
else
{
return it->second;
}
}
intCResultSet::GetInt(const char *key)
{
int idx =_GetIndex(key);
if(idx ==-1)
{
return0;
}
else
{
return atoi(m_row[idx]);// 有索引
}
}
char*CResultSet::GetString(const char *key)
{
int idx =_GetIndex(key);
if(idx ==-1)
{
returnNULL;
}
else
{
return m_row[idx];// 列
}
}
/////////////////////////////////////////
CPrepareStatement::CPrepareStatement()
{
m_stmt =NULL;
m_param_bind =NULL;
m_param_cnt =0;
}
CPrepareStatement::~CPrepareStatement()
{
if(m_stmt)
{
mysql_stmt_close(m_stmt);
m_stmt =NULL;
}
if(m_param_bind)
{
delete[] m_param_bind;
m_param_bind =NULL;
}
}
boolCPrepareStatement::Init(MYSQL *mysql, string &sql)
{
mysql_ping(mysql);// 当mysql连接丢失的时候，使用mysql_ping能够自动重连数据库
//g_master_conn_fail_num ++;
m_stmt = mysql_stmt_init(mysql);
if(!m_stmt)
{
log_error("mysql_stmt_init failed\n");
returnfalse;
}
if(mysql_stmt_prepare(m_stmt, sql.c_str(), sql.size()))
{
log_error("mysql_stmt_prepare failed: %s\n", mysql_stmt_error(m_stmt));
returnfalse;
}
m_param_cnt = mysql_stmt_param_count(m_stmt);
if(m_param_cnt >0)
{
m_param_bind =new MYSQL_BIND[m_param_cnt];
if(!m_param_bind)
{
log_error("new failed\n");
returnfalse;
}
memset(m_param_bind,0,sizeof(MYSQL_BIND)* m_param_cnt);
}
returntrue;
}
voidCPrepareStatement::SetParam(uint32_t index, int &value)
{
if(index >= m_param_cnt)
{
log_error("index too large: %d\n", index);
return;
}
m_param_bind[index].buffer_type = MYSQL_TYPE_LONG;
m_param_bind[index].buffer =&value;
}
voidCPrepareStatement::SetParam(uint32_t index, uint32_t &value)
{
if(index >= m_param_cnt)
{
log_error("index too large: %d\n", index);
return;
}
m_param_bind[index].buffer_type = MYSQL_TYPE_LONG;
m_param_bind[index].buffer =&value;
}
voidCPrepareStatement::SetParam(uint32_t index, string &value)
{
if(index >= m_param_cnt)
{
log_error("index too large: %d\n", index);
return;
}
m_param_bind[index].buffer_type = MYSQL_TYPE_STRING;
m_param_bind[index].buffer =(char*)value.c_str();
m_param_bind[index].buffer_length = value.size();
}
voidCPrepareStatement::SetParam(uint32_t index, const string &value)
{
if(index >= m_param_cnt)
{
log_error("index too large: %d\n", index);
return;
}
m_param_bind[index].buffer_type = MYSQL_TYPE_STRING;
m_param_bind[index].buffer =(char*)value.c_str();
m_param_bind[index].buffer_length = value.size();
}
boolCPrepareStatement::ExecuteUpdate()
{
if(!m_stmt)
{
log_error("no m_stmt\n");
returnfalse;
}
if(mysql_stmt_bind_param(m_stmt, m_param_bind))
{
log_error("mysql_stmt_bind_param failed: %s\n", mysql_stmt_error(m_stmt));
returnfalse;
}
if(mysql_stmt_execute(m_stmt))
{
log_error("mysql_stmt_execute failed: %s\n", mysql_stmt_error(m_stmt));
returnfalse;
}
if(mysql_stmt_affected_rows(m_stmt)==0)
{
log_error("ExecuteUpdate have no effect\n");
returnfalse;
}
returntrue;
}
uint32_tCPrepareStatement::GetInsertId()
{
return mysql_stmt_insert_id(m_stmt);
}
/////////////////////
CDBConn::CDBConn(CDBPool*pPool)
{
m_pDBPool = pPool;
m_mysql =NULL;
}
CDBConn::~CDBConn()
{
if(m_mysql)
{
mysql_close(m_mysql);
}
}
intCDBConn::Init()
{
m_mysql = mysql_init(NULL);// mysql_标准的mysql c client对应的api
if(!m_mysql)
{
log_error("mysql_init failed\n");
return1;
}
my_bool reconnect =true;
mysql_options(m_mysql, MYSQL_OPT_RECONNECT,&reconnect);// 配合mysql_ping实现自动重连
mysql_options(m_mysql, MYSQL_SET_CHARSET_NAME,"utf8mb4");// utf8mb4和utf8区别
// ip 端口 用户名 密码 数据库名
if(!mysql_real_connect(m_mysql, m_pDBPool->GetDBServerIP(), m_pDBPool->GetUsername(), m_pDBPool->GetPasswrod(),
m_pDBPool->GetDBName(), m_pDBPool->GetDBServerPort(),NULL,0))
{
log_error("mysql_real_connect failed: %s\n", mysql_error(m_mysql));
return2;
}
return0;
}
constchar*CDBConn::GetPoolName()
{
return m_pDBPool->GetPoolName();
}
boolCDBConn::ExecuteCreate(const char *sql_query)
{
mysql_ping(m_mysql);
// mysql_real_query 实际就是执行了SQL
if(mysql_real_query(m_mysql, sql_query,strlen(sql_query)))
{
log_error("mysql_real_query failed: %s, sql: start transaction\n", mysql_error(m_mysql));
returnfalse;
}
returntrue;
}
boolCDBConn::ExecuteDrop(const char *sql_query)
{
mysql_ping(m_mysql);// 如果端开了，能够自动重连
if(mysql_real_query(m_mysql, sql_query,strlen(sql_query)))
{
log_error("mysql_real_query failed: %s, sql: start transaction\n", mysql_error(m_mysql));
returnfalse;
}
returntrue;
}
CResultSet*CDBConn::ExecuteQuery(const char *sql_query)
{
mysql_ping(m_mysql);
if(mysql_real_query(m_mysql, sql_query,strlen(sql_query)))
{
log_error("mysql_real_query failed: %s, sql: %s\n", mysql_error(m_mysql), sql_query);
returnNULL;
}
// 返回结果
MYSQL_RES *res = mysql_store_result(m_mysql);// 返回结果
if(!res)
{
log_error("mysql_store_result failed: %s\n", mysql_error(m_mysql));
returnNULL;
}
CResultSet*result_set =newCResultSet(res);// 存储到CResultSet
return result_set;
}
/*
1.执行成功，则返回受影响的行的数目，如果最近一次查询失败的话，函数返回 -1
2.对于delete,将返回实际删除的行数.
3.对于update,如果更新的列值原值和新值一样,如update tables set col1=10 where id=1;
id=1该条记录原值就是10的话,则返回0。
mysql_affected_rows返回的是实际更新的行数,而不是匹配到的行数。
*/
boolCDBConn::ExecuteUpdate(const char *sql_query, bool care_affected_rows)
{
mysql_ping(m_mysql);
if(mysql_real_query(m_mysql, sql_query,strlen(sql_query)))
{
log_error("mysql_real_query failed: %s, sql: %s\n", mysql_error(m_mysql), sql_query);
//g_master_conn_fail_num ++;
returnfalse;
}
if(mysql_affected_rows(m_mysql)>0)
{
returntrue;
}
else
{// 影响的行数为0时
if(care_affected_rows)
{// 如果在意影响的行数时, 返回false, 否则返回true
log_error("mysql_real_query failed: %s, sql: %s\n\n", mysql_error(m_mysql), sql_query);
returnfalse;
}
else
{
log_warn("affected_rows=0, sql: %s\n\n", sql_query);
returntrue;
}
}
}
// bool CDBConn::ExecuteUpdate2(const char *sql_query, bool care_affected_rows)
// {
// again: // 不能这么写
// if (mysql_real_query(m_mysql, sql_query, strlen(sql_query)))
// {
// log_error("mysql_real_query failed: %s, sql: %s\n", mysql_error(m_mysql), sql_query);
// //g_master_conn_fail_num ++;
// return false;
// } else {
// mysql_ping(m_mysql);
// goto again;
// }
// if (mysql_affected_rows(m_mysql) > 0)
// {
// return true;
// }
// else
// { // 影响的行数为0时
// if (care_affected_rows)
// { // 如果在意影响的行数时, 返回false, 否则返回true
// log_error("mysql_real_query failed: %s, sql: %s\n\n", mysql_error(m_mysql), sql_query);
// return false;
// }
// else
// {
// log_warn("affected_rows=0, sql: %s\n\n", sql_query);
// return true;
// }
// }
// }
boolCDBConn::StartTransaction()
{
mysql_ping(m_mysql);
if(mysql_real_query(m_mysql,"start transaction\n",17))
{
log_error("mysql_real_query failed: %s, sql: start transaction\n", mysql_error(m_mysql));
returnfalse;
}
returntrue;
}
boolCDBConn::Rollback()
{
mysql_ping(m_mysql);
if(mysql_real_query(m_mysql,"rollback\n",8))
{
log_error("mysql_real_query failed: %s, sql: rollback\n", mysql_error(m_mysql));
returnfalse;
}
returntrue;
}
boolCDBConn::Commit()
{
mysql_ping(m_mysql);
if(mysql_real_query(m_mysql,"commit\n",6))
{
log_error("mysql_real_query failed: %s, sql: commit\n", mysql_error(m_mysql));
returnfalse;
}
returntrue;
}
uint32_tCDBConn::GetInsertId()
{
return(uint32_t)mysql_insert_id(m_mysql);
}
////////////////
CDBPool::CDBPool(constchar*pool_name,constchar*db_server_ip,uint16_t db_server_port,
constchar*username,constchar*password,constchar*db_name,int max_conn_cnt)
{
m_pool_name = pool_name;
m_db_server_ip = db_server_ip;
m_db_server_port = db_server_port;
m_username = username;
m_password = password;
m_db_name = db_name;
m_db_max_conn_cnt = max_conn_cnt;//
m_db_cur_conn_cnt = MIN_DB_CONN_CNT;// 最小连接数量
}
// 释放连接池
CDBPool::~CDBPool()
{
// lock_guard 构造函数加锁，析构函数解锁
std::lock_guard<std::mutex>lock(m_mutex);
m_abort_request =true;
m_cond_var.notify_all();// 通知所有在等待的
for(list<CDBConn*>::iterator it = m_free_list.begin(); it != m_free_list.end(); it++)
{
CDBConn*pConn =*it;
delete pConn;
}
m_free_list.clear();
}
intCDBPool::Init()
{
// 创建固定最小的连接数量
for(int i =0; i < m_db_cur_conn_cnt; i++)
{
CDBConn*pDBConn =newCDBConn(this);
int ret = pDBConn->Init();
if(ret)
{
delete pDBConn;
return ret;
}
m_free_list.push_back(pDBConn);
}
// log_info("db pool: %s, size: %d\n", m_pool_name.c_str(), (int)m_free_list.size());
return0;
}
/*
*TODO: 增加保护机制，把分配的连接加入另一个队列，这样获取连接时，如果没有空闲连接，
*TODO: 检查已经分配的连接多久没有返回，如果超过一定时间，则自动收回连接，放在用户忘了调用释放连接的接口
* timeout_ms默认为 0死等
* timeout_ms >0 则为等待的时间
*/
int wait_cout =0;
CDBConn*CDBPool::GetDBConn(const int timeout_ms)
{
std::unique_lock<std::mutex>lock(m_mutex);
if(m_abort_request)
{
log_warn("have aboort\n");
returnNULL;
}
if(m_free_list.empty())// 当没有连接可以用时
{
// 第一步先检测 当前连接数量是否达到最大的连接数量
if(m_db_cur_conn_cnt >= m_db_max_conn_cnt)
{
// 如果已经到达了，看看是否需要超时等待
if(timeout_ms <=0)// 死等，直到有连接可以用 或者 连接池要退出
{
log_info("wait ms:%d\n", timeout_ms);
m_cond_var.wait(lock,[this]
{
// log_info("wait:%d, size:%d\n", wait_cout++, m_free_list.size());
// 当前连接数量小于最大连接数量 或者请求释放连接池时退出
return(!m_free_list.empty())| m_abort_request;
});
}else{
// return如果返回 false，继续wait(或者超时), 如果返回true退出wait
// 1.m_free_list不为空
// 2.超时退出
// 3. m_abort_request被置为true，要释放整个连接池
m_cond_var.wait_for(lock,std::chrono::milliseconds(timeout_ms),[this]{
// log_info("wait_for:%d, size:%d\n", wait_cout++, m_free_list.size());
return(!m_free_list.empty())| m_abort_request;
});
// 带超时功能时还要判断是否为空
if(m_free_list.empty())// 如果连接池还是没有空闲则退出
{
returnNULL;
}
}
if(m_abort_request)
{
log_warn("have aboort\n");
returnNULL;
}
}
else// 还没有到最大连接则创建连接
{
CDBConn*pDBConn =newCDBConn(this);//新建连接
int ret = pDBConn->Init();
if(ret)
{
log_error("Init DBConnecton failed\n\n");
delete pDBConn;
returnNULL;
}
else
{
m_free_list.push_back(pDBConn);
m_db_cur_conn_cnt++;
// log_info("new db connection: %s, conn_cnt: %d\n", m_pool_name.c_str(), m_db_cur_conn_cnt);
}
}
}
CDBConn*pConn = m_free_list.front();// 获取连接
m_free_list.pop_front();// STL 吐出连接，从空闲队列删除
// pConn->setCurrentTime(); // 伪代码
m_used_list.push_back(pConn);//
return pConn;
}
voidCDBPool::RelDBConn(CDBConn *pConn)
{
std::lock_guard<std::mutex>lock(m_mutex);
list<CDBConn*>::iterator it = m_free_list.begin();
for(; it != m_free_list.end(); it++)// 避免重复归还
{
if(*it == pConn)
{
break;
}
}
if(it == m_free_list.end())
{
m_used_list.remove(pConn);
m_free_list.push_back(pConn);
m_cond_var.notify_one();// 通知取队列
}else
{
log_error("RelDBConn failed\n");
}
}
// 遍历检测是否超时未归还
// pConn->isTimeout(); // 当前时间 - 被请求的时间
// 强制回收 从m_used_list 放回 m_free_list
4.3、另外版本mysql连接池
对比上面代码，耦合性更强一点，但是比较好理解一点： 1、init创建多个连接对象，并放入到队列中，初始化信号量为连接数的个数 2、GetConnection从连接池中返回一个连接，先信号量--，然后拿锁，推出链接 3、ReleaseConnection返回一个连接，先拿锁，然后归还连接，然后信号量++ 4、GetInstance当前类是一个单例类，用来获取对象
1、锁、条件变量、信号量封装
#ifndef LOCKER_H
#define LOCKER_H
#include <exception>
#include <pthread.h>
#include <semaphore.h>
class sem
{
public:
sem()
{
if(sem_init(&m_sem,0,0)!=0)
{
throwstd::exception();
}
}
sem(int num)
{
if(sem_init(&m_sem,0, num)!=0)
{
throwstd::exception();
}
}
~sem()
{
sem_destroy(&m_sem);
}
boolwait()
{
return sem_wait(&m_sem)==0;
}
boolpost()
{
return sem_post(&m_sem)==0;
}
private:
sem_t m_sem;
};
class locker
{
public:
locker()
{
if(pthread_mutex_init(&m_mutex,NULL)!=0)
{
throwstd::exception();
}
}
~locker()
{
pthread_mutex_destroy(&m_mutex);
}
boollock()
{
return pthread_mutex_lock(&m_mutex)==0;
}
boolunlock()
{
return pthread_mutex_unlock(&m_mutex)==0;
}
pthread_mutex_t*get()
{
return&m_mutex;
}
private:
pthread_mutex_t m_mutex;
};
class cond
{
public:
cond()
{
if(pthread_cond_init(&m_cond,NULL)!=0)
{
//pthread_mutex_destroy(&m_mutex);
throwstd::exception();
}
}
~cond()
{
pthread_cond_destroy(&m_cond);
}
boolwait(pthread_mutex_t *m_mutex)
{
int ret =0;
//pthread_mutex_lock(&m_mutex);
ret = pthread_cond_wait(&m_cond, m_mutex);
//pthread_mutex_unlock(&m_mutex);
return ret ==0;
}
booltimewait(pthread_mutex_t *m_mutex, struct timespec t)
{
int ret =0;
//pthread_mutex_lock(&m_mutex);
ret = pthread_cond_timedwait(&m_cond, m_mutex,&t);
//pthread_mutex_unlock(&m_mutex);
return ret ==0;
}
boolsignal()
{
return pthread_cond_signal(&m_cond)==0;
}
boolbroadcast()
{
return pthread_cond_broadcast(&m_cond)==0;
}
private:
//static pthread_mutex_t m_mutex;
pthread_cond_t m_cond;
};
#endif
2、头文件sql_connection_pool.h
#ifndef _CONNECTION_POOL_
#define _CONNECTION_POOL_
#include <stdio.h>
#include <list>
#include <mysql/mysql.h>
#include <error.h>
#include <string.h>
#include <iostream>
#include <string>
#include "locker.h"
#include "log.h"
usingnamespacestd;
class connection_pool
{
public:
MYSQL *GetConnection();//获取数据库连接
boolReleaseConnection(MYSQL *conn);//释放连接
intGetFreeConn();//获取连接
voidDestroyPool();//销毁所有连接
//单例模式
static connection_pool *GetInstance();
voidinit(string url, string User, string PassWord, string DataBaseName, int Port, int MaxConn, int close_log);
private:
connection_pool();
~connection_pool();
int m_MaxConn;//最大连接数
int m_CurConn;//当前已使用的连接数
int m_FreeConn;//当前空闲的连接数
locker lock;
list<MYSQL *> connList;//连接池
sem reserve;
public:
string m_url;//主机地址
string m_Port;//数据库端口号
string m_User;//登陆数据库用户名
string m_PassWord;//登陆数据库密码
string m_DatabaseName;//使用数据库名
int m_close_log;//日志开关
};
class connectionRAII{
public:
connectionRAII(MYSQL **con, connection_pool *connPool);
~connectionRAII();
private:
MYSQL *conRAII;
connection_pool *poolRAII;
};
#endif
3、cpp文件
#include <mysql/mysql.h>
#include <stdio.h>
#include <string>
#include <string.h>
#include <stdlib.h>
#include <list>
#include <pthread.h>
#include <iostream>
#include "sql_connection_pool.h"
usingnamespacestd;
connection_pool::connection_pool()
{
m_CurConn =0;
m_FreeConn =0;
}
connection_pool *connection_pool::GetInstance()
{
static connection_pool connPool;
return&connPool;
}
//构造初始化
voidconnection_pool::init(string url, string User, string PassWord, string DBName, int Port, int MaxConn, int close_log)
{
m_url = url;
m_Port =Port;
m_User =User;
m_PassWord =PassWord;
m_DatabaseName =DBName;
m_close_log = close_log;
for(int i =0; i <MaxConn; i++)
{
MYSQL *con =NULL;
con = mysql_init(con);
if(con ==NULL)
{
LOG_ERROR("MySQL Error");
exit(1);
}
con = mysql_real_connect(con, url.c_str(),User.c_str(),PassWord.c_str(),DBName.c_str(),Port,NULL,0);
if(con ==NULL)
{
LOG_ERROR("MySQL Error");
exit(1);
}
connList.push_back(con);
++m_FreeConn;
}
reserve = sem(m_FreeConn);
m_MaxConn = m_FreeConn;
}
//当有请求时，从数据库连接池中返回一个可用连接，更新使用和空闲连接数
MYSQL *connection_pool::GetConnection()
{
MYSQL *con =NULL;
if(0== connList.size())
returnNULL;
reserve.wait();
lock.lock();
con = connList.front();
connList.pop_front();
--m_FreeConn;
++m_CurConn;
lock.unlock();
return con;
}
//释放当前使用的连接
boolconnection_pool::ReleaseConnection(MYSQL *con)
{
if(NULL== con)
returnfalse;
lock.lock();
connList.push_back(con);
++m_FreeConn;
--m_CurConn;
lock.unlock();
reserve.post();
returntrue;
}
//销毁数据库连接池
voidconnection_pool::DestroyPool()
{
lock.lock();
if(connList.size()>0)
{
list<MYSQL *>::iterator it;
for(it = connList.begin(); it != connList.end();++it)
{
MYSQL *con =*it;
mysql_close(con);
}
m_CurConn =0;
m_FreeConn =0;
connList.clear();
}
lock.unlock();
}
//当前空闲的连接数
intconnection_pool::GetFreeConn()
{
returnthis->m_FreeConn;
}
connection_pool::~connection_pool()
{
DestroyPool();
}
connectionRAII::connectionRAII(MYSQL **SQL, connection_pool *connPool)
{
*SQL = connPool->GetConnection();
conRAII =*SQL;
poolRAII = connPool;
}
connectionRAII::~connectionRAII(){
poolRAII->ReleaseConnection(conRAII);
}
4.4、mysql连接重连机制
1、设置启用（当发现连接断开时的）自动重连 my_bool reconnect = true; mysql_options(m_mysql, MYSQL_OPT_RECONNECT, &reconnect); // 配合mysql_ping实现自动重连
2、检测连接是否正常 int STDCALL mysql_ping(MYSQL *mysql); 描述： 检查与服务端的连接是否正常。连接断开时，如果自动重新连接功能未被禁用，则尝试重新连接服务 器。该函数可被客户端用来检测闲置许久以后，与服务端的连接是否关闭，如有需要，则重新连接。 返回值： 连接正常，返回0；如有错误发生，则返回非0值。返回非0值并不意味着服务器本身关闭掉，也有可能 是网络原因导致网络不通。
4.5、连接池连接设置数量
连接数 = ((核心数 * 2) + 有效磁盘数) 按照这个公式，即是说你的服务器 CPU 是 4核 i7 的，那连接池连接数大小应该为 ((4*2)+1)=9 这里只是一个经验公式。还要和线程池数量以及具体业务结合在一起。 CPU总核数 = 物理CPU个数 * 每颗物理CPU的核数 总逻辑CPU数 = 物理CPU个数 * 每颗物理CPU的核数 * 超线程数
// 查看CPU信息（型号）
cat /proc/cpuinfo | grep name | cut -f2 -d: | uniq -c
// 查看物理CPU个数
cat /proc/cpuinfo| grep "physical id"| sort| uniq| wc -l
// 查看每个物理CPU中core的个数(即核数)
cat /proc/cpuinfo| grep "cpu cores"| uniq
// 查看逻辑CPU的个数
cat /proc/cpuinfo| grep "processor"| wc -l
原文作者：讳疾忌医-note；如有侵权，请联系管理员删除。
上一篇：
Ubuntu 安装Go编译器
下一篇：
C++11 基础知识
请登录后发表评论
登录
全部评论
最新文章
- CUDA与C++多线程性能之争：为何CUDA在金融计算中更胜一筹？
- C++高频交易系统延迟痛点大揭秘：从毫秒到纳秒的调优之路
- 为什么你的C++程序又大又慢？大部份开发者忽视的LTO链接优化陷阱
- 中国人眼中的沃尔特?布莱特：D语言比C++ 更优？多数开发者的认知误区
- 榨干每一滴性能：编译器不想让你知道的C++性能秘诀
- 国外C++架构师分享：从架构小白到架构师，看看他是如何做到的？
- 国外C++程序员分享：当OCR遇上C++和OpenCV，图像文本提取变多简单？
- 探索C++语言的起源密码：一段鲜为人知的技术传奇
- 震惊！从物理跨界C++：如何凭 STL 重塑C++编程格局
- 打工人必备！GitHub惊现神器，上班提高效果再也不怕被抓
