首页 / REDIS / PostgreSQL异步客户端（并模拟redis 数据结构）

PostgreSQL异步客户端（并模拟redis 数据结构）

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了PostgreSQL异步客户端（并模拟redis 数据结构），小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含11801字，纯文字阅读大概需要17分钟。

内容图文

以前为了不在游戏逻辑（对象属性）变更时修改数据库，就弄了个varchar字段来表示json，由服务器逻辑（读取到内存）去操作它。

但这对运维相当不友好，也不能做一些此Json数据里查询。

所以后面就用了下ssdb，然而就在前几天才了解到postgresql支持json了（其实早在两年前就行了吧···）

就这点差不多就可以算当作mongodb用了，不过还是不支持redis的高级数据结构。

于是我就想模拟（实现）下redis（的数据结构）。

就抽空看了下它的c api库：libpq，发现其请求-等待模型，在网络延迟高的时候，特别影响qps。所以我就写了一个异步客户端，并简易模拟了redis的kv，hash。

开8个链接到pg server，其速度比1个链接快5倍。在我的测试中，每秒打到30k QPS

（目前不支持list，以及后期还要通过储存过程对现在的hash实现进行改造优化）

#include <string>
#include <list>
#include <iostream>
#include <unordered_map>
#include <memory>
#include <queue>
#include <assert.h>
#include <functional>
#include <sstream>
#include <chrono>

#include "fdset.h"

#include "libpq-events.h"
#include "libpq-fe.h"
#include "libpq/libpq-fs.h"usingnamespace std;

class AsyncPGClient
{
public:
    /*TODO::传递错误信息*/
    typedef std::function<void(const PGresult*)> RESULT_CALLBACK;
    typedef std::function<void(bool value)> BOOL_RESULT_CALLBACK;
    typedef std::function<void(conststring& value)> STRING_RESULT_CALLBACK;
    typedef std::function<void(const std::unordered_map<string, string>& value)> STRINGMAP_RESULT_CALLBACK;

    AsyncPGClient() : mKVTableName("kv_data"), mHashTableName("hashmap_data")
    {
        mfdset = ox_fdset_new();
    }

    ~AsyncPGClient()
    {
        for (auto& kv : mConnections)
        {
            PQfinish((*kv.second).pgconn);
        }

        ox_fdset_delete(mfdset);
        mfdset = nullptr;
    }

    voidget(conststring& key, const STRING_RESULT_CALLBACK& callback = nullptr)
    {
        mStringStream << "SELECT key, value FROM public." << mKVTableName << " where key = ‘" << key << "‘;";

        postQuery(mStringStream.str(), [callback](const PGresult* result){
            if (callback != nullptr && result != nullptr)
            {
                if (PQntuples(result) == 1 && PQnfields(result) == 2)
                {
                    callback(PQgetvalue(result, 0, 1));
                }
            }
        });
    }

    voidset(conststring& key, conststring& v, const BOOL_RESULT_CALLBACK& callback = nullptr)
    {
        mStringStream << "INSERT INTO public." << mKVTableName << "(key, value) VALUES(‘" << key << "‘, ‘" << v << "‘) ON CONFLICT(key) DO UPDATE SET value = EXCLUDED.value;";

        postQuery(mStringStream.str(), [callback](const PGresult* result){
            if (callback != nullptr)
            {
                if (PQresultStatus(result) == PGRES_COMMAND_OK)
                {
                    callback(true);
                }
                else
                {
                    cout << PQresultErrorMessage(result);
                    callback(false);
                }
            }
        });
    }

    void    hget(conststring& hashname, conststring& key, const STRING_RESULT_CALLBACK& callback = nullptr)
    {
        hmget(hashname, { key }, [callback](const std::unordered_map<string, string>& value){
            if (callback != nullptr && !value.empty())
            {
                callback((*value.begin()).second);
            }
        });
    }

    void    hmget(conststring& hashname, const std::vector<string>& keys, const STRINGMAP_RESULT_CALLBACK& callback = nullptr)
    {
        mStringStream << "SELECT key, value FROM public." << mHashTableName << " where ";
        auto it = keys.begin();
        do
        {
            mStringStream << "key=‘" << (*it) << "‘";

            ++it;
        } while (it != keys.end() && &(mStringStream << " or ") != nullptr);
        mStringStream << ";";

        postQuery(mStringStream.str(), [callback](const PGresult* result){
            if (callback != nullptr)
            {
                std::unordered_map<string, string> ret;
                if (PQresultStatus(result) == PGRES_TUPLES_OK)
                {
                    int num = PQntuples(result);
                    int fileds = PQnfields(result);
                    if (fileds == 2)
                    {
                        for (int i = 0; i < num; i++)
                        {
                            ret[PQgetvalue(result, i, 0)] = PQgetvalue(result, i, 1);
                        }
                    }
                }

                callback(ret);
            }
        });
    }

    void    hset(conststring& hashname, conststring& key, conststring& value, const BOOL_RESULT_CALLBACK& callback = nullptr)
    {
        mStringStream << "INSERT INTO public." << mHashTableName << "(hashname, key, value) VALUES(‘" << hashname << "‘, ‘" << key << "‘, ‘" << value
            << "‘) ON CONFLICT (hashname, key) DO UPDATE SET value = EXCLUDED.value;";

        postQuery(mStringStream.str(), [callback](const PGresult* result){
            if (callback != nullptr)
            {
                callback(PQresultStatus(result) == PGRES_COMMAND_OK);
            }
        });
    }

    void  hgetall(conststring& hashname, const STRINGMAP_RESULT_CALLBACK& callback = nullptr)
    {
        mStringStream << "SELECT key, value FROM public." << mHashTableName << " where hashname = ‘" << hashname << "‘;";
        postQuery(mStringStream.str(), [callback](const PGresult* result){
            if (callback != nullptr)
            {
                std::unordered_map<string, string> ret;
                if (PQresultStatus(result) == PGRES_TUPLES_OK)
                {
                    int num = PQntuples(result);
                    int fileds = PQnfields(result);
                    if (fileds == 2)
                    {
                        for (int i = 0; i < num; i++)
                        {
                            ret[PQgetvalue(result, i, 0)] = PQgetvalue(result, i, 1);
                        }
                    }
                }

                callback(ret);
            }
        });
    }

    void    postQuery(conststring&& query, const RESULT_CALLBACK& callback = nullptr)
    {
        mPendingQuery.push({ std::move(query), callback});
        mStringStream.str(std::string());
        mStringStream.clear();
    }

    void    postQuery(conststring& query, const RESULT_CALLBACK& callback = nullptr)
    {
        mPendingQuery.push({ query, callback });
        mStringStream.str(std::string());
        mStringStream.clear();
    }

public:
    void    poll(int millSecond)
    {
        ox_fdset_poll(mfdset, millSecond);

        std::vector<int> closeFds;

        for (auto& it : mConnections)
        {
            auto fd = it.first;
            auto connection = it.second;
            auto pgconn = connection->pgconn;

            if (ox_fdset_check(mfdset, fd, ReadCheck))
            {
                if (PQconsumeInput(pgconn) > 0 && PQisBusy(pgconn) == 0)
                {
                    bool successGetResult = false;

                    while (true)
                    {
                        auto result = PQgetResult(pgconn);
                        if (result != nullptr)
                        {
                            successGetResult = true;
                            if (connection->callback != nullptr)
                            {
                                connection->callback(result);
                                connection->callback = nullptr;
                            }
                            PQclear(result);
                        }
                        else
                        {
                            break;
                        }
                    }

                    if (successGetResult)
                    {
                        mIdleConnections.push_back(connection);
                    }
                }

                if (PQstatus(pgconn) == CONNECTION_BAD)
                {
                    closeFds.push_back(fd);
                }
            }

            if (ox_fdset_check(mfdset, fd, WriteCheck))
            {
                if (PQflush(pgconn) == 0)
                {
                    //移除可写检测                    ox_fdset_del(mfdset, fd, WriteCheck);
                }
            }
        }

        for (auto& v : closeFds)
        {
            removeConnection(v);
        }
    }

    void    trySendPendingQuery()
    {
        while (!mPendingQuery.empty() && !mIdleConnections.empty())
        {
            auto& query = mPendingQuery.front();
            auto& connection = mIdleConnections.front();

            if (PQsendQuery(connection->pgconn, query.request.c_str()) == 0)
            {
                cout << PQerrorMessage(connection->pgconn) << endl;
                if (query.callback != nullptr)
                {
                    query.callback(nullptr);
                }
            }
            else
            {
                ox_fdset_add(mfdset, PQsocket(connection->pgconn), WriteCheck);
                connection->callback = query.callback;
            }

            mPendingQuery.pop();
            mIdleConnections.pop_front();
        }
    }

    size_t  pendingQueryNum() const
    {
        return mPendingQuery.size();
    }

    size_t  getWorkingQuery() const
    {
        return mConnections.size() - mIdleConnections.size();
    }

    void    createConnection(  constchar *pghost, constchar *pgport,
                        constchar *pgoptions, constchar *pgtty,
                        constchar *dbName, constchar *login, constchar *pwd,
                        int num)
    {
        for (int i = 0; i < num; i++)
        {
            auto pgconn = PQsetdbLogin(pghost, pgport, pgoptions, pgtty, dbName, login, pwd);
            if (PQstatus(pgconn) == CONNECTION_OK)
            {
                auto connection = std::make_shared<Connection>(pgconn, nullptr);
                mConnections[PQsocket(pgconn)] = connection;
                PQsetnonblocking(pgconn, 1);
                ox_fdset_add(mfdset, PQsocket(pgconn), ReadCheck);
                mIdleConnections.push_back(connection);
            }
            else
            {
                cout << PQerrorMessage(pgconn);
                PQfinish(pgconn);
                pgconn = nullptr;
            }
        }

        if (!mConnections.empty())
        {
            sCreateTable((*mConnections.begin()).second->pgconn, mKVTableName, mHashTableName);
        }
    }

private:
    void    removeConnection(int fd)
    {
        auto it = mConnections.find(fd);
        if (it != mConnections.end())
        {
            auto connection = (*it).second;
            for (auto it = mIdleConnections.begin(); it != mIdleConnections.end(); ++it)
            {
                if ((*it)->pgconn == connection->pgconn)
                {
                    mIdleConnections.erase(it);
                    break;
                }
            }

            ox_fdset_del(mfdset, fd, ReadCheck | WriteCheck);
            PQfinish(connection->pgconn);
            mConnections.erase(fd);
        }
    }

private:
    staticvoid    sCreateTable(PGconn* conn, conststring& kvTableName, conststring& hashTableName)
    {
        {
            string query = "CREATE TABLE public.";
            query += kvTableName;
            query += "(key character varying NOT NULL, value json, CONSTRAINT key PRIMARY KEY(key))";
            PGresult* exeResult = PQexec(conn, query.c_str());
            auto status = PQresultStatus(exeResult);
            auto errorStr = PQresultErrorMessage(exeResult);
            PQclear(exeResult);
        }

        {
            string query = "CREATE TABLE public.";
            query += hashTableName;
            query += "(hashname character varying, key character varying, value json, ""CONSTRAINT hk PRIMARY KEY (hashname, key))";
            PGresult* exeResult = PQexec(conn, query.c_str());
            auto status = PQresultStatus(exeResult);
            auto errorStr = PQresultErrorMessage(exeResult);
            PQclear(exeResult);
        }
    }

private:
    struct QueryAndCallback
    {
        std::string request;
        RESULT_CALLBACK  callback;
    };

    struct Connection
    {
        PGconn* pgconn;
        RESULT_CALLBACK callback;

        Connection(PGconn* p, RESULT_CALLBACK c)
        {
            pgconn = p;
            callback = c;
        }
    };

    conststring                                    mKVTableName;
    conststring                                    mHashTableName;

    stringstream                                    mStringStream;
    fdset_s*                                        mfdset;

    std::unordered_map<int, shared_ptr<Connection>> mConnections;
    std::list<shared_ptr<Connection>>               mIdleConnections;

    std::queue<QueryAndCallback>                    mPendingQuery;

    /*TODO::监听wakeup支持*//*TODO::考虑固定分配connection给某业务*//*TODO::编写储存过程,替换现有的hashtable模拟方式,如循环使用jsonb_set以及 select value->k1, value->k2 from ...*//*TODO::编写储存过程,实现list*/
};

int main()
{
    using std::chrono::system_clock;

    AsyncPGClient asyncClient;
    asyncClient.createConnection("192.168.12.1", "5432", nullptr, nullptr, "postgres", "postgres", "19870323", 8);
    system_clock::time_point startTime = system_clock::now();

    auto nowTime = time(NULL);
    
    for (int i = 0; i < 100000; i++)
    {
        if(false)
        {
            string test = "INSERT INTO public.kv_data(key, value) VALUES (‘";
            test += std::to_string(nowTime*1000+i);
            test += "‘, ‘{\"hp\":100000}‘) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value;";

            asyncClient.postQuery(test);
        }
        else
        {
            asyncClient.postQuery("select * from public.kv_data where key=‘dd‘;");
        }
    }

    asyncClient.postQuery("INSERT INTO public.kv_data(key, value) VALUES (‘dodo5‘, ‘{\"hp\":100000}‘) "" ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value", [](const PGresult* result){
        cout << "fuck" << endl;
    });

    asyncClient.get("dd", [](conststring& value){
        cout << "get dd : " << value << endl;
    });

    asyncClient.set("dd", "{\"hp\":456}", [](bool isOK){
        cout << "set dd : " << isOK << endl;
    });

    asyncClient.hget("heros:dodo", "hp", [](conststring& value){
        cout << "hget heros:dodo:" << value << endl;
    });

    asyncClient.hset("heros:dodo", "hp", "{\"hp\":1}", [](bool isOK){
        cout << "hset heros:dodo:" << isOK << endl;
    });

    asyncClient.hmget("heros:dodo", { "hp", "money" }, [](const unordered_map<string, string>& kvs){
        cout << "hmget:" << endl;
        for (auto& kv : kvs)
        {
            cout << kv.first << " : " << kv.second << endl;
        }
    });

    asyncClient.hgetall("heros:dodo", [](const unordered_map<string, string>& kvs){
        cout << "hgetall:" << endl;
        for (auto& kv : kvs)
        {
            cout << kv.first << " : " << kv.second << endl;
        }
    });

    while (true)
    {
        asyncClient.poll(1);
        asyncClient.trySendPendingQuery();
        if (asyncClient.pendingQueryNum() == 0 && asyncClient.getWorkingQuery() == 0)
        {
            break;
        }
    }

    auto elapsed = system_clock::now() - startTime;
    cout << "cost :" << chrono::duration<double>(elapsed).count() << "s" << endl;
    cout << "enter any key exit" << endl;
    cin.get();
    return0;
}

代码地址：https://github.com/IronsDu/accumulation-dev/blob/master/examples/Pgedis.cpp

原文：http://www.cnblogs.com/irons/p/5369796.html

内容总结

以上是互联网集市为您收集整理的PostgreSQL异步客户端（并模拟redis 数据结构）全部内容，希望文章能够帮你解决PostgreSQL异步客户端（并模拟redis 数据结构）所遇到的程序开发问题。如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容，请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

扫描二维码推送至手机访问。

本文链接：https://qyyshop.com/info/1174366.html

来源：【匿名】

【上一篇】Redis集群的安装测试（伪分布模式 - 主从复制）【下一篇】redis可以做什么？

更多 ►

【PostgreSQL异步客户端（并模拟redis 数据结构）】教程文章相关的互联网学习教程文章

Redis学习笔记（五）——数据结构之哈希（Hash）【图】

一、介绍　　Redis hash是一个string类型的field和value的映射表，hash特别设于用于存储对象。　　Redis中每个hash可以存储232 - 1 键值对（40多亿）。　　基本命令：添加/更新：　　hset key field value （将hash表key中的字段field的值设为value）　　hsetnv key field value （只有在字段field不存在时，设置hash表字段的值）　　hmset key field1 value1 [field2 value2 ...] （同时将多个fiedl-value(域-值)对设置到hash...

Redis系列（六）：数据结构List双向链表LPUSH、LPOP、RPUSH、RPOP、LLEN命令【代码】【图】

1.介绍redis中的list既实现了栈（先进后出）又实现了队列（先进先出）1.示意图 2.各命令详解LPUSH/RPUSHLPUSH:从队列的左边入队一个或多个元素将所有指定的值插入到存于 key 的列表的头部。如果 key 不存在，那么在进行 push 操作前会创建一个空列表。如果 key 对应的值不是一个 list 的话，那么会返回一个错误。可以使用一个命令把多个元素 push 进入列表，只需在命令末尾加上多个指定的参数。元素是从最左端的到最右端的、一个...

redis 源码学习（核心数据结构剖析）【图】

redis是个key, value数据库，是个内存数据库。目前是个互联网公司的架构标配。支持的数据对象有string, list, set, zest和hash object。数据结构：数据库的核心结构是dict（实现是使用hashmap）：key: stringvalue: string或者list或者set或者zest或者hash object。dict数据结构定义：typedef struct dictht {// 哈希表数组dictEntry **table;// 哈希表大小unsigned long size;// 哈希表大小掩码，用于计算索引值// 总是等于 size ...

Redis系列 - 数据结构【图】

前言　　转载自微信公众号：三太子敖丙1、问：Redis有那些数据结构？　　常见的5种：字符串（String），散列（Hash），列表（List），集合（Set），有序集合（SortedSet）等；除此之外，还有HyperLogLog、Geo、Pub/Sub。　　注：之前项目组也有遇见过 BloomFilter（布隆过滤器），这个能很好地防止缓存穿透的发生，它的原理也很简单，就是利用高效的数据结构和算法快速判断出你这个 Key 是否在数据库中存在，不存在你 Return 就好...

PostgreSQL异步客户端（并模拟redis 数据结构）【代码】

以前为了不在游戏逻辑（对象属性）变更时修改数据库，就弄了个varchar字段来表示json，由服务器逻辑（读取到内存）去操作它。但这对运维相当不友好，也不能做一些此Json数据里查询。所以后面就用了下ssdb，然而就在前几天才了解到postgresql支持json了（其实早在两年前就行了吧···）就这点差不多就可以算当作mongodb用了，不过还是不支持redis的高级数据结构。于是我就想模拟（实现）下redis（的数据结构）。就抽空看了下它的c a...

redis基本数据结构-字符串

reids字符串数据结构相关命令序号命令命令实例意义 set key valueset bar 1设置key为bar的值为"1" incr keyincr bar键bar的值加1，此时bar为"2" get keyget bar获取键为bar的值，为"2" incrby key increment　incr bar 10　键bar的值加10，此时bar为"12" decr keydecr bar键bar的值减1，此时bar为"11" decrby key decrementdecrby bar 5键bar的值减去5，此时bar为"6"可以减到负数 incrbyfloat key incrementincrbyfloat...

Redis数据结构-字典【代码】

1. 简介字典在Redis中应用广泛，Redis数据库的底层就是使用字典来实现的，对数据库的增删查改也都是建立在字典的操作之上的。此外，字典还是哈希键的实现之一。2. 实现2.1 哈希表哈希表由dict.h/dictht 结构定义：typedef struct dictht {// 哈希表数组dictEntry **table;//哈希表大小unsigned long size;//哈希表大小掩码，用于计算索引值，总是等于size-1unsigned long sizemask;//该哈希表已有节点数量unsigned long used; } di...

Redis 数据结构

Redis 有 String，List，Set，Sorted Set，Hash 共 5 种数据类型String：键 => 字符串 key => string用法示例：set key stringList：键 => 双向链表key => [0 => string1 <-> 1 => string2 <-> 2 => string3 <-> ... ]用法示例：lpush key string1lpush key string2lset key index string1Set：键 => 字符串无序集合key => [string1, string2, string3, ... ]用法示例：sadd key string1sadd key string2Sorted Set：键 => 按数值（...

Redis数据结构之链表【图】

Redis使用的链表是双向无环链表，链表节点可用于保存各种不同类型的值。一、链表结构定义1. 链表节点结构定义：2. 链表结构定义：示例：二、链表在Redis中的用途1. 作为列表键的底层实现之一：当一个列表键包含了数量比较多的元素，又或者列表中包含的元素都是比较长的字符串时，Redis就会使用链表作为列表键的底层实现。2. 除此之外，发布与订阅、慢查询、监视器等功能也用到了链表，Redis服务器本身还使用链表来保存多个客户端...

2017-4-20/Redis的数据结构及应用场景

1. 谈谈你对redis的理解，它的应用场景。Redis是一个key-value存储系统，它支持存储的value类型包括string字符串、list链表、set集合、sorted Set有序集合和hash哈希等数据类型。这些数据类型都支持push/pop、add/remove及取交集并集和差集及更丰富的操作，而且这些操作都是原子性的，支持各种不同方式的排序。为了保证效率，Redis将数据都缓存在内存中，并周期性的把更新的数据写入磁盘或者把修改操作写入追加的记录文件，在此基础...

Redis数据结构：跳跃表【代码】

1. 跳跃表是有序集合（zset）的底层实现之一；2. 由zskiplist和zskiplistNode组成；3. 每个跳跃表节点的层数都是1-32之间的随机数（每创建一个节点的时候，程序会随机生成一个数[1-32]作为level数组的大小）；4. 同一个跳跃表中，多个节点可以包含相同的分值，但节点的成员对象是唯一的；5. 按分值排序，若分值相同就按成员对象大小排序。zskiplist数据结构typedef struct zskiplist {//头节点和尾节点struct zskiplistNode *heade...

Redis学习笔记之Redis基本数据结构【代码】【图】

Redis基础数据结构Redis有5种基本数据结构：String(字符串)、list(列表)、set(集合)、hash(哈希)、zset(有序集合)字符串string字符串类型是Redis的value最简单的数据结构，类似与Java语言中的ArrayList(数字列表)，不过在Redis里String是一种动态字符串Redis里的String采用预分配冗余空间的方法set & get>set keyname test OK>get keyname test//key如果存在就返回0 >setnx keyname test 0>exists keyname>del keyname 1//批量设置...

首页 / REDIS / PostgreSQL异步客户端（并模拟redis 数据结构）

PostgreSQL异步客户端（并模拟redis 数据结构）

内容导读

内容图文

内容总结

内容备注

内容手机端

【PostgreSQL异步客户端（并模拟redis 数据结构）】教程文章相关的互联网学习教程文章

Redis学习笔记（五）——数据结构之哈希（Hash）【图】

Redis系列（六）：数据结构List双向链表LPUSH、LPOP、RPUSH、RPOP、LLEN命令【代码】【图】

redis 源码学习（核心数据结构剖析）【图】

Redis系列 - 数据结构【图】

PostgreSQL异步客户端（并模拟redis 数据结构）【代码】

redis基本数据结构-字符串

Redis数据结构-字典【代码】

Redis 数据结构

Redis数据结构之链表【图】

2017-4-20/Redis的数据结构及应用场景

Redis数据结构：跳跃表【代码】

Redis学习笔记之Redis基本数据结构【代码】【图】

redis数据结构、持久化、缓存淘汰策略【代码】【图】

什么是redis,为什么要用redis来做缓存？redis 常见数据结构？

.Net Core使用分布式缓存Redis：数据结构【代码】

REDIS - 相关标签

数据结构 - 相关标签

POST - 相关标签

REDIS - 技术教程分类

REDIS - 最新教程

REDIS - 最热教程