C#使用SSDB管理增量日志并提供查询

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› c#›C#教程

原作者: [db:作者] 来自: [db:来源] 收藏邀请

Program.cs

using System;
using System.Text;
using CommonLinkLibrary.Util;
using Newtonsoft.Json;

namespace SSDB
{
    class Program
    {
        static void Main(string[] args)
        {
            InitSsdb();

            var channelName = "PersonUpdateLog";
            var channelNameKeys = channelName + "_Keys";
            var ssdb = ConnectionPool.GetSocket();
            ssdb.zclear(channelName);
            ssdb.del(channelNameKeys);
            Console.WriteLine("清空"+channelName+"成功！");

            for (var i = 1; i <= 100; i++)
            {
                var pk = GetPrimaryKey(ssdb, channelNameKeys);
                var myBean = new PersonInfoLogBean
                {
                    Action = "新增",
                    NewPersonName = "黄海" + i,
                    OldPersonName = "黄海" + i,
                    PersonId = i.ToString(),
                    OldAge = i.ToString(),
                    NewAge = i.ToString()
                };
                var msg = JsonConvert.SerializeObject(myBean);
                ssdb.zset(Encoding.UTF8.GetBytes(channelName), Encoding.UTF8.GetBytes(msg), pk);
            }

            for (var i = 1; i <= 100; i++)
            {
                var pk = GetPrimaryKey(ssdb, channelNameKeys);
                var myBean = new PersonInfoLogBean
                {
                    Action = "修改",
                    NewPersonName = "(新)黄海" + i,
                    OldPersonName = "(旧)黄海" + i,
                    PersonId = i.ToString(),
                    OldAge = i.ToString(),
                    NewAge = i.ToString()
                };
                var msg = JsonConvert.SerializeObject(myBean);
                ssdb.zset(Encoding.UTF8.GetBytes(channelName), Encoding.UTF8.GetBytes(msg), pk);
            }

            for (var i = 1; i <= 100; i++)
            {
                var pk = GetPrimaryKey(ssdb, channelNameKeys);
                var myBean = new PersonInfoLogBean
                {
                    Action = "删除",
                    NewPersonName = "(新)黄海" + i,
                    OldPersonName = "(旧)黄海" + i,
                    PersonId = i.ToString(),
                    OldAge = i.ToString(),
                    NewAge = i.ToString()
                };
                var msg = JsonConvert.SerializeObject(myBean);
                ssdb.zset(Encoding.UTF8.GetBytes(channelName), Encoding.UTF8.GetBytes(msg), pk);
            }
            Console.WriteLine("保存成功！");

            const int size = 10;
            var startId = 0;
                while (true)
                {
                    var keys = ssdb.zscan(channelName, "", startId, 999999999999, size);
                    for (var i = 0; i < keys.Length; i++)
                    {
                        Console.WriteLine(keys[i].Key + "      " + keys[i].Value);
                    }
                    if (size != keys.Length)
                    {
                        break;
                    }
                    else
                    {
                        startId = startId + size;
                    }
                }
                ConnectionPool.PutSocket(ssdb);
                Console.ReadKey();
        }
        public static void InitSsdb()
        {
            //连接池的初始化工作
            var serverIp = "10.10.6.199";
            ConnectionPool.InitializeConnectionPool(serverIp, 8888, 20, 100);
        }
        
        public static long GetPrimaryKey(SsdbClient ssdb, string primaryKeyName)
        {
            return ssdb.incr(Encoding.UTF8.GetBytes(primaryKeyName), 1);
        }
    }
}

ConnectionPool.cs

using System;
using System.Collections.Generic;
using System.Threading;

namespace CommonLinkLibrary.Util
{
    public static class ConnectionPool
    {
        /// <summary>
        ///     Queue of available socket connections.
        /// </summary>
        private static Queue<SsdbClient> _availableSockets;

        /// <summary>
        ///     The maximum size of the connection pool.
        /// </summary>
        private static int _poolMaxSize = 20;

        private static string _hostIpAddress;
        private static int _hostPortNumber;

        /// <summary>
        ///     Created host Connection counter
        /// </summary>
        private static int _socketCounter;

        public static bool Initialized;

        /// <summary>
        ///     Initialize host Connection pool
        /// </summary>
        /// <param name="hostPortNumber"></param>
        /// <param name="minConnections">Initial number of connections</param>
        /// <param name="maxConnections">The maximum size of the connection pool</param>
        /// <param name="hostIpAddress"></param>
        public static void InitializeConnectionPool(string hostIpAddress, int hostPortNumber, int minConnections,
            int maxConnections)
        {
            _socketCounter = 0;
            _poolMaxSize = maxConnections;
            _hostIpAddress = hostIpAddress;
            _hostPortNumber = hostPortNumber;

            _availableSockets = new Queue<SsdbClient>();

            for (var i = 0; i < minConnections; i++)
            {
                var cachedSocket = OpenSocket(hostIpAddress, hostPortNumber);
                PutSocket(cachedSocket);
            }

            Initialized = true;
        }

        /// <summary>
        ///     Get an open socket from the connection pool.
        /// </summary>
        /// <returns>Socket returned from the pool or new socket opened. </returns>
        public static SsdbClient GetSocket()
        {
            //如果连接池中还有可用的连接，那么调用
            if (_availableSockets.Count > 0)
            {
                lock (_availableSockets)
                {
                    SsdbClient socket = null;
                    while (_availableSockets.Count > 0)
                    {
                        socket = _availableSockets.Dequeue();

                        if (socket.Connected)
                        {
                            return socket;
                        }
                        socket.Close();
                        Interlocked.Decrement(ref _socketCounter);
                    }
                }
            }
            //如果没有可用的连接，那么新打开一个连接
            return OpenSocket(_hostIpAddress, _hostPortNumber);
        }

        /// <summary>
        ///     Return the given socket back to the socket pool.
        /// </summary>
        /// <param name="socket">Socket connection to return.</param>
        public static void PutSocket(SsdbClient socket)
        {
            lock (_availableSockets)
            {
                if (_availableSockets.Count < _poolMaxSize) // Configuration Value
                {
                    if (socket != null)
                    {
                        if (socket.Connected)
                        {
                            _availableSockets.Enqueue(socket);
                        }
                        else
                        {
                            socket.Close();
                        }
                    }
                }
                else
                {
                    socket.Close();
                }
            }
        }

        /// <summary>
        ///     Open a new socket connection.
        /// </summary>
        /// <returns>Newly opened socket connection.</returns>
        private static SsdbClient OpenSocket(string hostIpAddress, int hostPortNumber)
        {
            if (_socketCounter < _poolMaxSize)
            {
                Interlocked.Increment(ref _socketCounter);
                var client = new SsdbClient(hostIpAddress, hostPortNumber);
                return client;
            }
            throw new Exception("Connection Pool reached its limit");
        }
    }
}

Link.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Net.Sockets;
using System.Text;

namespace CommonLinkLibrary.Util
{
    internal class Link : IDisposable
    {
        private MemoryStream _recvBuf = new MemoryStream(8*1024);
        private TcpClient _sock;
        //供程序员显式调用的Dispose方法
        public void Dispose()
        {
            _recvBuf.Dispose();
            close();
        }

        //protected的Dispose方法，保证不会被外部调用。
        //传入bool值disposing以确定是否释放托管资源
      

        //供GC调用的析构函数
        ~Link()
        {
            close();
        }
        public Link(string host, int port)
        {
            _sock = new TcpClient(host, port) {NoDelay = true};
            _sock.Client.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, true);
        }
        
        public void close()
        {
            if (_sock != null)
            {
                _sock.Close();
            }
            _sock = null;
        }

        public List<byte[]> request(string cmd, params string[] args)
        {
            var req = new List<byte[]>(1 + args.Length) {Encoding.UTF8.GetBytes(cmd)};
            foreach (var s in args)
            {
                req.Add(Encoding.UTF8.GetBytes(s));
            }
            return request(req);
        }

        public List<byte[]> request(string cmd, params byte[][] args)
        {
            var req = new List<byte[]>(1 + args.Length) {Encoding.UTF8.GetBytes(cmd)};
            req.AddRange(args);
            return request(req);
        }

        public List<byte[]> request(List<byte[]> req)
        {
            var buf = new MemoryStream();
            foreach (var p in req)
            {
                if(p!=null)
                {
                    var len = Encoding.UTF8.GetBytes(p.Length.ToString());
                    buf.Write(len, 0, len.Length);
                    buf.WriteByte((byte)'\n');
                    buf.Write(p, 0, p.Length);
                    buf.WriteByte((byte)'\n');
                }                
            }
            buf.WriteByte((byte) '\n');

            var bs = buf.GetBuffer();
            _sock.GetStream().Write(bs, 0, (int) buf.Length);
            //Console.Write(Encoding.UTF8.GetString(bs, 0, (int)buf.Length));
            return recv();
        }

        private List<byte[]> recv()
        {
            while (true)
            {
                var ret = parse();
                if (ret != null)
                {
                    return ret;
                }
                var bs = new byte[8192];
                var len = _sock.GetStream().Read(bs, 0, bs.Length);
                //Console.WriteLine("<< " + Encoding.UTF8.GetString(bs));
                _recvBuf.Write(bs, 0, len);
            }
        }

        private static int memchr(byte[] bs, byte b, int offset)
        {
            for (var i = offset; i < bs.Length; i++)
            {
                if (bs[i] == b)
                {
                    return i;
                }
            }
            return -1;
        }

        private List<byte[]> parse()
        {
            var list = new List<byte[]>();
            var buf = _recvBuf.GetBuffer();

            var idx = 0;
            while (true)
            {
                var pos = memchr(buf, (byte) '\n', idx);
                //System.out.println("pos: " + pos + " idx: " + idx);
                if (pos == -1)
                {
                    break;
                }
                if (pos == idx || (pos == idx + 1 && buf[idx] == '\r'))
                {
                    idx += 1; // if '\r', next time will skip '\n'
                    // ignore empty leading lines
                    if (list.Count == 0)
                    {
                        continue;
                    }
                    var left = (int) _recvBuf.Length - idx;
                    _recvBuf = new MemoryStream(8192);
                    if (left > 0)
                    {
                        _recvBuf.Write(buf, idx, left);
                    }
                    return list;
                }
                var lens = new byte[pos - idx];
                Array.Copy(buf, idx, lens, 0, lens.Length);
                var len = int.Parse(Encoding.UTF8.GetString(lens));

                idx = pos + 1;
                if (idx + len >= _recvBuf.Length)
                {
                    break;
                }
                var data = new byte[len];
                Array.Copy(buf, idx, data, 0, data.Length);

                //Console.WriteLine("len: " + len + " data: " + Encoding.UTF8.GetString(data));
                idx += len + 1; // skip '\n'
                list.Add(data);
            }
            return null;
        }
    }
}

PersonInfoLogBean.cs

namespace SSDB
{
    class PersonInfoLogBean
    {
        public string Action { get; set; }
        public string PersonId { get; set; }
        public string OldPersonName { get; set; }
        public string NewPersonName { get; set; }

        public string OldAge { get; set; }
        public string NewAge { get; set; }

    }
}

SsdbClient.cs

using System;
using System.Collections.Generic;
using System.Net.Sockets;
using System.Text;

namespace CommonLinkLibrary.Util
{
    public class SsdbClient : TcpClient
    {
        private readonly Link _link;
        private string _respCode;

        public SsdbClient(string hostIpAddress, int hostPortNumber) : base(hostIpAddress, hostPortNumber)
        {
            _link = new Link(hostIpAddress, hostPortNumber);
        }

        public List<byte[]> request(string cmd, params string[] args)
        {
            return _link.request(cmd, args);
        }

        public List<byte[]> request(string cmd, params byte[][] args)
        {
            return _link.request(cmd, args);
        }

        public List<byte[]> request(List<byte[]> req)
        {
            return _link.request(req);
        }

        private byte[] _bytes(string s)
        {
            if (s == null) return null;
            return Encoding.UTF8.GetBytes(s);
        }

        private string _string(byte[] bs)
        {
            return Encoding.UTF8.GetString(bs);
        }

        private KeyValuePair<string, byte[]>[] parse_scan_resp(List<byte[]> resp)
        {
            _respCode = _string(resp[0]);

            var size = (resp.Count - 1)/2;
            var kvs = new KeyValuePair<string, byte[]>[size];
            for (var i = 0; i < size; i += 1)
            {
                var key = _string(resp[i*2 + 1]);
                var val = resp[i*2 + 2];
                kvs[i] = new KeyValuePair<string, byte[]>(key, val);
            }
            return kvs;
        }

        /***** kv *****/

        public bool exists(byte[] key)
        {
            var resp = request("exists", key);
            _respCode = _string(resp[0]);
            if (_respCode == "not_found")
            {
                return false;
            }
            if (resp.Count != 2)
            {
                throw new Exception("Bad response!");
            }
            return (_string(resp[1]) == "1" ? true : false);
        }

        public bool exists(string key)
        {
            return exists(_bytes(key));
        }

        public void set(byte[] key, byte[] val)
        {
            var resp = request("set", key, val);
            _respCode = _string(resp[0]);
        }

        public void set(string key, string val)
        {
            set(_bytes(key), _bytes(val));
        }

        /// <summary>
        /// </summary>
        /// <param name="key"></param>
        /// <param name="val"></param>
        /// <returns>returns true if name.key is found, otherwise returns false.</returns>
        public bool get(byte[] key, out byte[] val)
        {
            val = null;
            var resp = request("get", key);
            _respCode = _string(resp[0]);
            if (_respCode == "not_found")
            {
                return false;
            }
            if (resp.Count != 2)
            {
                throw new Exception("Bad response!");
            }
            val = resp[1];
            return true;
        }

        public bool get(string key, out byte[] val)
        {
            return get(_bytes(key), out val);
        }

        public bool get(string key, out string val)
        {
            val = null;
            byte[] bs;
            if (!get(key, out bs))
            {
                return false;
            }
            val = _string(bs);
            return true;
        }

        public void del(byte[] key)
        {
            var resp = request("del", key);
            _respCode = _string(resp[0]);
        }

        public void del(string key)
        {
            del(_bytes(key));
        }

        public KeyValuePair<string, byte[]>[] scan(string key_start, string key_end, long limit)
        {
            var resp = request("scan", key_start, key_end, limit.ToString());
            return parse_scan_resp(resp);
        }

        public KeyValuePair<string, byte[]>[] rscan(string key_start, string key_end, long limit)
        {
            var resp = request("rscan", key_start, key_end, limit.ToString());
            return parse_scan_resp(resp);
        }

        /***** hash *****/

        public void hset(byte[] name, byte[] key, byte[] val)
        {
            var resp = request("hset", name, key, val);
            _respCode = _string(resp[0]);
        }

        public void hset(string name, string key, byte[] val)
        {
            hset(_bytes(name), _bytes(key), val);
        }

        public void hset(string name, string key, string val)
        {
            hset(_bytes(name), _bytes(key), _bytes(val));
        }

        /// <summary>
        /// </summary>
        /// <param name="name"></param>
        /// <param name="key"></param>
        /// <param name="val"></param>
        /// <returns>returns true if name.key is found, otherwise returns false.</returns>
        public bool hget(byte[] name, byte[] key, out byte[] val)
        {
            val = null 
                       
                    
                     
                      



鲜花




握手




雷人




路过




鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

使用c#特性，给方法或类打自定义标签再反射获取发布时间：2022-07-10

【C#】ConcurrentBagT方法发布时间：2022-07-10

剪的笔顺,诠释剪的笔画,认识剪的部首

六六分期app的软件客服如何联系？(六六分期

2023-10-27

florent37/ViewAnimator: A fluent Android

2022-08-15

florent37/Shrine-MaterialDesign2: implem

2022-08-17

CVE-2020-36276

2022-09-23

SimpleSoftwareIO/simple-sms: Send and re

2022-08-13

阅读排行榜

1 六六分期app的软件客服如何联系？(六六分期

六六分期app的软件客服如何联系？不知道吗？加qq群【895510560】即可！标题：六六分期

阅读：19235|2023-10-27

2 可心卡盟:win10系统火狐flash插件崩溃怎么

今天小编告诉大家如何处理win10系统火狐flash插件总是崩溃的问题，可能很多用户都不知

阅读：10000|2022-11-06

3 亲亲特价:怎么删除回收站图标

今天小编告诉大家如何对win10系统删除桌面回收站图标进行设置，可能很多用户都不知道

阅读：8331|2022-11-06

4 济南大学虚拟社区:鲁大师节能降温的具体办

今天小编告诉大家如何对win10系统电脑设置节能降温的设置方法，想必大家都遇到过需要

阅读：8701|2022-11-06

5 xlueops.exe:无线网络安装向导

我们在使用xp系统的过程中,经常需要对xp系统无线网络安装向导设置进行设置，可能很多

阅读：8646|2022-11-06

6 女斗合众国:win7系统cf与主机连接不稳定怎

今天小编告诉大家如何处理win7系统玩cf老是与主机连接不稳定的问题，可能很多用户都不

阅读：9668|2022-11-06

7 0xc000022-[cf烟雾头]cf怎么调烟雾头

电脑对日常生活的重要性小编就不多说了，可是一旦碰到win7系统设置cf烟雾头的问题，很

阅读：8632|2022-11-06

8 qizideyouhuo:应用程序无法正常启动0xc0000

我们在日常使用电脑的时候，有的小伙伴们可能在打开应用的时候会遇见提示应用程序无法

阅读：8005|2022-11-06

9 ipz-185:win7系统vcf文件怎么打开

今天小编告诉大家如何对win7系统打开vcf文件进行设置，可能很多用户都不知道怎么对win

阅读：8666|2022-11-06

10 傻哥蹦迪:win10系统s4怎么打开usb调试

今天小编告诉大家如何对win10系统s4开启USB调试模式进行设置，可能很多用户都不知道怎

阅读：7540|2022-11-06

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服（服务时间 9:00～18:00）

在线QQ客服

地址：深圳市南山区西丽大学城创智工业园

电邮：jeky_zhao#qq.com

移动电话：139-2527-9053

客服电话

电子邮件

C#使用SSDB管理增量日志并提供查询

请发表评论

全部评论

上一篇：

下一篇：

台山市属于哪个市

kojino/Harvard-Robust-Machine-Learning:

文的笔顺,诠释文的笔画,解读文的部首

MatthewPeterKelly/IntroMatlabDynamics: A

zbai/MATLAB-Groves: Principles of GNSS,