跨平台网络编程

Posted 2020-07-31 JohnKing_

　　程序员也三年了，工作以来，一直都在写游戏逻辑，刚好最近闲下来，难得有时间，自己研究一些新的东西。因此赶紧看看关于网络方面的东西，总结一下，好记心不如烂键盘。

　　其实总的来说，网络也就那么回事，比我原本想像中要简单很多，只有简单的几个接口。

　　一开始关于大数据，我还想复杂了，原本以为需要自己分包，再合包，而且我一开始也是这么做的，都已经做好了，再问别人才发现，这是没有必要的。（伤心！！） 

　　网络就是一个数据流，发送的时候，可以分多次send，接收时，也可以多次recv , 因为有缓存机制。

　　首先，需要做的只是给每条消息加上消息头，定义好包的大小及消息类型等其他业务相关的数据。

　　第二步， 发送消息时，先加上包头，后面跟数据。

　　第三步，接收消息时，先接收包头，再决定后面接收多大的数据。

 

以下直接贴代码：

客户端：

SocketClient.h

#ifndef __NET_SOCKET_CLIENT_H__
#define __NET_SOCKET_CLIENT_H__

#ifdef WIN32
#include <WinSock2.h>
#else
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/types.h>
#include <sys/ioctl.h>
#include <arpa/inet.h>
#include <unistd.h>

#ifndef typedef_socket_
#define typedef_socket_
typedef int SOCKET;
#endif
//typedef unsigned int SOCKET;
#define INVALID_SOCKET -1
#define SOCKET_ERROR -1
#endif

#include "ByteArray.h"
#include <thread>
#include <mutex>
#include <list>
#include <string>


#define  MsgPack_ptr std::shared_ptr<MsgPack>

class SocketClient {
private:
	SOCKET m_sock;
	std::shared_ptr<std::thread> m_thread;
	bool m_connected;
	std::mutex m_mutex;
	std::list<MsgPack*> m_requests;
	std::list<MsgPack_ptr> m_responseList;
	bool	m_running;
	std::string	m_callback;
	int32_t		m_recvTimeout;
public:
	SocketClient();
	~SocketClient();
	bool	_connect_timeout(std::string ip, unsigned short port, int timeout);
	bool	_connect_block(std::string ip, unsigned short port);
	bool	connect(std::string ip, unsigned short port, int timeout);
	int		send(const char* buf, int len);
	int		recv(char* buf, int len);
	int		close();
	void	startRecvThread();
	bool	isConnected() const { return m_connected; }
	void	setRecvTimeout(int32_t ms){ m_recvTimeout = ms; }

	bool sendRequest(MsgPack* reqest);
	MsgPack* recvMessage();

	std::list<MsgPack_ptr>&	getResponseList() { return m_responseList; }
	void lockMutex(); 
	void unlockMutex();

protected:
	void runRecv();
};


#endif

　　

SocketClient.cpp

#include "SocketClient.h"
#include <iostream>
#include "bytearray.h"
#include "msg.h"
//#include "../Utils/Utils.h"
#include "cocos2d.h"
USING_NS_CC;
#define NETLOG CCLOG

#ifdef WIN32
#pragma comment(lib, "wsock32")
class SocketIniter {
public:
	SocketIniter() {
		WSADATA wsaData;
		WORD version = MAKEWORD(2, 0);
		int rt = WSAStartup(version, &wsaData);
		if (rt) {
			std::cerr << "Initilize winsock error !" << std::endl;
		}
	}

	~SocketIniter() {
		WSACleanup();
	}
};

static SocketIniter s_socketIniter;
#endif


SocketClient::SocketClient() {
	m_connected = false;
	m_running = true;
	m_recvTimeout = 5;
}

SocketClient::~SocketClient() {
	close();
	m_running = false;
}

int SocketClient::close() {
	m_connected = false;
	::shutdown(m_sock, 2);
#ifdef WIN32
	::closesocket(m_sock);
#else
	::close(m_sock);
#endif

	m_mutex.lock();
//	for (auto i : m_responseList)
//		delete i;
	m_responseList.clear();
	m_mutex.unlock();

	return 0;
}

bool SocketClient::_connect_timeout(std::string ip, unsigned short port, int timeout)
{
	if (m_connected) {
		return true;
	}

	unsigned  long  non_blocking = 1;
	unsigned  long  blocking = 0;
	bool ret = false;

	int tryTimes = timeout / 2;
	for (int i = 0; i<tryTimes; ++i)
	{
		close();

		m_sock = socket(AF_INET, SOCK_STREAM, 0);

		int rcv_size = 1024 * 1024;
		setsockopt(m_sock, SOL_SOCKET, SO_RCVBUF, (char*)&rcv_size, sizeof(int));

		bool bKeppAlive = true;
		setsockopt(m_sock, SOL_SOCKET, SO_KEEPALIVE, (char*)&bKeppAlive, sizeof(bKeppAlive));

#ifdef WIN32
		if (-1 == ioctlsocket(m_sock, FIONBIO, (unsigned long*)&non_blocking))
			ret = false;
#else
		ioctl(m_sock, FIONBIO, &non_blocking);
#endif

		struct sockaddr_in sock_addr;
		sock_addr.sin_family = AF_INET;
		sock_addr.sin_addr.s_addr = inet_addr(ip.c_str());
		sock_addr.sin_port = htons(port);
		int rt = ::connect(m_sock, (struct sockaddr*)&sock_addr, sizeof(sock_addr));
		if (rt != 0) {
			struct timeval tv = { 2, 0 };
			fd_set writefds;
			FD_ZERO(&writefds);
			FD_SET(m_sock, &writefds);
			if (select(m_sock + 1, 0, &writefds, 0, &tv) > 0) {
#ifdef WIN32
				int error;
				int err_len = sizeof(error);
				getsockopt(m_sock, SOL_SOCKET, SO_ERROR, (char*)&error, &err_len);
#else
				int error;
				socklen_t err_len = sizeof(error);
				getsockopt(m_sock, SOL_SOCKET, SO_ERROR, &error, &err_len);
#endif
				if (error == 0) {
					ret = true;
				}
				else {
					ret = false;
				}
			}
			else {
				ret = false;
			}
		}
		else {
			ret = true;
		}
		if (ret)
			break;
	}

	//if (!ret)
	//    return false;

#ifdef WIN32
	ioctlsocket(m_sock, FIONBIO, (unsigned long*)&blocking);
#else
	ioctl(m_sock, FIONBIO, &blocking);
#endif	

#if(CC_TARGET_PLATFORM==CC_PLATFORM_WIN32)
	DWORD tv2 = m_recvTimeout * 1000;
#else
	struct timeval tv2 = { m_recvTimeout, 0 };
#endif
	//接收不设置超时
	//setsockopt(m_sock, SOL_SOCKET, SO_RCVTIMEO, (char*)&tv2, sizeof(tv2));

	m_connected = ret;
	m_requests.clear();

	if (ret)
		startRecvThread();

	MsgPack* msg = new MsgPack(this->isConnected() ? 1 : 2);//conn return
	if (msg) {
		NETLOG("connect msg: %d", msg->getMsgID());
		m_mutex.lock();
		m_responseList.push_back(MsgPack_ptr(msg));
		m_mutex.unlock();
	}
	/*
	cocos2d::Director::getInstance()->getScheduler()->performFunctionInCocosThread([&, this]{
	std::string sout;
	g_pLuaSystem->callLuaFunc(this->m_callback, this->isConnected()?1:0, sout);
	});
	*/

	//int ret = CCLuaEngine::defaultEngine()->getLuaStack()->executeFunctionByHandler(_nStartReconnectHandler, 0);

	return ret;

}

bool SocketClient::connect(std::string ip, unsigned short port, int timeout)
{
	auto t = std::thread(&SocketClient::_connect_timeout, this, ip, port, timeout);
	t.detach();
	return true;
}

bool SocketClient::_connect_block(std::string ip, unsigned short port) {
	//std::lock_guard<std::mutex> lock(m_mutex);
	if (m_connected) {
		return true;
	}
	close();// add close
	m_mutex.lock();
//	for (auto i : m_responseList)
//		delete i;
	m_responseList.clear();
	m_mutex.unlock();

	m_sock = socket(AF_INET, SOCK_STREAM, 0);
	{
		/*
		#if(CC_TARGET_PLATFORM==CC_PLATFORM_WIN32)
		DWORD tv = 10000;
		DWORD tv2 = 10000;
		#else
		struct timeval tv = {10,0};
		struct timeval tv2 = {10,0};
		#endif
		setsockopt(m_sock, SOL_SOCKET, SO_SNDTIMEO, (char*)&tv, sizeof(tv));
		setsockopt(m_sock, SOL_SOCKET, SO_RCVTIMEO, (char*)&tv2, sizeof(tv2));
		*/

		bool bKeppAlive = true;
		setsockopt(m_sock, SOL_SOCKET, SO_KEEPALIVE, (char*)&bKeppAlive, sizeof(bKeppAlive));

		int rcv_size0;
		int optLen = sizeof(rcv_size0);
		//getsockopt(m_sock, SOL_SOCKET, SO_RCVBUF, (char*)&rcv_size0, &optLen);
		//NETLOG("socket-recv-buffer= %d %d", rcv_size0, optLen);

		int rcv_size = 1024 * 1024;
		setsockopt(m_sock, SOL_SOCKET, SO_RCVBUF, (char*)&rcv_size, sizeof(int));

		//getsockopt(m_sock, SOL_SOCKET, SO_RCVBUF, (char*)&rcv_size0, &optLen);
		//NETLOG("socket-recv-buffer= %d %d", rcv_size0, optLen);

	}

	struct sockaddr_in svraddr;
	svraddr.sin_family = AF_INET;
	svraddr.sin_addr.s_addr = inet_addr(ip.c_str());
	svraddr.sin_port = htons(port);
	int rt = ::connect(m_sock, (struct sockaddr*)&svraddr, sizeof(svraddr));
	if (rt == SOCKET_ERROR) {
		return false;
	}
	m_requests.clear();
	m_connected = true;

	startRecvThread();

	return true;
}
/*
static const std::string key = "P%2BViyZLtO^gRT2";
bool encryptByAES(std::string in, std::string& out, bool isEncrypt) {
std::string iv;
iv.resize(AES_BLOCK_SIZE, 0);
AES_KEY aesKey;
int enc(AES_ENCRYPT);
if (isEncrypt) {
AES_set_encrypt_key((const unsigned char*)key.c_str(),
AES_BLOCK_SIZE * 8, &aesKey);
} else {
AES_set_decrypt_key((const unsigned char*)key.c_str(),
AES_BLOCK_SIZE * 8, &aesKey);
enc = AES_DECRYPT;
}
short inLen = in.length();
short block = (inLen + AES_BLOCK_SIZE - 1) / AES_BLOCK_SIZE;
in.resize(AES_BLOCK_SIZE * block);
out.resize(inLen);
short begin = 0;
try {
while (true) {
int len = inLen - begin >= AES_BLOCK_SIZE ?
AES_BLOCK_SIZE : inLen - begin;
AES_cbc_encrypt((unsigned char*)&inStr[begin], (unsigned char*)&outStr[begin], AES_BLOCK_SIZE, &aesKey,
(unsigned char*)&iv[0], enc);
begin += AES_BLOCK_SIZE;
if (begin >= inLen) {
break;
}
}
} catch (...) {
SYLAR_LOG_ERROR(g_logger) << "Bad aes buffer";
return false;
}
out.resize(inLen);
return true;
}*/

int SocketClient::send(const char* buf, int len){
	if (!m_connected) {
		return -1;
	}
	int bytes = 0;
	int count = 0;
	while (count < len) {
		bytes = ::send(m_sock, buf + count, len - count, 0);
		if (bytes == -1 || bytes == 0) {
			m_connected = false;
			NETLOG("socket send false");
			return -1;
		}
		count += bytes;
	}
	return count;
}

int SocketClient::recv(char* buf, int len) {
	if (!m_connected) {
		return -1;
	}
	return ::recv(m_sock, buf, len, MSG_WAITALL);
/*	
	int bytes = 0;
	int count = 0;
	while (count < len) {
		bytes = ::recv(m_sock, buf + count, len - count, MSG_WAITALL);
		if (bytes < 0) {
			m_connected = false;
			NETLOG("socket recv false");
			return count;
		}
		count += bytes;
	}
	return count;
	*/
	
}

void SocketClient::startRecvThread() {
	if (m_thread) {
		//CCLOG("Recv Thread is Running");
		return;
	}
	m_thread.reset(new std::thread(&SocketClient::runRecv, this));
	if (!m_thread->joinable())
		m_thread->join();
}

bool SocketClient::sendRequest(MsgPack* reqest)
{
	MsgPack* buf = reqest;
	int len = buf->getLen();

	return send(buf->getData(), len) > 0;
}

//char recbuff[4096];

MsgPack* SocketClient::recvMessage()
{
	MsgHeader header;

	if (recv((char *)&header, sizeof(MsgHeader)) <= 0)
		return 0;

	MsgPack* buff = new MsgPack(header);
	int lenLeft = recv(buff->getData(), header.nPkgSize);
	NETLOG("-----recv-end   = %d", lenLeft);
	if (lenLeft <= 0)
	{
		delete buff;
		return 0;
	}
	//msg read begin
	buff->setCur(0);
	return buff;

}

void SocketClient::runRecv() {
	while (m_running) {
#ifdef WIN32
		Sleep(1);
#else
		usleep(1);
#endif

		try {
			while (m_connected)
			{
				MsgPack* msg = recvMessage();
				if (msg) {
					NETLOG("is in runRecv msgid:%d msgLen:%d ", msg->getMsgID(), msg->getMax());
					MsgHeader &header = msg->getHeader();
					if (header.nPkgTotall > 1)
					{
						NETLOG("%d / %d", header.nPkgIndex, header.nPkgTotall);
					}
					m_mutex.lock();
					m_responseList.push_back(MsgPack_ptr(msg));
					m_mutex.unlock();
				}
			}
		}
		catch (std::bad_alloc& e) {
			NETLOG("recv-exception %s", e.what());
		}
		catch (...) {
			NETLOG("recv-exception unkown.");
		}
	}
	NETLOG("exit thread runRecv.");
}


void SocketClient::lockMutex()
{
	m_mutex.lock();
}
void SocketClient::unlockMutex()
{
	m_mutex.unlock();
}

　　以下我的定义的消息头，当然其实，有的东西其实是没必要的

typedef struct
{
	int32_t nId; //消息ID
	int32_t nPkgTotall; //总包数
	int32_t nPkgIndex;  //当前包
	int32_t nPkgSize; //当前包内容大小
} MsgHeader;

MsgHeader* createMsgHeader(int32_t msgId, int32_t pkgTotal, int32_t pkgIndex, int32_t pkgSz);

　　

 

服务端：

NetServer.h

#pragma once
//服务器

#define _MAX_HEAD 64				//最大消息头大小
#define _MAX_MSGSIZE 8 * 1024		// 暂定一个消息最大为16k
#define BLOCKSECONDS	3			// INIT函数阻塞时间
#define INBUFSIZE	(64*1024)		//?	具体尺寸根据剖面报告调整  接收数据的缓存
#define OUTBUFSIZE	(4*1024-64)		//? 具体尺寸根据剖面报告调整。 发送数据的缓存，当不超过8K时，FLUSH只需要SEND一次

//Server.cpp
#include <iostream>
#include <winsock2.h>

#pragma comment(lib, "ws2_32.lib")
DWORD WINAPI ClientThread(LPVOID lpParameter);
DWORD WINAPI server_main(PVOID pParam);
void CALLBACK startServer(HWND hWnd, UINT nMsg, UINT nTimerid, DWORD dwTime);

bool SendMsg(int msgId ,void* pBuf, int nSize, SOCKET &m_sockClient);

　NetServer.cpp

//Server.cpp
#include "stdafx.h"
#include "Server.h"
#include "WinServerDlg.h"
#include <fstream>
#include <vector>
#include "msg.h"
#include "Business.h"
using namespace std;

#define PORT 9001
#define IP_ADDRESS "127.0.0.1"

bool GetLocalIP(char* ip)
{
	//1.初始化wsa
	WSADATA wsaData;
	int ret = WSAStartup(MAKEWORD(2, 2), &wsaData);
	if (ret != 0)
	{
		return false;
	}
	//2.获取主机名
	char hostname[256];
	ret = gethostname(hostname, sizeof(hostname));
	if (ret == SOCKET_ERROR)
	{
		return false;
	}
	//3.获取主机ip
	HOSTENT* host = gethostbyname(hostname);
	if (host == NULL)
	{
		return false;
	}
	//4.转化为char*并拷贝返回
	strcpy(ip, inet_ntoa(*(in_addr*)*host->h_addr_list));
	return true;
}


DWORD WINAPI ClientThread(LPVOID lpParameter)
{
	SOCKET CientSocket = (SOCKET)lpParameter;
	int Ret = 0;
	char RecvBuffer[_MAX_MSGSIZE];
	int headLen = sizeof(MsgHeader);

	//包队列，出现多包时，合包用
	MsgBigCache* vecMsg[100];
	memset(vecMsg, 0, sizeof(vecMsg));

	while (true)
	{
		memset(RecvBuffer, 0x00, sizeof(RecvBuffer));
		Ret = recv(CientSocket, RecvBuffer, _MAX_MSGSIZE, 0);
		if (Ret == 0 || Ret == SOCKET_ERROR)
		{
			Log("客户端退出!");
			break;
		}

		MsgHeader *header = reinterpret_cast<MsgHeader*> (RecvBuffer);
		if (header->nPkgTotall > 1)
		{
			CString str;
			str.Format("此消息共%d ,当前%d", header->nPkgTotall, header->nPkgIndex);
			Log(str);
		}
//		Log("接收到客户信息为:%s", RecvBuffer+headLen);
		if (1 == header->nPkgTotall)
		{
			progressMsg(RecvBuffer + headLen, CientSocket, header);
		}
		else if (header->nPkgTotall > 1000 || header->nPkgTotall < 0)
		{
			return 0;
		}
		else
		{
			MsgBigCache *msg = new MsgBigCache();
			char *buff = new char[header->nPkgSize];
			memset(buff, 0, header->nPkgSize);
			memcpy(buff, RecvBuffer + headLen, header->nPkgSize);
			msg->header = *header;
			msg->buff = buff;
			vecMsg[header->nPkgIndex] = msg;
			
			bool allRec = true;
			for (int i = 1; i <= header->nPkgTotall; i++)
			{
				if (!vecMsg[i])
				{
					allRec = false;
					break;
				}
			}

			if (allRec)
			{
				ofstream f("1.png" , ios_base::binary);
				for (int i = 1; i <= header->nPkgTotall; i++)
				{
					f.write((char *)vecMsg[i]->buff, vecMsg[i]->header.nPkgSize);
					delete vecMsg[i]->buff;
					delete vecMsg[i];
				}
				memset(vecMsg,0,sizeof(vecMsg));
			}
		}
	}

	return 0;
}

DWORD WINAPI server_main(PVOID pParam)
{
	WSADATA  Ws;
	SOCKET ServerSocket, ClientSocket;
	struct sockaddr_in LocalAddr, ClientAddr;
	int Ret = 0;
	int AddrLen = 0;
	HANDLE hThread = NULL;
	
	char buffIp[128];
	char buffPort[20];
	ifstream f("config.ini");
	if (f.fail())
	{
		GetLocalIP(buffIp);
		strcpy(buffPort, "9001");
	}
	else
	{
		f.getline(buffIp, sizeof(buffIp));
		f.getline(buffPort, sizeof(buffPort));
	}
	int port = atoi(buffPort);
	//Init Windows Socket
	if (WSAStartup(MAKEWORD(2, 2), &Ws) != 0)
	{
		Log("Init Windows Socket Failed::%d" , GetLastError());
		return -1;
	}

	//Create Socket
	ServerSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	if (ServerSocket == INVALID_SOCKET)
	{
		Log ("Create Socket Failed::%d" ,GetLastError());
		return -1;
	}

	LocalAddr.sin_family = AF_INET;
	LocalAddr.sin_addr.s_addr = inet_addr(buffIp);
	LocalAddr.sin_port = htons(port);
	memset(LocalAddr.sin_zero, 0x00, 8);

	//Bind Socket
	Ret = ::bind(ServerSocket, (struct sockaddr*)&LocalAddr, sizeof(LocalAddr));
	if (Ret != 0)
	{
		Log("Bind Socket Failed::%d", GetLastError());
		return -1;
	}
	//listen
	Ret = listen(ServerSocket, 10);
	if (Ret != 0)
	{
		Log("listen Socket Failed::", GetLastError());
		return -1;
	}

	Log("服务端启动成功！");
	Log(CString("ip：") + buffIp);
	Log(CString("port:") + buffPort);

	while (true)
	{
		AddrLen = sizeof(ClientAddr);
		ClientSocket = accept(ServerSocket, (struct sockaddr*)&ClientAddr, &AddrLen);
		if (ClientSocket == INVALID_SOCKET)
		{
			Log ("Accept Failed::%d" , GetLastError());
			break;
		}
		CString str;
		str.Format("客户端连接:: %s : %d", inet_ntoa(ClientAddr.sin_addr), ClientAddr.sin_port);
		Log(str);
		sendConnSuc(ClientSocket);

		hThread = CreateThread(NULL, 0, ClientThread, (LPVOID)ClientSocket, 0, NULL);
		if (hThread == NULL)
		{
			Log("Create Thread Failed!");
			break;
		}

		CloseHandle(hThread);
	}

	closesocket(ServerSocket);
	closesocket(ClientSocket);
	WSACleanup();

	return 0;
}

bool g_start = false;


void CALLBACK startServer(HWND hWnd, UINT nMsg, UINT nTimerid, DWORD dwTime)
{
	if (g_start)
		return;
//	server_main(0);
	auto hThread = CreateThread(NULL, 0, server_main, 0, 0, NULL);
//	if (hThread == NULL)
//	{
//		Log("Create Thread Failed!");
//	}
//	else
		g_start = true;
	KillTimer(hWnd, nTimerid);
}

////////////////////////////////////////发消息给客户端

bool hasError()
{
#ifdef WIN32
	int err = WSAGetLastError();
	if (err != WSAEWOULDBLOCK) {
#else
	int err = errno;
	if (err != EINPROGRESS && err != EAGAIN) {
#endif
		return true;
	}

	return false;
}

char m_bufOutput[_MAX_MSGSIZE];
int m_nOutbufLen;
bool SendMsg(int msgid, void* pBuf, int nSize, SOCKET &m_sockClient)
{
	if (pBuf == 0 || nSize <= 0) {
		return false;
	}

	if (m_sockClient == INVALID_SOCKET) {
		return false;
	}
	char *contentBuff = (char *)pBuf;
	MsgHeader header;
	header.nId = msgid;
	header.nPkgIndex = 1;
	header.nPkgTotall = 1;// ceil(1.0 * nSize / OUTBUFSIZE);
	int headLen = sizeof(header);

	header.nPkgSize = nSize;
	m_nOutbufLen = nSize + headLen;
	CString str;
	str.Format("回复消息%d ！大小 %d 个字节！", header.nId, header.nPkgSize);
	Log(str);
	
	// 发送一段数据
	int	outsize;
	outsize = send(m_sockClient, (char*)&header, headLen, 0);
	if (outsize <= 0)
	{
		if (hasError())
		{
			Log("发送失败，中断！");
			return false;
		}
	}
	outsize = send(m_sockClient, (char*)pBuf, nSize, 0);
	if (outsize <= 0)
	{
		if (hasError())
		{
			Log("发送失败，中断！");
			return false;
		}
	}

	return true;
}