|
|
@@ -1,44 +1,41 @@
|
|
|
package service
|
|
|
|
|
|
import (
|
|
|
- "context"
|
|
|
+ "app.yhyue.com/moapp/jybase/redis"
|
|
|
"database/sql"
|
|
|
- "encoding/json"
|
|
|
"fmt"
|
|
|
+ "github.com/tal-tech/go-zero/core/logx"
|
|
|
"log"
|
|
|
"strconv"
|
|
|
"strings"
|
|
|
"time"
|
|
|
|
|
|
- "app.yhyue.com/moapp/MessageCenter/util"
|
|
|
- "go.etcd.io/etcd/client/v3/concurrency"
|
|
|
-
|
|
|
"app.yhyue.com/moapp/MessageCenter/entity"
|
|
|
"app.yhyue.com/moapp/MessageCenter/rpc/message"
|
|
|
+ "app.yhyue.com/moapp/MessageCenter/util"
|
|
|
)
|
|
|
|
|
|
// 类型的顺序
|
|
|
const order = "1,4"
|
|
|
-const EtcdCount = "Count.%s.%s" //etcd 消息未读数量 Count.用户id.消息类型=数量
|
|
|
+const MsgCountKey = "count_%s_%s" //redis 消息未读数量 Count.用户id.消息类型=数量
|
|
|
+const redisModule = "msgCount"
|
|
|
+
|
|
|
+/*var (
|
|
|
+ UserLockMap = map[string]*sync.Mutex{}
|
|
|
+ //MainLock = sync.Mutex{}
|
|
|
+)*/
|
|
|
|
|
|
func SendMsg(this message.SendMsgRequest) (int64, string) {
|
|
|
- //orm := entity.Engine.NewSession()
|
|
|
- //defer orm.Close()
|
|
|
- //err := orm.Begin()
|
|
|
- //fmt.Println(err)
|
|
|
- //count := entity.Mysql11.Query("conversation", map[string]interface{}{"receive_id": this.ReceiveUserId, "send_id": this.SendUserId})
|
|
|
|
|
|
r, err := entity.Mysql11.Query("select count(*) as c from conversation where receive_id = ? and send_id = ? ", this.ReceiveUserId, this.SendUserId)
|
|
|
c := 0
|
|
|
- log.Println("查询结果", r)
|
|
|
for r.Next() {
|
|
|
err := r.Scan(&c)
|
|
|
if err != nil {
|
|
|
panic(err.Error())
|
|
|
}
|
|
|
}
|
|
|
-
|
|
|
- log.Println("查询数量：", c)
|
|
|
+ logx.Info("查询数量：", c)
|
|
|
|
|
|
sql3 := `INSERT INTO message(appid,receive_userid,receive_name,send_userid,send_name,title,content,msg_type,link,cite_id,createtime,isRead,isdel)
|
|
|
values ("%s",'%s','%s','%s','%s','%s','%s','%d','%s',0,'%s',0,1);`
|
|
|
@@ -68,7 +65,7 @@ func SendMsg(this message.SendMsgRequest) (int64, string) {
|
|
|
}
|
|
|
_, err = entity.Mysql11.Exec(sql3)
|
|
|
if err == nil {
|
|
|
- EtcdCountAdd(this.ReceiveUserId, strconv.Itoa(int(this.MsgType)))
|
|
|
+ MsgCountAdd(this.ReceiveUserId, strconv.Itoa(int(this.MsgType)))
|
|
|
return 1, "消息发送成功"
|
|
|
}
|
|
|
return 0, "消息发送失败"
|
|
|
@@ -94,12 +91,8 @@ func FindUserMsg(this message.FindUserMsgReq) message.FindUserMsgRes {
|
|
|
//count, err = orm.Table("message").Where("((receive_userid = ? and send_userid = ?) or (receive_userid = ? and send_userid = ?)) and isdel = ? and appid = ?"+q, this.UserId, this.ReceiveUserId, this.ReceiveUserId, this.UserId, 1, this.Appid).Count()
|
|
|
data := message.FindUserMsgRes{}
|
|
|
if count > 0 {
|
|
|
- /*err = orm.Table("message").Select("*").Where("((receive_userid = ? and send_userid = ?) or (receive_userid = ? and send_userid = ?)) and isdel = ? and appid = ?"+q, this.UserId, this.ReceiveUserId, this.ReceiveUserId, this.UserId, 1, this.Appid).
|
|
|
- OrderBy("createtime desc").
|
|
|
- Limit(int(this.PageSize), (int(this.OffSet)-1)*int(this.PageSize)).
|
|
|
- Find(&messages)*/
|
|
|
res := entity.Mysql.Find("message", cquery, "", "createtime desc", (int(this.OffSet)-1)*int(this.PageSize), int(this.PageSize))
|
|
|
- log.Println("数据：", res)
|
|
|
+ //log.Println("数据：", res)
|
|
|
if res != nil && len(*res) > 0 {
|
|
|
for _, v := range *res {
|
|
|
_id := util.Int64All(v["id"])
|
|
|
@@ -136,8 +129,6 @@ func FindUserMsg(this message.FindUserMsgReq) message.FindUserMsgRes {
|
|
|
|
|
|
// 指定分类未读消息合计
|
|
|
func ClassCountUnread(msgType int, userId string, appId string) (int64, string, int64) {
|
|
|
- //orm := entity.Engine
|
|
|
- //count, err := orm.Table("message").Where("msg_type=? and receive_userid=? and isdel=1 and appid=? and isRead=0", msgType, userId, appId).Count()
|
|
|
query := map[string]interface{}{
|
|
|
"msg_type": msgType,
|
|
|
"receive_userid": userId,
|
|
|
@@ -149,162 +140,29 @@ func ClassCountUnread(msgType int, userId string, appId string) (int64, string,
|
|
|
return 1, "查询指定分类未读消息成功", count
|
|
|
}
|
|
|
|
|
|
-// etcd数量加1
|
|
|
-func EtcdCountAdd(userId, msgType string) {
|
|
|
- s1, err := concurrency.NewSession(entity.EtcdCli)
|
|
|
- if err != nil {
|
|
|
- log.Println(err)
|
|
|
- }
|
|
|
- defer s1.Close()
|
|
|
- keyString := fmt.Sprintf(EtcdCount, userId, msgType)
|
|
|
- m1 := concurrency.NewMutex(s1, keyString)
|
|
|
- // 会话s1获取锁
|
|
|
- if err := m1.Lock(context.TODO()); err != nil {
|
|
|
- log.Println("EtcdCountAdd获取锁失败", err)
|
|
|
- }
|
|
|
- // 操作数量
|
|
|
- ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
|
|
- resp, err := s1.Client().Get(ctx, keyString)
|
|
|
- cancel()
|
|
|
- if err != nil {
|
|
|
- fmt.Printf("get from etcd failed, err:%v\n", err)
|
|
|
-
|
|
|
- // 释放锁
|
|
|
- if err := m1.Unlock(context.TODO()); err != nil {
|
|
|
- log.Println("EtcdCountAdd释放锁失败", err)
|
|
|
- }
|
|
|
- return
|
|
|
- }
|
|
|
- var count int
|
|
|
- for _, ev := range resp.Kvs {
|
|
|
- if ev.Value != nil {
|
|
|
- err := json.Unmarshal([]byte(ev.Value), &count)
|
|
|
- if err != nil {
|
|
|
- log.Println("etcd get err:", err)
|
|
|
- } else {
|
|
|
- break
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- }
|
|
|
- ctx, cancel = context.WithTimeout(context.Background(), time.Second)
|
|
|
- _, err = s1.Client().Put(ctx, keyString, strconv.Itoa(count+1))
|
|
|
- cancel()
|
|
|
- if err != nil {
|
|
|
- fmt.Printf("put to etcd failed, err:%v\n", err)
|
|
|
- // 释放锁
|
|
|
- if err := m1.Unlock(context.TODO()); err != nil {
|
|
|
- log.Println("EtcdCountAdd释放锁失败2", err)
|
|
|
- }
|
|
|
- return
|
|
|
- }
|
|
|
-
|
|
|
- // 释放锁
|
|
|
- if err := m1.Unlock(context.TODO()); err != nil {
|
|
|
- log.Println("EtcdCountAdd释放锁失败3", err)
|
|
|
- }
|
|
|
+// MsgCountAdd 消息未读数量加1
|
|
|
+func MsgCountAdd(userId, msgType string) bool {
|
|
|
+ keyString := fmt.Sprintf(MsgCountKey, userId, msgType)
|
|
|
+ in := redis.Incr(redisModule, keyString)
|
|
|
+ return in > 0
|
|
|
}
|
|
|
|
|
|
-//单条消息，-1
|
|
|
-func EtcdCountMinusOne(userId, msgType string) {
|
|
|
- s1, err := concurrency.NewSession(entity.EtcdCli)
|
|
|
- if err != nil {
|
|
|
- log.Println(err)
|
|
|
- }
|
|
|
- defer s1.Close()
|
|
|
- keyString := fmt.Sprintf(EtcdCount, userId, msgType)
|
|
|
- m1 := concurrency.NewMutex(s1, keyString)
|
|
|
- // 会话s1获取锁
|
|
|
- if err := m1.Lock(context.TODO()); err != nil {
|
|
|
- log.Println("EtcdCountMinusOne获取锁失败", err)
|
|
|
- }
|
|
|
- // 操作数量
|
|
|
- ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
|
|
- resp, err := s1.Client().Get(ctx, keyString)
|
|
|
- cancel()
|
|
|
- if err != nil {
|
|
|
- fmt.Printf("get from etcd failed, err:%v\n", err)
|
|
|
-
|
|
|
- // 释放锁
|
|
|
- if err := m1.Unlock(context.TODO()); err != nil {
|
|
|
- log.Println("EtcdCountMinusOne释放锁失败", err)
|
|
|
- }
|
|
|
- return
|
|
|
- }
|
|
|
- var count int
|
|
|
- for _, ev := range resp.Kvs {
|
|
|
- if ev.Value != nil {
|
|
|
- err := json.Unmarshal([]byte(ev.Value), &count)
|
|
|
- if err != nil {
|
|
|
- log.Println("etcd get err:", err)
|
|
|
- } else {
|
|
|
- break
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- }
|
|
|
- ctx, cancel = context.WithTimeout(context.Background(), time.Second)
|
|
|
- var fin string
|
|
|
- if count > 0 {
|
|
|
- fin = strconv.Itoa(count - 1)
|
|
|
- } else {
|
|
|
- fin = "0"
|
|
|
- }
|
|
|
- _, err = s1.Client().Put(ctx, keyString, fin)
|
|
|
- cancel()
|
|
|
- if err != nil {
|
|
|
- fmt.Printf("put to etcd failed, err:%v\n", err)
|
|
|
- // 释放锁
|
|
|
- if err := m1.Unlock(context.TODO()); err != nil {
|
|
|
- log.Println("EtcdCountMinusOne释放锁失败2", err)
|
|
|
- }
|
|
|
- return
|
|
|
- }
|
|
|
-
|
|
|
- // 释放锁
|
|
|
- if err := m1.Unlock(context.TODO()); err != nil {
|
|
|
- log.Println("EtcdCountMinusOne释放锁失败3", err)
|
|
|
- }
|
|
|
+// MsgCountMinusOne 根据消息类型未读消息数量减1
|
|
|
+func MsgCountMinusOne(userId, msgType string) bool {
|
|
|
+ keyString := fmt.Sprintf(MsgCountKey, userId, msgType)
|
|
|
+ in := redis.Decrby(redisModule, keyString, 1)
|
|
|
+ return in > 0
|
|
|
}
|
|
|
|
|
|
-// 消息类别置0
|
|
|
-func EtcdSetCountZero(userId, msgType string) {
|
|
|
- log.Println(" 消息类别置0", userId, msgType)
|
|
|
- s1, err := concurrency.NewSession(entity.EtcdCli)
|
|
|
- if err != nil {
|
|
|
- log.Println(err)
|
|
|
- }
|
|
|
- defer s1.Close()
|
|
|
- keyString := fmt.Sprintf(EtcdCount, userId, msgType)
|
|
|
- m1 := concurrency.NewMutex(s1, keyString)
|
|
|
- // 会话s1获取锁
|
|
|
- if err := m1.Lock(context.TODO()); err != nil {
|
|
|
- log.Println("EtcdSetCountZero获取锁失败", err)
|
|
|
- }
|
|
|
- // 操作数量
|
|
|
-
|
|
|
- ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
|
|
- _, err = s1.Client().Put(ctx, keyString, "0")
|
|
|
- cancel()
|
|
|
- if err != nil {
|
|
|
- fmt.Printf("put to etcd failed, err:%v\n", err)
|
|
|
- // 释放锁
|
|
|
- if err := m1.Unlock(context.TODO()); err != nil {
|
|
|
- log.Println("EtcdSetCountZero释放锁失败", err)
|
|
|
- }
|
|
|
- return
|
|
|
- }
|
|
|
-
|
|
|
- // 释放锁
|
|
|
- if err := m1.Unlock(context.TODO()); err != nil {
|
|
|
- log.Println("EtcdSetCountZero释放锁失败", err)
|
|
|
- }
|
|
|
+// MsgCountZero 把该消息类型未读数量置0
|
|
|
+func MsgCountZero(userId, msgType string) bool {
|
|
|
+ keyString := fmt.Sprintf(MsgCountKey, userId, msgType)
|
|
|
+ return redis.Put(redisModule, keyString, 0, -1)
|
|
|
}
|
|
|
|
|
|
func MultSave(this message.MultipleSaveMsgReq) (int64, string) {
|
|
|
userIdArr := strings.Split(this.UserIds, ",")
|
|
|
userNameArr := strings.Split(this.UserNames, ",")
|
|
|
- log.Println("参数：", len(userIdArr), len(userNameArr))
|
|
|
if len(userIdArr) > 0 {
|
|
|
var errCount int64
|
|
|
for k, v := range userIdArr {
|
|
|
@@ -313,45 +171,37 @@ func MultSave(this message.MultipleSaveMsgReq) (int64, string) {
|
|
|
}
|
|
|
userName := userNameArr[k]
|
|
|
//消息数组
|
|
|
- bT := time.Now()
|
|
|
c := entity.Mysql.Count("conversation", map[string]interface{}{"receive_id": v, "send_id": this.SendUserId})
|
|
|
- // log.Println("count", c)
|
|
|
- //m := entity.Mysql.SelectBySql("select count(id) from conversation where receive_id=? and send_id=?", v, this.SendUserId)
|
|
|
sql3 := `INSERT INTO message(appid,receive_userid,receive_name,send_userid,send_name,title,content,msg_type,link,cite_id,createtime,isRead,isdel,msg_log_id) values ("%s",'%s','%s','%s','%s','%s','%s',%d,'%s',0,'%s',0,1,%d);`
|
|
|
sql3 = fmt.Sprintf(sql3, this.Appid, v, userName, this.SendUserId, this.SendName, this.Title, this.Content, this.MsgType, this.Link, time.Now().Format("2006-01-02 15:04:05"), this.MsgLogId)
|
|
|
- log.Println("插入消息表sql3：", sql3)
|
|
|
if c <= 0 {
|
|
|
sql1 := `INSERT INTO conversation(appid,secret_key,user_id,receive_id,receive_name,send_id,send_name,sort,createtime) values ('%s','','%s','%s','%s','%s','%s',0,'%s');`
|
|
|
sql1 = fmt.Sprintf(sql1, this.Appid, this.SendUserId, v, userName, this.SendUserId, this.SendName, time.Now().Format("2006-01-02 15:04:05"))
|
|
|
- log.Println("插入会话表sql1：", sql1)
|
|
|
ok := entity.Mysql.ExecTx("发送消息事务", func(tx *sql.Tx) bool {
|
|
|
//插入会话表
|
|
|
- //开始时间
|
|
|
in1 := entity.Mysql.InsertBySqlByTx(tx, sql1)
|
|
|
sql2 := `INSERT INTO conversation(appid,secret_key,user_id,receive_id,receive_name,send_id,send_name,sort,createtime) values ('%s','','%s','%s','%s','%s','%s',0,'%s');`
|
|
|
sql2 = fmt.Sprintf(sql2, this.Appid, v, this.SendUserId, this.SendName, v, userName, time.Now().Format("2006-01-02 15:04:05"))
|
|
|
- log.Println("插入会话表sql2：", sql2)
|
|
|
in2 := entity.Mysql.InsertBySqlByTx(tx, sql2)
|
|
|
//插入消息表
|
|
|
in3 := entity.Mysql.InsertBySqlByTx(tx, sql3)
|
|
|
- log.Println("插入消息返回 in3 id:", in3)
|
|
|
- log.Println("存储耗时：", time.Since(bT))
|
|
|
- log.Println(in1, in2, in3)
|
|
|
+ logx.Info(in1, in2, in3)
|
|
|
|
|
|
return in1 > -1 && in2 > -1 && in3 > -1
|
|
|
})
|
|
|
- log.Println("执行事务是否成功：", ok)
|
|
|
+ logx.Info("执行事务是否成功：", ok)
|
|
|
if !ok {
|
|
|
errCount++
|
|
|
continue
|
|
|
}
|
|
|
- EtcdCountAdd(v, strconv.Itoa(int(this.MsgType)))
|
|
|
-
|
|
|
+ ok1 := MsgCountAdd(v, strconv.Itoa(int(this.MsgType)))
|
|
|
+ log.Println("存redis：", ok1)
|
|
|
} else {
|
|
|
in := entity.Mysql.InsertBySql(sql3)
|
|
|
- log.Println("插入消息返回 in1 id:", in)
|
|
|
+ logx.Info("插入消息返回 in1 id:", in)
|
|
|
if in > -1 {
|
|
|
- EtcdCountAdd(v, strconv.Itoa(int(this.MsgType)))
|
|
|
+ ok := MsgCountAdd(v, strconv.Itoa(int(this.MsgType)))
|
|
|
+ log.Println("存redis：", ok)
|
|
|
} else {
|
|
|
errCount++
|
|
|
}
|
