当前位置：首页 > news >正文

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

news 文章来源：https://blog.csdn.net/robinfoxnan/article/details/135880916 2025/5/15 13:06:41

一、使用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)，它的影响如下：

分区键 (pk): 数据将按照 pk 的值进行分区。相同 pk 的数据会被存储在同一分区中。
聚簇键 (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来聚簇；

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”。