mysql — 基于协程的mysql异步代理服务

迷南。 2022-05-20 20:34 204阅读 0赞

# 基于协程的mysql异步代理服务 #

最近在TX实习，由于组里需要一个mysql异步代理来提升并发量和效率，导师给我一个这样的任务. 使用协程的mysql异步代理服务

那么为什么需要一个mysql异步代理？ 如果没有mysql异步代理，单个进程的最大只有3000多QPS. 但是利用司内原生态spp框架

以及协程，再加上连接接池这些优化单机单进程可以达到25000 QPS

当前司内开发的海外项目多采用mysql作为数据存储，开发过程碰到的主要困难有三个：1)由于Mysql原生只提供同步API,为了保

证并发能力，基于异步框架例如spp的服务需要自行改造mysql client 2)后端服务需要自己管理mysql连接，在进行复杂数据处理

，特别是事务时容易引入bug 3).DB配置与服务器强相关，扩缩容时还需要变更DB访问权限授权.

**首先你需要了解协程的概念：[\->>戳进去][-]**

为什么仅仅只有这么几个优化点，最大并发量就能翻那么多倍. 主要从以下几点来看：

>1.如果你使用原生mysql库提供的函数，这些函数对于一个进程的话是同步的，也就是说你在等待网络IO的时候是一个阻塞的状态，

那么你可以考虑一下如果你同时又1000个用户来请求你，那么请求的耗时会有多少？ 当然有人又说了那我开一个多线程版本！ 那

么这对cpu的压力非常大，还有资源的利用率如何？ 同一时刻会有大量的线程在等待网络IO. 最后我们如果可以在进程等待网络IO

的时候，让进程执行别的请求该有多好？ 协程！ 买不了吃亏买不了上当！ 一个协程对应N个请求，当正在执行的socket发生网

络IO的时候，协程立马切换去执行另一个socket请求，当前面的socket就绪之后再切回来，只有资源利用率100%，处理并发的效率

也非常的高. 单进程模式下可以跑到25000QPS.

>2.当你使用命令请求mysql服务器的时候，那么你首先会经历三次握手，然后mysql服务器的握手认证，接下来发送命令请求，接

收到命令的结果，最后发送quit数据报，然后四次挥手，close掉socket. 其实你发现我们每次只需要发送命令请求和接收到命令

结果，其他的我们都没有必要去重复的做，而且这些操作耗时非常大，也浪费资源. 我们我们可以维护一个连接池让这个连接池

中存储的数据一直都是已经认证成功的socket，也就是每次只需要发送命令收结果就可以了，当你不使用时再把她们放回

连接池. **这样10000 QPS 你就能节省将近****10000次三次握手 四次挥手 mysql服务器认证3次握手 mysql quit认证.** 所以响应

时间，最大并发数，以及CPU的有效使用率都会有相应的提升！ 这是第二个优化点.

>3.上面已经提到了，如果我们想实现异步版本的mysql代理，我们需要丢弃掉原生提供的函数，直接实现与mysql服务器的交互，

这是一个相当复杂的过程.首先你得熟悉各种各样的mysql网络协议包格式，由于组里的需求我这里只实现select/delect/update

/insert四种命令.但是需要熟悉握手包格式，命令包格式，ok/error/eof包的格式，回应结果包格式，直接对接收到的连续数据

做解包，然后需要判断超时，判断包的完整性，整个过程相当复杂，但是写成之后我们就可以在send命令到mysql服务器的时候使

用协程来执行，实现异步执行.mysql网络协议包格式我有一篇博客：[\->>>戳进来][- 1]

举个例子：我现在要处理mysql服务的握手包我该如何操作呢？ 这里附上代码 解包 打包过程：

struct st_handshake_server_packet_41
    {
    	char version_no; //版本号
    	char version_info[32];  //版本信息
    	unsigned int connection_id; //连接id
    	char auth_plugin_data_part_1[9]; //8位挑战数
    	char fill;   //填充物
    	char character_set; //字符集
    	short server_status; //服务器状态
    	char auth_plugin_data_len; //下半部分挑战数的长度.
    	char reserved[11];   //预留字段
    	char auth_plugin_data_part_2[257]; //剩余的挑战数
    	u_capability capability;  //能力标识
    	unsigned int sequence_id; //连接id
    
    
    	void init()
    	{
    		capability.value=0;
    		version_no='\0';
    		memset(version_info,0,32);
    		connection_id=0;
    		memset(auth_plugin_data_part_1,0,9);
    		fill='\0';
    		character_set='\0';
    		server_status=0;
    		auth_plugin_data_len='\0';
    		memset(reserved,0,11);
    		memset(auth_plugin_data_part_2,0,257);
    		sequence_id=0;
    	}
    
    	void display()
    	{
    
    		SPP_NORMAL_LOG("version_no %d,version_info:%s,connection_id:%u,auth_plugin_data_part_1:%s,fill:%d,character_set:%d,server_status:%d,auth_plugin_data_len:%d,auth_plugin_data_part_2:%s,capability.value:%x,sequence_id:%u"
    				,int(version_no)
    				,version_info
    				,connection_id
    				,auth_plugin_data_part_1
    				,int(fill)
    				,int(character_set)
    				,int(server_status)
    				,int(auth_plugin_data_len)
    				,auth_plugin_data_part_2
    				,capability.value
    				,sequence_id
    				);
    	}
    };
    
    
    struct st_handshake_client_response_packet_41
    {
    	unsigned int capability;   //能力标识
    	unsigned int max_packet_size; //最大消息长度
    	char character_set; //字符编码
    	char reserved [22]; //填充值
    	std::string username; //user name
    	unsigned long auth_response_len; //服务器挑战数响应数据长度.
    	std::string auth_response;  //挑战认证数据
    	std::string database;   //库名字
    	std::string auth_plugin_name; //身份认证插件
    	std::map<std::string,std::string> key_value_list;
    
    
    	void display()
    	{
    
    		SPP_NORMAL_LOG("capability %u,max_packet_size:%u,character_set:%u,username:%s,auth_response_len:%ld,auth_response:%s,database:%s,auth_plugin_name:%s"
    				,capability
    				,max_packet_size
    				,int(character_set)
    				,username.c_str()
    				,auth_response_len
    				,auth_response.c_str()
    				,database.c_str()
    				,auth_plugin_name.c_str()
    				);
    	}
    };
    //客户端打包回包给服务端.
    int handshake_client_response_packet_41(void* buf, int& len,const st_handshake_client_response_packet_41& data,unsigned int sequence_id)
    {
    	char* init_buf=(char*)buf;
    	buf+=4;
    	(*(unsigned int*)buf)= data.capability;  //客户端权能标志
    	buf+=4;
    	(*(unsigned int*)buf)= data.max_packet_size; //最大消息长度
    	buf+=4;
    	(*(unsigned char*)buf)= data.character_set; //字符编码
    	buf+=1;
    	memset(buf,0,23);    //填充值
    	buf+=23;
    	memcpy(buf,data.username.c_str(),data.username.size()); //user name
    	buf+=data.username.size();
    	(*(unsigned char*)buf)='\0';
    	buf+=1;
    	
    	//服务器：了解中的身份验证响应数据的长度编码整数 Protocol::HandshakeResponse41
    	//客户端：身份验证响应数据 Protocol::HandshakeResponse41 的长度是长度编码的整数
    	if(data.capability & CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA )
    	{
    		SPP_NORMAL_LOG("CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA");
    		unsigned long len = data.auth_response.size();
    		pack_lenenc_int(buf,len);
    		memcpy(buf,data.auth_response.c_str(),len);  //挑战认证数据
    		buf += len;
    	}
    	//服务器：支持 Authentication::Native41
    	//客户端：支持 Authentication::Native41
    	else if (data.capability & CLIENT_SECURE_CONNECTION )
    	{
    		SPP_NORMAL_LOG("CLIENT_SECURE_CONNECTION");
    		unsigned long len = data.auth_response.size();
    		pack_lenenc_int(buf,len);
    		memcpy(buf,data.auth_response.c_str(),len);  //挑战认证数据
    		buf += len;
    	}
    	else
    	{
    		SPP_NORMAL_LOG("default");
    		unsigned long len = data.auth_response.size();
    		memcpy(buf,data.auth_response.c_str(),len); //挑战认证数据
    		buf += len;
    		(*(unsigned char*)buf)='\0';
    		buf+=1;
    
    	}
    	
    	
    	//服务器： 支持模式名称 Handshake Response Packet
    	//客户端： Handshake Response Packet 包含一个模式名称
    	if(data.capability & CLIENT_CONNECT_WITH_DB )
    	{
    		SPP_NORMAL_LOG("CLIENT_CONNECT_WITH_DB");
    		unsigned long len = data.database.size();
    		memcpy(buf,data.database.c_str(),len);  //数据库名称
    		buf += len;
    		(*(unsigned char*)buf)='\0';
    		buf+=1;
    	}
    	
    	//服务器： 在初始握手数据包中发送额外的数据 并支持可插入的身份验证协议
    	//客户端： 支持身份验证插件
    	if(data.capability & CLIENT_PLUGIN_AUTH  ) 
    	{
    		SPP_NORMAL_LOG("CLIENT_PLUGIN_AUTH");
    		unsigned long len = data.auth_plugin_name.size();
    		memcpy(buf,data.auth_plugin_name.c_str(),len);
    		buf += len;
    		(*(unsigned char*)buf)='\0';
    		buf+=1;
    	}
    	
    	//服务器：允许连接属性 || 客户端：发送连接属性
    	if(data.capability & CLIENT_CONNECT_ATTRS)  
    	{
    		SPP_NORMAL_LOG("CLIENT_CONNECT_ATTRS");
    
    		char temp[10240];
    		void* buf_temp = temp;
    		long total_len=0;
    		std::map<std::string,std::string>::const_iterator iter = data.key_value_list.begin();
    		for(;iter !=data.key_value_list.end(); iter ++ )
    		{
    			unsigned long len = iter->first.size();
    			pack_lenenc_int(buf_temp,len);
    			memcpy(buf_temp,iter->first.c_str(),len);
    			buf_temp += len;
    			total_len+=len;
    			len = iter->second.size();
    			pack_lenenc_int(buf_temp,len);
    			memcpy(buf_temp,iter->second.c_str(),len);
    			buf_temp += len;
    			total_len+=len;
    		}
    		void* buf_all_len = buf;
    		pack_lenenc_int(buf,total_len);
    		memcpy(buf,buf_temp,total_len);
    		buf += total_len;
    	}
    
    	len=(long)buf-(long)init_buf;
    	(*(unsigned int*)init_buf)=len-4;
    	init_buf[3]=sequence_id;
    	SPP_NORMAL_LOG("len:%d",len);
    
    }
    
    //客户端打解服务器的握手初始化包
    //https://dev.mysql.com/doc/internals/en/connection-phase-packets.html#packet-Protocol::Handshake
    int unpack_handshake_server_packet_41(void* buf, int len,st_handshake_server_packet_41& data,unsigned int & sequence_id)
    {
    	SPP_NORMAL_LOG("mysql len:%d",len);
    	if(len <= 4)
    	{
    		SPP_NORMAL_LOG("len error:%d",len);
    		return -1;
    	}
    	
    	//解析包的头信息 一共四个字节.
    	data3 h;
    	h.value=0;
    	memcpy(h.data,buf,3);
    	if(h.value > len )
    	{
    		SPP_NORMAL_LOG("complete len:%d",h.value);
    		return -3;
    	}
    	sequence_id=(*(unsigned char*)(buf+3));
    
    	
    	int index = 4;
    	data.version_no=*(char*)(buf+index);  //版本号
    	index+=1;
    	SPP_NORMAL_LOG("data.version_no:%d",int(data.version_no));
    	int tmp_len = 0;
    	for(; tmp_len < h.value; tmp_len ++)
    	{
    		if('\0' == *(char*)(buf+index+tmp_len))
    		{
    			tmp_len++;
    			break;
    		}
    	}
    	memcpy(data.version_info,buf+index,tmp_len);  //版本信息
    	index+=(tmp_len);
    	SPP_NORMAL_LOG("index:%d,len:%d",index,h.value);
    	SPP_NORMAL_LOG("data.version_info:%s",data.version_info);
    	data.connection_id=ntohl(*(unsigned int*)(buf+index));  //连接id
    	index+=4;
    	SPP_NORMAL_LOG("data.connection_id:%u",unsigned(data.connection_id));
    
    	memcpy(data.auth_plugin_data_part_1,buf+index,8); //8位挑战数
    	data.auth_plugin_data_part_1[8]='\0';
    	index+=8;
    	SPP_NORMAL_LOG("data.auth_plugin_data_part_1:%s",data.auth_plugin_data_part_1);
    
    	data.fill=*(char*)(buf+index);  //填充值
    	index+=1;
    	SPP_NORMAL_LOG("data.fill:%d",int(data.fill));
    
    	memcpy(data.capability.data,buf+index,2);  //服务器权能标志(上半部分)
    	index+=2;
    	SPP_NORMAL_LOG("data.capability:%x",data.capability.value);
    
    	data.character_set=*(char*)(buf+index); //字符编码
    	index+=1;
    	SPP_NORMAL_LOG("data.character_set:%d",int(data.character_set));
    
    	data.server_status=ntohl(*(unsigned short*)(buf+index)); //服务器状态
    	index+=2;
    	SPP_NORMAL_LOG("data.server_status:%d",data.server_status);
    
    	memcpy(data.capability.data+2,buf+index,2); //服务器权能标志(下半部分)
    	index+=2;
    	SPP_NORMAL_LOG("data.capability:%x",data.capability.value);
    	
    	
    	if(data.capability.value&CLIENT_PLUGIN_AUTH )
    	{
    		//挑战数长度.
    		data.auth_plugin_data_len=*(char*)(buf+index);
    		index+=1;
    		SPP_NORMAL_LOG("data.auth_plugin_data_len:%d",int(data.auth_plugin_data_len));
    	}
    	else
    	{
    		//
    		data.auth_plugin_data_len='\0';
    		index+=1;
    		SPP_NORMAL_LOG("data.auth_plugin_data_len:%d",int(data.auth_plugin_data_len));
    	}
    
    	SPP_NORMAL_LOG("index:%d,len:%d",index,h.value);
    	//
    	memcpy(data.reserved,buf+index,10);  //保留字段
    	index+=10;
    	SPP_NORMAL_LOG("index:%d,len:%d",index,h.value);
    	if(data.capability.value&CLIENT_SECURE_CONNECTION )
    	{
    		//利用挑战数长度
    		int tmp_len = (data.auth_plugin_data_len-8)>13?(data.auth_plugin_data_len-8):13;
    		memcpy(data.auth_plugin_data_part_2,buf+index,tmp_len);
    		data.auth_plugin_data_part_2[tmp_len]='\0';
    		index+=tmp_len;
    		SPP_NORMAL_LOG("data.auth_plugin_data_part_2:%s",data.auth_plugin_data_part_2);
    	}
    	else if(data.capability.value&CLIENT_PLUGIN_AUTH )
    	{
    		//到空位置停止
    		int tmp_len = 0;
    		for(; tmp_len < h.value; tmp_len ++)
    		{
    			if('\0' == *(char*)(buf+index+tmp_len))
    			{
    				tmp_len++;
    				break;
    			}
    		}
    		memcpy(data.auth_plugin_data_part_2,buf+index,tmp_len);
    		index+=(tmp_len);
    		SPP_NORMAL_LOG("data.auth_plugin_data_part_2:%s",data.auth_plugin_data_part_2);
    
    	}
    	data.display();
    	return 0;
    }

当以上几点都实现了以后，你会发现你的mysql代理的性能会非常的nice！ 当然肯定还会有更好的优化方法. 对于保证包的完整性

和包的超时这里其实也是一个相当复杂的过程，希望大家多多思考！ 这是我的一点思路，并且将其已经实现，下面是压测报告：

![70][]

![70 1][]

[-]: https://blog.csdn.net/Dawn_sf/article/details/79509433
[- 1]: https://blog.csdn.net/Dawn_sf/article/details/80800183
[70]: /images/20220521/e0bde50306a14fb7a5bcfc3e836a1d07.png
[70 1]: /images/20220521/80e1e43b451d451b9c805b37759ccbb2.png