提交

api/etc/message-api.yaml

@@ -1,6 +1,6 @@
 Name: message-api
 Host: 0.0.0.0
-Port: 8888
+Port: 8911
 
 MessageCenter:
   Etcd:

entity/message.go

@@ -16,7 +16,7 @@ var Mysql *mysql.Mysql
 var Mysql11 *sql.DB
 
 type Message struct {
-	Id            string       `xorm:"id" form:"id" json:"id"`
+	Id            string    `xorm:"id" form:"id" json:"id"`
 	AppId         string    `xorm:"appid" form:"appid" json:"appid"`
 	ReceiveUserid string    `xorm:"receive_userid" form:"receive_userid" json:"receive_userid"` // 接收方用户id
 	ReceiveName   string    `xorm:"receive_name" form:"receive_name" json:"receive_name"`
@@ -33,9 +33,9 @@ type Message struct {
 }
 
 type Conversation struct {
-	Id          string       `xorm:"id" form:"id" json:"id"`
+	Id          string    `xorm:"id" form:"id" json:"id"`
 	AppId       string    `xorm:"appid" form:"appid" json:"appid"`
-	Key         string    `xorm:"key" form:"key" json:"key"`
+	SecretKey   string    `xorm:"secret_key" form:"secret_key" json:"secret_key"`
 	UserId      string    `xorm:"user_id" form:"user_id" json:"user_id"`
 	ReceiveId   string    `xorm:"receive_id" form:"receive_id" json:"receive_id"`
 	ReceiveName string    `xorm:"receive_name" form:"receive_name" json:"receive_name"`

rpc/etc/message.yaml

@@ -5,15 +5,15 @@ Etcd:
   - 127.0.0.1:2379
   - 127.0.0.1:2379
   Key: message.rpc
-  Timeout: 6000
+Timeout: 10000
 Mysql: jianyu:topnet@123@tcp(am-2ze6crwd6bb0283jn167320o.ads.aliyuncs.com:3306)/message?timeout=10s&interpolateParams=true
 DataSource:
-    DbName: message
-    Address: am-2ze6crwd6bb0283jn167320o.ads.aliyuncs.com
+    DbName: messagetest
+    Address: 192.168.3.109:4000
     UserName: jianyu
-    PassWord: topnet@123
-    MaxOpenConns: 5
-    MaxIdleConns: 5
+    PassWord: top@123
+    MaxOpenConns: 10
+    MaxIdleConns: 10
 FileSystemConf:
   Etcd:
     Hosts:

rpc/message.go

@@ -10,7 +10,6 @@ import (
 	"app.yhyue.com/moapp/MessageCenter/rpc/internal/svc"
 	"app.yhyue.com/moapp/MessageCenter/rpc/message"
 	"app.yhyue.com/moapp/jybase/mysql"
-	"database/sql"
 	"flag"
 	"fmt"
 	"github.com/tal-tech/go-zero/core/conf"
@@ -72,15 +71,15 @@ func init() {
 
 	//连接mysql
 	//log.Println(config.ConfigJson.DataSource)
-	//entity.Engine, err = xorm.NewEngine("mysql", config.ConfigJson.DataSource)
+	//entity.Engine, err = xorm.NewEngine("mysql", config.ConfigJson.Mysql)
 	//log.Println(err)
 	//entity.Engine.ShowSQL(true)
 	//if err != nil {
 	//	log.Fatal("数据库连接失败:", err)
 	//}
-	//fmt.Println(config.ConfigJson.DataSource + "链接成功")
+	//fmt.Println(config.ConfigJson.Mysql + "链接成功")
 
-	entity.Mysql11, err = sql.Open("mysql", config.ConfigJson.Mysql)
+	/*entity.Mysql11, err = sql.Open("mysql", config.ConfigJson.Mysql)
 	if err != nil {
 		panic(err.Error())
 	}
@@ -89,5 +88,5 @@ func init() {
 	// 设置最大闲置的连接数
 	entity.Mysql11.SetMaxIdleConns(20)
 	// 设置连接的最大生命周期，默认连接总是可重用。
-	entity.Mysql11.SetConnMaxLifetime(time.Hour)
+	entity.Mysql11.SetConnMaxLifetime(time.Hour)*/
 }

rpc/test/message_test.go

@@ -27,14 +27,11 @@ func  Test_DeleteSingleMessage(t *testing.T)  {
 	ctx, _ := context.WithTimeout(context.Background(), 5*time.Second)
 	std := messageclient.NewMessage(zrpc.MustNewClient(zrpc.RpcClientConf{Etcd: discov.EtcdConf{Hosts: []string{"127.0.0.1:2379"}, Key: "message.rpc"}}))
 	//std := stdlibclient.NewStdlib(zrpc.MustNewClient(zrpc.RpcClientConf{Etcd: discov.EtcdConf{Hosts: []string{"127.0.0.1:2379"}, Key: "jydocs.stdlib.rpc"}}))
-	req := &messageclient.DeleteSingleMessageRequest{Id: "3",Appid: "10000"}
-	res, err := std.DeleteSingleMessage(ctx, req)
+	//req := &messageclient.DeleteSingleMessageRequest{Id: "1400523738618150912",Appid: "10000"}
+	req := &messageclient.ChangeReadStatusRequest{Id: "1400523738618150912",Appid: "10000",ReadStatus:1}
+	res, err := std.ChangeReadStatus(ctx, req)
 	log.Println("err ", err,res)
-	//if res.State == true {
-	//	log.Println("req true")
-	//} else {
-	//	log.Println("req false", res)
-	//}
+
 }
 func  Test_DeleteMultipleMessage(t *testing.T)  {
 	ctx, _ := context.WithTimeout(context.Background(), 5*time.Second)

rpc/test/send_test.go

@@ -38,12 +38,12 @@ func Test_FindUserMsg(t *testing.T) {
 	std := messageclient.NewMessage(zrpc.MustNewClient(zrpc.RpcClientConf{Etcd: discov.EtcdConf{Hosts: []string{"127.0.0.1:2379"}, Key: "message.rpc"}}))
 	req := &messageclient.FindUserMsgReq{
 		Appid:         "10000",
-		ReceiveUserId: "6042120adca8410f1ef2ec84",
-		UserId:        "5fa3bb6d059e75bcdf8dab6a",
+		ReceiveUserId: "60d3ded1ecf51e0c15edd5b4",
+		UserId:        "60d3ded1ecf51e0c15edd5b4",
 		OffSet:        int64(0),
 		PageSize:      int64(10),
-		MsgType:       -1,
-		Read:          -1,
+		MsgType:       7,
+		Read:          0,
 	}
 	res, err := std.FindUserMsg(ctx, req)
 	log.Println("err ", err, res.Code, res.Message, res.Count, res.Data)

service/sendMsg.go

@@ -302,41 +302,40 @@ func EtcdSetCountZero(userId, msgType string) {
 func MultSave(this message.MultipleSaveMsgReq) (int64, string) {
 	userIdArr := strings.Split(this.UserIds, ",")
 	userNameArr := strings.Split(this.UserNames, ",")
-	log.Println(len(userIdArr), len(userNameArr))
+	log.Println("参数：", len(userIdArr), len(userNameArr))
 	if len(userIdArr) > 0 {
 		var errCount int64
 		for k, v := range userIdArr {
 			log.Println("k--------", k, v)
 			if v == "" {
-				return 0, "调用结束"
+				return errCount, "调用结束"
 			}
 			userName := userNameArr[k]
 			//消息数组
+			bT := time.Now()
 			c := entity.Mysql.Count("conversation", map[string]interface{}{"receive_id": v, "send_id": this.SendUserId})
-			log.Println("查询数量：", c)
+			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) values ("%s",'%s','%s','%s','%s','%s','%s','%d','%s',0,'%s',0,1);`
 			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"))
-			if c < 1 {
-				sql1 := `INSERT INTO conversation(appid,` + "`key`" + `,user_id,receive_id,receive_name,send_id,send_name,sort,createtime) values ('%s','','%s','%s','%s','%s','%s',0,'%s');`
+			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"))
 				ok := entity.Mysql.ExecTx("发送消息事务", func(tx *sql.Tx) bool {
 					//插入会话表
-					bT := time.Now() //开始时间
-					_, err := entity.Mysql.DB.Exec(sql1)
-					log.Println("**********1", err)
-					sql2 := `INSERT INTO conversation(appid,` + "`key`" + `,user_id,receive_id,receive_name,send_id,send_name,sort,createtime) values ('%s','','%s','%s','%s','%s','%s',0,'%s');`
+					//开始时间
+					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"))
-					_, err = entity.Mysql.DB.Exec(sql2)
-					log.Println("**********2", err)
+					in2 := entity.Mysql.InsertBySqlByTx(tx, sql2)
 					//插入消息表
-					_, err = entity.Mysql.DB.Exec(sql3)
-					eT := time.Since(bT) // 从开始到当前所消耗的时间
-					log.Println("存储耗时：", eT)
-					if err != nil {
-						return false
-					}
-					return true
+					in3 := entity.Mysql.InsertBySqlByTx(tx, sql3)
+					log.Println("存储耗时：", time.Since(bT))
+					log.Println(in1, in2, in3)
+
+					return in1 > -1 && in2 > -1 && in3 > -1
 				})
+				log.Println("执行事务是否成功：", ok)
 				if !ok {
 					errCount++
 					continue
@@ -344,8 +343,8 @@ func MultSave(this message.MultipleSaveMsgReq) (int64, string) {
 				go EtcdCountAdd(v, strconv.Itoa(int(this.MsgType)))
 
 			} else {
-				_, err := entity.Mysql.DB.Exec(sql3)
-				if err == nil {
+				in := entity.Mysql.InsertBySql(sql3)
+				if in > -1 {
 					go EtcdCountAdd(v, strconv.Itoa(int(this.MsgType)))
 				} else {
 					errCount++