时尚大气网站设计,网站建设企业网站制作平台,网站特色分析,工信部企业网站备案一、使用scyllaDb的原因
目前开源的聊天软件主要还是使用mysql存储数据#xff0c;数据量大的时候比较麻烦#xff1b;
我打算使用scyllaDB存储用户的聊天记录#xff0c;主要考虑的优点是#xff1a;
1#xff09;方便后期线性扩展服务器#xff1b;
2#xff09;p…一、使用scyllaDb的原因
目前开源的聊天软件主要还是使用mysql存储数据数据量大的时候比较麻烦
我打算使用scyllaDB存储用户的聊天记录主要考虑的优点是
1方便后期线性扩展服务器
2partition更方便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 chatdatacluster.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来聚簇
相关代码如下
// 定义表的元数据
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:pkUid1 int db:uid1Uid2 int db:uid2Id int db:idUsid int db:usidTm time.Time db:tmTm1 time.Time db:tm1Tm2 time.Time db:tm2Draf string db:drafIo bool db:ioDel bool db:delT 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 index1000; 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中加载已经落库的部分再根据时间段加载
这里测试的是从某个ID900的条目之后加载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”。
