hiredis和rapidjson库的使用小结

杨永贞

发布于 2022-05-07 10:22:37

1.1K0

发布于 2022-05-07 10:22:37

文章被收录于专栏：独行猫a的沉淀积累总结

Hiredis 简介

Hiredis 是Redis官方发布的C版本客户端 hiredis库。redis的源码中也有使用hiredis。比如redis-cli和Redis中的哨兵机制和主从机制，集群等都使用了hiredis。

hiredis 提供了同步、异步访问，异步 API需要与一些事件库协同工作。

它的大致工作流程：

建立连接->发送命令->等待结果并处理->释放连接。

Hiredis简单使用

使用中也遇到过一些坑，这里一并总结下。

坑一、比如那个mset批量提交数据指令。看下面的代码：

    // mset key1 value1 key2 value2 ........
	int ret = REDIS_ERR;
	std::ostringstream out;
	for (auto val : keys_vals) {
	   out << " "<< val.first << " " << val.second;
	}
	reply = (redisReply *)redisCommand(context, "MSET %s", out.str().c_str());
	if (reply != nullptr) {
	    serverLog(LL_NOTICE, "mSetWithCommit:%s\n", reply->str);
	    freeReplyObject(reply);
	    //serverLog(LL_NOTICE, "ok\n");
	    //return REDIS_OK;
	}

其实是有问题的，若value中有空格，就会报：Hiredis MSET,error:ERR wrong number of arguments for MSET。

使用hiredis的API进行调用时如果是如下命令：

hmset userid:1001 username 'xiao ming'

这种语法，使用redis-cli是没有问题的，但如果使用hiredis就会有问题。

报ERR wrong number of arguments for HMSET错误。

原因就是xiao ming那有个空格，他当成了username 'xiao，另外一个就是 ming'后面缺值，就报错了。这里有点坑。

解决办法：使用redisCommandArgv拼接。

//! \brief 批量提交数据入库
//! \param keys_vals 
//! \return REDIS_OK/REDIS_ERR
int mSetWithCommit(const std::map<std::string, std::string>& keys_vals) {
	serverLog(LL_NOTICE, "->mSetWithCommit:\n");

	redisReply *reply;
	if (context == nullptr) return REDIS_ERR;

	int ret = REDIS_ERR;
	
	std::vector<std::string> tVec;
	tVec.push_back("MSET");
	for (auto it : keys_vals) {
		tVec.push_back(it.first);
		tVec.push_back(it.second);
	}
	std::vector<const char *> argv(tVec.size());
	std::vector<size_t> argvlen(tVec.size());
	int j = 0;
	for (auto i = tVec.begin(); i != tVec.end(); ++i, ++j){
		argv[j] = i->c_str();
		argvlen[j] = i->length();
	}
	reply = (redisReply *)redisCommandArgv(context, argv.size(), &(argv[0]), &(argvlen[0]));
	if (reply != nullptr) {
		ret = REDIS_OK;
		if (reply->type == REDIS_REPLY_ERROR) {
			serverLog(LL_WARNING, "MSET error,error:%s\n", reply->str);
			ret = REDIS_ERR;
		}
		freeReplyObject(reply);
	}
	if (ret == REDIS_OK) {
		serverLog(LL_NOTICE, "ok\n");
	}
	return ret;
}

这还没完，坑二：

使用mget时也可能遇到坑。比如，假如value值为空，或者key为""时会怎样？

经测试验证value值为空时倒也不影响。问题出在类型上，假若有其他类型如list,

mget批量获取后，key为list类型的，会返回nil

使用redisCommand接口，mget了1000个key,结果竟然返回了999个，差了一个。导致郁闷的不知道如何修复。好在，在测试客户端中验证都是正常的，有解决办法了。

对这种mget和mset设置多个数据项的，安全起见统一使用redisCommandArgv吧。

//! \brief MGET批量读取数据
//! \param keys,values
//! \return REDIS_OK/REDIS_ERR
int mGetAllValues(const std::vector<std::string> &keys, std::vector<std::string> &values) {
	serverLog(LL_DEBUG, "mGetAllValues:\n");

	redisReply *reply;
	if (context == nullptr) return REDIS_ERR;

	int ret = REDIS_ERR;

	std::vector<std::string> tVec;
	tVec.emplace_back("MGET");
	for (auto it : keys) {
		tVec.emplace_back(it);
	}
	std::vector<const char *> argv(tVec.size());
	std::vector<size_t> argvlen(tVec.size());
	size_t j = 0;
	for (auto i = tVec.begin(); i != tVec.end(); i++, j++) {
		argv[j] = i->c_str();
		argvlen[j] = i->length();
	}
	reply = (redisReply *)redisCommandArgv(context, argv.size(), &(argv[0]), &(argvlen[0]));
	if (reply != nullptr) {
		if (reply->type == REDIS_REPLY_ERROR) {
			freeReplyObject(reply);
			return REDIS_ERR;
		}
		redisReply** repy;
		int count = reply->elements;
		serverLog(LL_DEBUG, "count:%d\n", count);
		repy = reply->element;
		for (int i = 0; i < count; i++) {
			// 必须判空
			if ((*repy)->str) {
				values.emplace_back((*repy)->str);
			}else {
				serverLog(LL_WARNING, "value is null,key=%s\n", argv[i+1]);
				values.emplace_back("");
			}
			repy++;
		}
		freeReplyObject(reply);
		serverLog(LL_DEBUG, "ok\n");
		return REDIS_OK;
	}
	return REDIS_ERR;
}

连接相关

// Redis连接配置相关
static const char* HostIP = "127.0.0.1";
static const char* Auth = "XXXXX";
static const int Hostport = 6379;

static redisContext *context;

/* Connect to the server. If force is not zero the connection is performed
* even if there is already a connected socket. */
static int cliConnect(int force) {
	serverLog(LL_NOTICE, "->cliConnect:\n");
	if (context == NULL || force) {
		if (context != NULL) {
			redisFree(context);
		}
		context = redisConnect(HostIP, Hostport);
		if (context->err) {
			fprintf(stderr, "Could not connect to Redis at ");
			fprintf(stderr, "%s: %s\n", HostIP, context->errstr);
			redisFree(context);
			context = NULL;
			return REDIS_ERR;
		}

		// Do AUTH and select the right DB
		if (cliAuth(Auth) != REDIS_OK)
			return REDIS_ERR;
		if (cliSelect(0) != REDIS_OK)
			return REDIS_ERR;
		serverLog(LL_NOTICE, "OK\n");
		return REDIS_OK;
	}
	return REDIS_ERR;
}

/* Send AUTH command to the server */
static int cliAuth(const char* auth) {
	redisReply *reply;
	if (auth == NULL) return REDIS_OK;

	reply = (redisReply *)redisCommand(context, "AUTH %s", auth);
	if (reply != NULL) {
		freeReplyObject(reply);
		return REDIS_OK;
	}
	return REDIS_ERR;
}

/* Send SELECT dbnum to the server */
static int cliSelect(int dbnum) {
	redisReply *reply;
	if (dbnum == 0) return REDIS_OK;
	reply = (redisReply *)redisCommand(context, "SELECT %d", dbnum);
	if (reply != NULL) {
		int result = REDIS_OK;
		if (reply->type == REDIS_REPLY_ERROR) result = REDIS_ERR;
		freeReplyObject(reply);
		return result;
	}
	return REDIS_ERR;
}

常用接口封装

//! \brief 建立连接
//! \param force 
//! \return REDIS_OK/REDIS_ERR
int wrapper_cliConnect(int force) {
	return cliConnect(force);
}

//! \brief 获取单个的key-value(String)
//! \param key 
//! \param value 
//! \return REDIS_OK/REDIS_ERR
int getValueString(const std::string &key, std::string &value) {
	redisReply *reply;

	if (context == nullptr) return REDIS_ERR;

	reply = (redisReply *)redisCommand(context, "GET %s", key.c_str());
	if (reply != NULL) {
		int result = REDIS_ERR;
		if (reply->type == REDIS_REPLY_ERROR) result = REDIS_ERR;

		if (reply->type == REDIS_REPLY_STRING) {
			//int len = reply->len;
			//serverLog(LL_NOTICE, "reply->len =%d\n", len);
			//serverLog(LL_NOTICE, "reply->str =%s\n", reply->str);
			value = reply->str;
			//serverLog(LL_NOTICE, "%s\n", reply->str);
			result = REDIS_OK;
		}

		freeReplyObject(reply);
		return result;
	}
	return REDIS_ERR;
}

//! \brief 设置单个的key-value(String)
//! \param key 
//! \param value 
//! \return REDIS_OK/REDIS_ERR
int setValueString(const std::string &key, std::string &value) {
	redisReply *reply;

	if (context == nullptr) return REDIS_ERR;

	reply = (redisReply *)redisCommand(context, "SET %s %s", key.c_str(), value.c_str());
	if (reply != NULL) {
		int result = REDIS_OK;
		if (reply->type == REDIS_REPLY_ERROR) {
			serverLog(LL_WARNING, "SET error,key:%s,val:%s,error:%s\n", key.c_str(), value.c_str(), reply->str);
			result = REDIS_ERR;
		}
		freeReplyObject(reply);
		return result;
	}
	return REDIS_ERR;
}

//! \brief 获取所有的键值数据（不建议用keys*有影响）
//! \param keys_vals 
//! \return REDIS_OK/REDIS_ERR
int getAllKeysVals(std::map<std::string, std::string> &keys_vals) {
	redisReply *reply;
	int ret = REDIS_ERR;

	if (context == nullptr) return REDIS_ERR;

	reply = (redisReply *)redisCommand(context, "keys *");
	if (reply != nullptr) {
		redisReply** repy;
		int count = reply->elements;
		repy = reply->element;
		for (int i = 0; i < count; i++) {
			//存储key-val值
			std::string val;
			ret = getValueString((*repy)->str,val);
			if (ret == REDIS_OK) {
				keys_vals.emplace((*repy)->str, val);
			}else {
				serverLog(LL_WARNING, "GET error\n");
			}
			repy++;
		}
		freeReplyObject(reply);
		return REDIS_OK;
	}
	return REDIS_ERR;
	
}

//! \brief 获取所有的键值数据(SCAN 指令分批次读取 每次1000条)
//! \param keys_vals 
//! \return REDIS_OK/REDIS_ERR
int getAllKeysVals_S(std::map<std::string, std::string> &keys_vals) {
	redisReply *reply;
	int ret = REDIS_ERR;

	if (context == nullptr) return REDIS_ERR;

	int index = 0;
	do {
		reply = (redisReply *)redisCommand(context, "SCAN %d MATCH * COUNT 1000", index);
		if (reply == nullptr) return REDIS_ERR;

		if (reply->type != REDIS_REPLY_ARRAY) {
			freeReplyObject(reply);
			return ret;
		}

		index = atoi(reply->element[0]->str);
		//serverLog(LL_NOTICE,"index:%d", index);
		if (1 == reply->elements) {
			serverLog(LL_WARNING, "no data\n");
			freeReplyObject(reply);
			return ret;
		}
		if (reply->element[1]->type != REDIS_REPLY_ARRAY) {
			serverLog(LL_WARNING,"redis scan keys reply not array");
			freeReplyObject(reply);
			return ret;
		}
		int count = reply->element[1]->elements;
		//serverLog(LL_NOTICE, "count:%d", count);
		for (int i = 0; i < count; i++) {
			std::string val;
			std::string key = reply->element[1]->element[i]->str;
			//serverLog(LL_NOTICE,"i:%d,key:%s\n", i, key.c_str());
			ret = getValueString(key, val);
			if (ret == REDIS_OK) {
				keys_vals.emplace(key, val);
			}else {
				serverLog(LL_WARNING, "GET error\n");
			}
		}

	} while (0 != index);

	freeReplyObject(reply);
	return REDIS_OK;
}

//! \brief 批量提交数据入库
//! \param keys_vals 
//! \return REDIS_OK/REDIS_ERR
int mSetWithCommit(const std::map<std::string, std::string>& keys_vals) {
	serverLog(LL_NOTICE, "->mSetWithCommit:\n");

	redisReply *reply;
	if (context == nullptr) return REDIS_ERR;

	int ret = REDIS_OK;
	for (auto val : keys_vals) {
		ret = setValueString(val.first, val.second);
		if (ret != REDIS_OK) {
			serverLog(LL_WARNING, "SET error\n");
			return ret;
		}
	}
	serverLog(LL_NOTICE, "ok\n");
	/*
	// mset key1 value1 key2 value2 ........
	int ret = REDIS_ERR;
	std::ostringstream out;
	for (auto val : keys_vals) {
	out << " "<< val.first << " " << val.second;
	}
	reply = (redisReply *)redisCommand(context, "MSET %s", out.str().c_str());
	if (reply != nullptr) {
	serverLog(LL_NOTICE, "mSetWithCommit:%s\n", reply->str);
	freeReplyObject(reply);
	//serverLog(LL_NOTICE, "ok\n");
	//return REDIS_OK;
	}
	*/
	return ret;
}

//! \brief 获取当前日期 格式 YYYYMMDD 
//! \param 空 
//! \return string
std::string getNowDate() {
	time_t lt;
	struct tm * now;
	char tmbuf[64];

	lt = time(NULL);
	now = localtime(&lt);
	// %Y%m%d%H%M%S
	strftime(tmbuf, sizeof(tmbuf), "%Y%m%d", now);
	std::string nowDate(tmbuf);
	return std::move(nowDate);
}
//! \brief 存储json文件 二进制的形式保存和读取
//! \param strjson 
//! \return 0 成功 非0失败
int saveDumpJson(const std::string &strjson) {
	serverLog(LL_NOTICE, "->saveDumpJson:\n");
	std::ofstream outFile;
	outFile.open(TEMP_FILE_NAME, std::ios::binary);
	if (!outFile.is_open()) {
		outFile.close();
		serverLog(LL_NOTICE, "Error\n");
		return -1;
	}
	outFile << strjson << std::endl;
	outFile.close();
	// 先写临时文件，成功后移动操作为正式文件
	auto ret = MoveFileExA(TEMP_FILE_NAME.c_str(), (DUMP_DIR_PATH + DUMP_FILE_PRIFIX + getNowDate()).c_str(),
		MOVEFILE_REPLACE_EXISTING | MOVEFILE_COPY_ALLOWED | MOVEFILE_WRITE_THROUGH);
	if (ret) {
		serverLog(LL_NOTICE, "OK\n");
		return 0;
	}
	return -2;

}

// 读取文件操作
std::string readAll(const std::string& fileName){
	std::ifstream in(fileName, std::ios::binary);
	std::istreambuf_iterator<char> begin(in);
	std::istreambuf_iterator<char> end;
	return std::string{ begin, end };
}

RapidJSON简介

RapidJSON是腾讯开源的一个高效的C++ JSON解析器及生成器，它是只有头文件的C++库。RapidJSON是跨平台的，支持Windows, Linux, Mac OS X及iOS, Android。

它的源码在https://github.com/Tencent/rapidjson/，稳定版本为2016年发布的1.1.0版本。

RapidJSON特点

(1). RapidJSON小而全：它同时支持SAX和DOM风格的API，SAX解析器只有约500行代码。

(2). RapidJSON快：它的性能可与strlen()相比，可支持SSE2/SSE4.2加速，使用模版及内联函数去降低函数调用开销。

(3). RapidJSON独立：它不依赖于BOOST等外部库，它甚至不依赖于STL。

(4). RapidJSON对内存友好：在大部分32/64位机器上，每个JSON值只占16字节(除字符串外)，它预设使用一个快速的内存分配器，令分析器可以紧凑地分配内存。

(5). RapidJSON对Unicode友好：它支持UTF-8、UTF-16、UTF-32(大端序/小端序)，并内部支持这些编码的检测、校验及转码。例如，RapidJSON可以在分析一个UTF-8文件至DOM (Document Object Model, 文件对象模型)时，把当中的JSON字符串转码至UTF-16。它也支持代理对(surrogate pair)及"\u0000"(空字符)。

每个JSON值都储存为Value类，而Document类则表示整个DOM，它存储了一个DOM 树的根Value。RapidJSON的所有公开类型及函数都在rapidjson命名空间中。

解析和生成JSON的耗时（越低越好）：

解析至DOM后的内存用量（越低越好）：

简单使用

rapidjson的小坑，rapidjson::Document doc; doc.Parse时要看内容是否为空，为空则会崩。

还有，if (doc.Parse(contents.c_str()).HasParseError())判断是否是合法的json，这还不算完，

后面若要使用hasmember等，还会崩。因此还需要个判断，if (!doc.IsObject())。

std::string objectToString(const rapidjson::Value& valObj)
{
	rapidjson::StringBuffer buffer;
	rapidjson::Writer<rapidjson::StringBuffer> jWriter(buffer);
	valObj.Accept(jWriter);
	return buffer.GetString();
}

//! \brief 键值对转换为jsonString
//! \param keyVals 
//! \return string
std::string createJsonString(std::map<std::string, std::string> keyVals) {
	/*
	rapidjson::Document doc;
	rapidjson::Document::AllocatorType &allocator = doc.GetAllocator();
	doc.SetObject();
	//rapidjson::Value root(rapidjson::kObjectType);
	rapidjson::Value key(rapidjson::kStringType);
	rapidjson::Value value(rapidjson::kStringType);
	for (auto it : keyVals) {
	key.SetString(it.first.c_str(), allocator);
	value.SetString(it.second.c_str(), allocator);
	doc.AddMember(key, value, allocator);
	}

	rapidjson::StringBuffer buffer;
	rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
	doc.Accept(writer);
	return std::move(std::string(buffer.GetString()));
	*/
	rapidjson::StringBuffer strBuf;
	rapidjson::Writer<rapidjson::StringBuffer> writer(strBuf);
	writer.StartObject();
	for (auto it : keyVals) {
		writer.Key(it.first.c_str());
		writer.RawNumber(it.second.c_str(), it.second.length());
	}
	writer.EndObject();
	return std::move(std::string(strBuf.GetString()));
}

//! \brief 遍历文件夹
//! \brief 查找指定前缀的文件并获取文件名+sort排序 降序，日期 由大到小 
//! \param path,prefix,file_list
//! \return 0成功 非0失败
int loadFileList(const std::string path, const std::string prefix,
	std::vector<std::string>& file_list) {

	struct _finddata_t c_file;
	intptr_t   hFile;

	if (_chdir(path.c_str())) {
		serverLog(LL_WARNING, "Dir error,not exist:%s\n", path.c_str());
		return -1;
	}
	// 先查找第一个
	hFile = _findfirst((prefix+"*").c_str(), &c_file);
	if (hFile == -1) {
		serverLog(LL_WARNING,"No files in current directory!\n");
		return -2;
	}
	//serverLog(LL_NOTICE, "c_file.name:%s\n", c_file.name);
	file_list.emplace_back(c_file.name);
	// Find the rest of the files 
	while (_findnext(hFile, &c_file) == 0) {
		//serverLog(LL_NOTICE, "c_file.name:%s\n", c_file.name);
		file_list.emplace_back(c_file.name);
	}
	_findclose(hFile);

	// 排序 greater--降序 日期 由大到小 
	sort(file_list.begin(), file_list.end(), std::greater<std::string>());

	return 0;
}


//! \brief 加载数据
//! \brief
//! \param 空
//! \return 0成功 非0失败
int loadDumpJson() {
	std::vector<std::string> fileList;
	std::string loadFileName;

	// 遍历所有文件名
	loadFileList(DUMP_DIR_PATH, DUMP_FILE_PRIFIX, fileList);
	for (auto f : fileList) {
		serverLog(LL_NOTICE, "file:%s\n", f.c_str());
	}
	if (fileList.empty()) {
		serverLog(LL_WARNING, "fileList empty\n");
		return -1;
	}
	// 始终加载读取最近的一个备份文件
	loadFileName = DUMP_DIR_PATH + fileList[0];
	std::string contents = readAll(loadFileName);
	// 检测内容合法性
	rapidjson::Document doc;
	if (doc.Parse(contents.c_str()).HasParseError()) {
		std::ostringstream outmsg;
		outmsg << "loadDumpJson,name:%s,parse json error," << fileList[0] << doc.GetErrorOffset() << ", " << doc.GetParseError() << std::endl;
		serverLog(LL_WARNING, outmsg.str().c_str());
		return -1;
	}
	serverLog(LL_NOTICE, "load %s ok\n", fileList[0].c_str());
	// 包装成key-value形式的数据
	std::map<std::string, std::string> key_values;
	for (auto item = doc.MemberBegin(); item != doc.MemberEnd(); ++item) {
		std::string key = item->name.GetString();
		//!!注意这里不能再用objectToString包装一次，否则会出现多个转义字符
		//std::string value = objectToString(item->value).c_str();
		std::string value = item->value.GetString();
		key_values.insert(std::make_pair(key.c_str(), value));
	}
	return 0;

}
//! \brief 存储json文件 二进制的形式保存和读取
//! \param strjson 
//! \return 0 成功 非0失败
int saveDumpJson(std::string &strjson) {
	serverLog(LL_NOTICE, "->saveDumpJson:\n");
	std::ofstream outFile;
	outFile.open(TEMP_FILE_NAME, std::ios::binary);
	if (!outFile.is_open()) {
		outFile.close();
		serverLog(LL_NOTICE, "Error\n");
		return -1;
	}
	outFile << strjson << std::endl;
	outFile.close();
	// 先写临时文件，成功后移动操作为正式文件
	auto ret = MoveFileExA(TEMP_FILE_NAME.c_str(), (DUMP_DIR_PATH + DUMP_FILE_PRIFIX + getNowDate()).c_str(),
		MOVEFILE_REPLACE_EXISTING | MOVEFILE_COPY_ALLOWED | MOVEFILE_WRITE_THROUGH);
	if (ret) {
		serverLog(LL_NOTICE, "OK\n");
		return 0;
	}
	return -2;

}

//! \brief 备份机制工作线程
//! \brief 执行连接redis并存储数据为dump备份文件操作
//! \param 空
//! \return 
DWORD WINAPI SaveWorkerThread(LPVOID lpParam) {
	serverLog(LL_NOTICE, "SaveWorkerThread: ENTER\n");

	// 创建事件对象
	gWorkEvent = CreateEvent(NULL, TRUE, FALSE, TEXT("gWorkEvent"));
	// 等待线程启动信号
	WaitForSingleObject(gWorkEvent,INFINITE);
	// 清除信号
	ResetEvent(gWorkEvent);
	// 加载备份数据入库
	loadDumpJson();
	serverLog(LL_NOTICE, "SaveWorkerThread: In\n");
	while (true) {
		serverLog(LL_NOTICE, "SaveWorkerThread: Wait...\n");
		DWORD waitResult = WaitForSingleObject(
			gWorkEvent, 
			INFINITE);   
		doSaveWork();
		// 清除信号
		ResetEvent(gWorkEvent);
	}
}


//! \brief 启动工作线程,外部调用(server.c中)
//! \param 空
//! \return 
void StartSaveWorkerThread() {
	serverLog(LL_NOTICE, "SaveWorkerThread: Start\n");
	if (!SetEvent(gWorkEvent)) {
		serverLog(LL_NOTICE, "Start SaveWorkerThread failed (%d)\n", GetLastError());
	}
}