使用scyllaDb 或者cassandra存储聊天记录

一、使用scyllaDb的原因

目前开源的聊天软件主要还是使用mysql存储数据，数据量大的时候比较麻烦；

我打算使用scyllaDB存储用户的聊天记录，主要考虑的优点是：

1）方便后期线性扩展服务器；

2）partition更方便，clustering 可以将一组数据放在一起，加载更快；

我的后端服务使用go来写，

使用的库为https://github.com/scylladb/gocqlx/，目前版本为2.8

go get -u github.com/scylladb/gocqlx/v2

二、测试代码

1. 连接数据库

cluster := gocql.NewCluster("127.0.0.1:9042")cluster.Keyspace = "chatdata"cluster.Authenticator = gocql.PasswordAuthenticator{Username: "cassandra",Password: "cassandra",}session, err := cluster.CreateSession()if err != nil {fmt.Println("创建会话时发生错误:", err)return}defer session.Close()sessionx, err := gocqlx.WrapSession(session, nil)if err != nil {}defer sessionx.Close()

我是测试的机器，只有一个节点，后续在数据一致性要求也都写一个节点；

2. 定义数据结构

P2P的聊天，使用如下表：

CREATE TABLE pchat (pk int,        // 分区uid1 bigint,   // 用户自己，P2P时写扩散，每个用户存储一份数据uid2 bigint,   // 对方id bigint,     // 消息全局唯一ID，服务器分配usid bigint,   // 发送方的消息唯一标记tm timestamp,   // 时间戳tm1 timestamp,  // 接收tm2 timestamp,  // 已读draf text,      // 数据io boolean,     // 收，发del boolean,    // 删除标记t smallint,     // 消息类型PRIMARY KEY (pk, uid1, tm, id))

在 Cassandra 中，PRIMARY KEY 的定义影响了数据如何进行分区（Partitioning）和在分区内如何进行排序（Clustering）。对于表定义 PRIMARY KEY (pk, uid1, tm, id)，它的影响如下：

  1. 分区键 (pk): 数据将按照 pk 的值进行分区。相同 pk 的数据会被存储在同一分区中。

  2. 聚簇键 (uid1,tm, id): 在同一分区内，数据将按照 (uid1, tm, id) 进行排序。这意味着相同 pk 的分区内的数据将按照 uid1 的值进行子分区，然后在每个子分区内按照 tm, id 的值进行排序。

简单来说，数据会先按照 pk 进行分区，然后在每个分区内，按照 (uid1, tm, id) 进行排序。这样的设计允许你在查询时方便地按照 pk、uid1 和tm,  id 进行范围查询。

  • 一对一的聊天，都是2个用户，使用写扩散方式每个用户1份数据，这样的的好处是，使用用户ID聚簇，可以提高加载速度。并且减少数据的加载次数，具体在用户的会话区分上，可以在客户端一侧，执行本地的SQLITE存储。
  • 对比tinode的策略，它是按照每个会话做一个逻辑，需要管理当前所有的会话，逐个加载或者订阅，而且在测试过程中发现BUG，当如同微信一样删除了某个会话，等于拉了黑名单，无法后续会话了，这个不符合我们的习惯。
  • 对于群组聊天，可以使用读扩散的方式，因为写扩散毕竟太占用系统资源了；按照组ID来聚簇；

相关代码如下：

// 定义表的元数据
var pchatMetadata = table.Metadata{Name:    "pchat",Columns: []string{"pk", "uid1", "uid2", "id", "usid", "tm", "tm1", "tm2", "draf", "io", "del", "t"},PartKey: []string{"pk"},SortKey: []string{"uid1", "id"},
}// 创建表对象
var pchatTable = table.New(pchatMetadata)// 定义数据结构
type PchatData struct {Pk   int       `db:"pk"`Uid1 int       `db:"uid1"`Uid2 int       `db:"uid2"`Id   int       `db:"id"`Usid int       `db:"usid"`Tm   time.Time `db:"tm"`Tm1  time.Time `db:"tm1"`Tm2  time.Time `db:"tm2"`Draf string    `db:"draf"`Io   bool      `db:"io"`Del  bool      `db:"del"`T    int       `db:"t"`
}func PchatDataToSlice(data PchatData) []interface{} {return []interface{}{data.Pk,data.Uid1,data.Uid2,data.Id,data.Usid,data.Tm,data.Tm1,data.Tm2,data.Draf,data.Io,data.Del,data.T,}
}

3. 单条数据写入

func insertData(session *gocqlx.Session) error {data := PchatData{Pk:   1,Uid1: 123456,Uid2: 789012,Id:   987654,Usid: 654321,Tm:   time.Now(),Tm1:  time.UnixMilli(0),Tm2:  time.UnixMilli(0),Draf: "你的草稿内容",Io:   true,Del:  false,T:    42,}// Insert using query builder.insertChat := qb.Insert("chatdata.pchat").Columns(pchatMetadata.Columns...).Query(*session).Consistency(gocql.One)insertChat.BindStruct(data)if err := insertChat.ExecRelease(); err != nil {fmt.Println(err)return err}return nil
}

4. 批量插入

func insertBatch(session *gocqlx.Session) error {// 创建 Batchbatch := session.Session.NewBatch(gocql.LoggedBatch)// 创建 Batch//batch := gocql.NewBatch(gocql.LoggedBatch)batch.Cons = gocql.LocalOneindex := 1// 构建多个插入语句for i := index; i < index+1000; i++ {data := PchatData{Pk:   1,Uid1: 1001,Uid2: 1005,Id:   i,Usid: i,Tm:   time.Now(),Tm1:  time.UnixMilli(0),Tm2:  time.UnixMilli(0),Draf: "你的草稿内容",Io:   true,Del:  false,T:    1,}insertChatQry := qb.Insert("chatdata.pchat").Columns(pchatMetadata.Columns...).Query(*session).Consistency(gocql.One)batch.Query(insertChatQry.Statement(),PchatDataToSlice(data)...)}if err := session.ExecuteBatch(batch); err != nil {return err}return nil
}

挺快的，我远程插入云主机，1000条数据，使用了50毫秒左右；

5.  查询所有

这里就是一个测试，真正使用中，不会这么用

func queryData(session *gocqlx.Session) error {var dataList []PchatDataq := qb.Select("chatdata.pchat").Columns(pchatMetadata.Columns...).Query(*session).Consistency(gocql.One)if err := q.Select(&dataList); err != nil {return err}//for _, c := range dataList {//	fmt.Printf("%+v \n", c)//}for _, d := range dataList {fmt.Printf("pk: %d, uid1: %d, uid2: %d, id: %d, usid: %d, tm: %v, tm1: %v, tm2: %v, draf: %s, io: %t, del: %t, t: %d\n",d.Pk, d.Uid1, d.Uid2, d.Id, d.Usid, d.Tm, d.Tm1, d.Tm2, d.Draf, d.Io, d.Del, d.T)}return nil
}

6. 游标与分页

库内部提供了一些分页机制，但是我总觉得似乎不是我想要的；测试发现比较慢，目前没深入去研究内部机制：

func queryDataByPage(session *gocqlx.Session) error {var pageSize = 10//chatTable := table.New(pchatMetadata)builder := qb.Select("chatdata.pchat").Columns(pchatMetadata.Columns...)builder.Where(qb.Eq("uid1"))builder.AllowFiltering()q := builder.Query(*session)defer q.Release()q.PageSize(pageSize)q.Consistency(gocql.One)q.Bind(1001)getUserChatFunc := func(userID int64, page []byte) (chats []PchatData, nextPage []byte, err error) {if len(page) > 0 {q.PageState(page)}iter := q.Iter()return chats, iter.PageState(), iter.Select(&chats)}var (dataList []PchatDatanextPage []byteerr      error)for i := 1; ; i++ {dataList, nextPage, err = getUserChatFunc(1001, nextPage)if err != nil {fmt.Println(err)return err}fmt.Printf("Page %d: \n", i)for _, d := range dataList {//fmt.Printf("pk: %d, uid1: %d, uid2: %d, id: %d, usid: %d, tm: %v, tm1: %v, tm2: %v, draf: %s, io: %t, del: %t, t: %d\n",//	d.Pk, d.Uid1, d.Uid2, d.Id, d.Usid, d.Tm, d.Tm1, d.Tm2, d.Draf, d.Io, d.Del, d.T)fmt.Printf("pk: %d, uid1: %d, uid2: %d, id: %d \n", d.Pk, d.Uid1, d.Uid2, d.Id)}if len(nextPage) == 0 {break}}return nil
}

7. 按用户与id号来加载

我设想的用法是，既然按照user id 聚簇了，支持多个客户端使用时，某个客户端初次加载（冷加载），可以加载最近的部分，然后根据需要在根据条件加载；持续更新的用户（热加载）首先是考虑从redis中加载，已经落库的部分再根据时间段加载；

这里测试的是，从某个ID=900的条目之后，加载10条

func queryDataByIdPage(session *gocqlx.Session) error {var pageSize uint = 10//chatTable := table.New(pchatMetadata)builder := qb.Select("chatdata.pchat").Columns(pchatMetadata.Columns...)builder.Where(qb.Eq("uid1"), qb.Gt("id"))builder.AllowFiltering()builder.Limit(pageSize)q := builder.Query(*session)defer q.Release()q.Consistency(gocql.One)q.Bind(1002, 900)var dataList []PchatDataerr := q.Select(&dataList)if err != nil {fmt.Println(err)return err}fmt.Printf("size= %d: \n", len(dataList))for _, d := range dataList {//fmt.Printf("pk: %d, uid1: %d, uid2: %d, id: %d, usid: %d, tm: %v, tm1: %v, tm2: %v, draf: %s, io: %t, del: %t, t: %d\n",//	d.Pk, d.Uid1, d.Uid2, d.Id, d.Usid, d.Tm, d.Tm1, d.Tm2, d.Draf, d.Io, d.Del, d.T)fmt.Printf("pk: %d, uid1: %d, uid2: %d, id: %d tm: %v \n", d.Pk, d.Uid1, d.Uid2, d.Id, d.Tm)}return nil
}

8. 按照时间范围来找

func string2timeLoc(dateString string) (time.Time, error) {// 设置东八区（中国标准时间）的地理位置loc, err := time.LoadLocation("Asia/Shanghai")if err != nil {fmt.Println("加载地理位置错误:", err)return time.Now(), err}// 使用地理位置信息进行日期解析parsedTime, err := time.ParseInLocation("2006-01-02 15:04:05", dateString, loc)if err != nil {fmt.Println("日期解析错误:", err)return time.Now(), err}return parsedTime, nil
}
func queryDataBytmPage(session *gocqlx.Session) error {//var pageSize uint = 10//chatTable := table.New(pchatMetadata)builder := qb.Select("chatdata.pchat").Columns(pchatMetadata.Columns...)builder.Where(qb.Eq("uid1"), qb.GtOrEq("tm"), qb.LtOrEq("tm"))builder.AllowFiltering()//builder.Limit(pageSize)q := builder.Query(*session)defer q.Release()q.Consistency(gocql.One)tm1, _ := string2timeLoc("2024-01-27 13:24:00")tm2, _ := string2timeLoc("2024-01-27 13:25:56")q.Bind(1001, tm1, tm2)var dataList []PchatDataerr := q.Select(&dataList)if err != nil {fmt.Println(err)return err}fmt.Printf("size= %d: \n", len(dataList))for _, d := range dataList {//fmt.Printf("pk: %d, uid1: %d, uid2: %d, id: %d, usid: %d, tm: %v, tm1: %v, tm2: %v, draf: %s, io: %t, del: %t, t: %d\n",//	d.Pk, d.Uid1, d.Uid2, d.Id, d.Usid, d.Tm, d.Tm1, d.Tm2, d.Draf, d.Io, d.Del, d.T)fmt.Printf("pk: %d, uid1: %d, uid2: %d, id: %d tm: %v \n", d.Pk, d.Uid1, d.Uid2, d.Id, d.Tm)}return nil
}

9. 倒序

这个库的说明并不详细，readme.md还是过时的，chatgtp给的信息也是错误很多，目前根据测试发现，在设置排序方式时：

在 Cassandra 中，ORDER BY 子句需要按照聚簇键的声明顺序指定。在表定义中，聚簇键是 (uid1, tm, id)，所以需要按照这个顺序指定 ORDER BY。

在代码中，需要按照以下方式指定 ORDER BY：

builder := qb.Select("chatdata.pchat").Columns(pchatMetadata.Columns...)builder.Where(qb.Eq("pk"), qb.Eq("uid1"), qb.GtOrEq("tm"), qb.LtOrEq("tm"))builder.OrderBy("uid1", qb.DESC)//builder.OrderBy("tm", qb.DESC)//builder.OrderBy("id", qb.DESC)// 写一个就够了builder.AllowFiltering()//builder.Limit(pageSize)q := builder.Query(*session)defer q.Release()q.Consistency(gocql.One)tm1, _ := string2timeLoc("2024-01-27 13:24:00")tm2, _ := string2timeLoc("2024-01-27 13:25:56")q.Bind(1, 1001, tm1, tm2)

其中，pk 作为分区键，不能排序，而聚簇的键需要按照顺序指定，其中不能混！要么都是升序，要么都是降序，否则执行时候报错“Unsupported order by relation”。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/651688.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Redis数据类型及底层实现

Redis数据类型及底层实现

文章目录 1.3.1 5种基本数据类型1.3.1.1 总结篇1.3.1.2 底层源码引入篇1.3.1.2.1 redis是字典数据库KV键值对到底是什么1.3.1.2.2 数据类型视角1.3.1.2.3 数据模型解析&#xff08;重点&#xff09;1.3.1.2.4 redisObjec1.3.1.2.5 SDS 1.3.1.3 String1.3.1.3.1 底层分析1.3.1.3…
阅读更多...
uniCloud 免费版和商用版

uniCloud 免费版和商用版

概述 uniCloud为每个开发者提供一个免费的服务空间&#xff0c;更低门槛按量付费是serverless的特色&#xff0c;如果没有消耗硬件资源&#xff0c;就完全不用付款serverless比传统的云主机更便宜传统云主机一旦被攻击&#xff0c;高防价格非常昂贵。而uniCloud无需支付高防费…
阅读更多...
k8s的图形化工具rancher

k8s的图形化工具rancher

1、rancher&#xff1a;是一个开源的企业级多集群的k8s管理平台 2、rancher和k8s的区别 &#xff08;1&#xff09;都是为了容器的调度和编排系统 &#xff08;2&#xff09;但rancher不仅能够调度&#xff0c;还能管理k8s集群&#xff0c;自带监控&#xff08;普罗米修斯&a…
阅读更多...
MQ面试题合集

MQ面试题合集

前言 前文介绍了mysql,redis相关知识&#xff0c;并概述了向量数据库相关。今天我们学习消息中间件相关知识。 MQ面试题合集 什么是消息队列&#xff08;MQ&#xff09;&#xff1f; 消息队列&#xff08;MQ&#xff09;是一种应用程序对应用程序的通信方法&#xff0c;通过…
阅读更多...
QEMU用户态模拟运行，S390X、ARMv7L/ELF程序（Ubuntu）

QEMU用户态模拟运行，S390X、ARMv7L/ELF程序（Ubuntu）

准备虚拟机&#xff1a; sudo apt-get install qemu-system qemu-user-static -y 准备交叉编译器环境&#xff1a; 在 Ubuntu 18.04 x86_64 上面安装 Linux-ARMv7 A/L GCC编译器-CSDN博客 S390X交叉编译器 ############### sudo apt install gcc-s390x-linux-gnu sudo apt in…
阅读更多...
【Linux】第三十六站：信号

【Linux】第三十六站：信号

文章目录 一、信号的概念1.信号概念2.前台与后台进程3.信号的处理4.硬件层面5.信号与我们的代码是异步的 二、信号的产生1.产生的方式2.键盘组合键3.kill命令4.系统调用4.1 kill系统调用4.2 raise4.3 abort 5.异常软件条件5.1 异常产生信号5.2 alarm&#xff08;软件条件产生信…
阅读更多...
【MySQL】学习如何通过DML更新数据库的数据

【MySQL】学习如何通过DML更新数据库的数据

&#x1f308;个人主页: Aileen_0v0 &#x1f525;热门专栏: 华为鸿蒙系统学习|计算机网络|数据结构与算法 ​&#x1f4ab;个人格言:“没有罗马,那就自己创造罗马~” #mermaid-svg-QIqURn9fNFMjLD9l {font-family:"trebuchet ms",verdana,arial,sans-serif;font-siz…
阅读更多...
形态学操作之腐蚀及膨胀的数学原理——数字图像处理

形态学操作之腐蚀及膨胀的数学原理——数字图像处理

在数学上&#xff0c;腐蚀和膨胀是集合论的操作&#xff0c;通常应用于图像处理中的二值图像。可以将二值图像视为一个集合&#xff0c;其中包含所有像素值为1&#xff08;或高亮&#xff09;的像素点的集合。这些操作都是基于图像与一个给定的结构元素&#xff08;一个小的形状…
阅读更多...
【Go 快速入门】数组 | 切片 | 映射 | 函数 | 结构体 | 方法和接收者

【Go 快速入门】数组 | 切片 | 映射 | 函数 | 结构体 | 方法和接收者

文章目录 数组切片append 函数copy 函数删除元素 映射delete 函数 函数init 特殊的函数defer 语句panic / recover 错误处理 类型结构体内存对齐JSON 序列化与反序列化方法和接收者 项目代码地址&#xff1a;03-ArraySliceMapFuncStruct 数组 基本格式&#xff1a;var 数组变…
阅读更多...
Go 命令行解析 flag 包之快速上手

Go 命令行解析 flag 包之快速上手

本篇文章是 Go 标准库 flag 包的快速上手篇。 概述 开发一个命令行工具&#xff0c;视复杂程度&#xff0c;一般要选择一个合适的命令行解析库&#xff0c;简单的需求用 Go 标准库 flag 就够了&#xff0c;flag 的使用非常简单。 当然&#xff0c;除了标准库 flag 外&#x…
阅读更多...
STL标准库（五） 算法，伪函数与空间适配器

STL标准库（五） 算法，伪函数与空间适配器

伪函数 仿函数&#xff1a;像函数但不是函数&#xff0c;一般有两种形式struct或class struct或class之所以能够进行仿函数的编写是因为他们可以进行()的运算符重载 int Min(int nNumberA, int nNumberB) 这是一个函数 { return nNumberA < nNumberB ? nNumberA : nNum…
阅读更多...
Linux 网络流量相关工具

Linux 网络流量相关工具

本文聚焦于网络流量的查看、端口占用查看。至于网络设备的管理和配置&#xff0c;因为太过复杂且不同发行版有较大差异&#xff0c;这里就不赘述&#xff0c;后面看情况再写。 需要注意的是&#xff0c;这里列出的每一个工具都有丰富的功能&#xff0c;流量/端口信息查看只是其…
阅读更多...
使用vue_cli脚手架创建Vue项目（cmd和图形化方式）

使用vue_cli脚手架创建Vue项目（cmd和图形化方式）

使用vue_cli脚手架创建Vue项目&#xff08;cmd和图形化方式&#xff09; 创建项目(cmd方式) vue create vue_cli1.方向键选择manually select feature(手动选择方式创建)&#xff0c;回车 2.按空格键选择需要的组件&#xff1a;Babel、PWA、Router、Vuex、CSS&#xff0c;回…
阅读更多...
Linux - 数据流重定向、管道符、环境变量配置文件的加载

Linux - 数据流重定向、管道符、环境变量配置文件的加载

概述 想了解Linux编程&#xff0c;shell脚本是绕不开的关键知识点&#xff0c;原计划写一个整篇来分享shell的来龙去脉&#xff0c;但知识点过于繁杂&#xff0c;先分享一下学习shell的准备工作&#xff0c;数据流重定向、管道符、环境变量配置文件的加载&#xff0c;有助于知…
阅读更多...
Linux之安装配置CentOS 7

Linux之安装配置CentOS 7

一、CentOS简介 CentOS&#xff08;Community Enterprise Operating System&#xff0c;中文意思是社区企业操作系统&#xff09;是Linux发行版之一&#xff0c;它是来自于Red Hat Enterprise Linux依照开放源代码规定释出的源代码所编译而成。由于出自同样的源代码&#xff0c…
阅读更多...
YOLOv8改进 | Conv篇 | 结合Dual思想利用HetConv创新一种全新轻量化结构CSPHet（参数量下降70W）

YOLOv8改进 | Conv篇 | 结合Dual思想利用HetConv创新一种全新轻量化结构CSPHet（参数量下降70W）

一、本文介绍 本文给大家带来的改进机制是我结合Dual的思想利用HetConv提出一种全新的结构CSPHet，我们将其用于替换我们的C2f结构，可以将参数降低越75W，GFLOPs降低至6.6GFLOPs，同时本文结构为我独家创新，全网无第二份，非常适合用于发表论文，该结构非常灵活，利用Dual卷…
阅读更多...
CSS探索浏览器兼容性

CSS探索浏览器兼容性

学习如何探索浏览器的兼容性对于编写跨浏览器兼容的CSS代码非常重要。以下是一些学习CSS兼容性的方法&#xff1a; MDN文档&#xff1a;Mozilla开发者网络&#xff08;MDN&#xff09;提供了广泛而详细的CSS文档&#xff0c;其中包含有关CSS属性、选择器和功能的信息。在MDN上…
阅读更多...
机器学习之pandas库学习

机器学习之pandas库学习

这里写目录标题 pandas介绍pandas核心数据结构SeriesDataFrameDataFrame的创建列访问列添加列删除行访问行添加行删除数据修改 pandas介绍 pandas是基于NumPy 的一种工具&#xff0c;该工具是为了解决数据分析任务而创建的。Pandas 纳入 了大量库和一些标准的数据模型&#xff…
阅读更多...
谷歌seo服务商如何选择？

谷歌seo服务商如何选择？

选择谷歌SEO服务商时&#xff0c;要考虑他们的经验、专业知识、成功案例、透明度、合规性、定制能力、时间线、客户支持、沟通以及是否能够建立长期合作关系。综合评估这些因素&#xff0c;确保找到一个可信赖的合作伙伴&#xff0c;能够帮助您提升网站在谷歌搜索中的表现&…
阅读更多...
PHP操作Mysql记录数多引发的空白错误

PHP操作Mysql记录数多引发的空白错误

1 错误由来 php操作三张表&#xff0c;一张表有近四十万条记录&#xff0c;另外两张表记录数在三万左右&#xff0c;三张表又关联。应用左连接left join。 $qLStr "select pu.pd_no, pu.common_name, pu.purchase_cost, pu.medication_area, pu.total_dosage, pu.contro…
阅读更多...
最新文章

Copyright @ 2022~2023