自定义发送消息

src/customerService/customService.go

@@ -19,6 +19,7 @@ import (
 )
 
 func CustomSendMsgService(param *customMsg, sendStatus int, loginUserName string, loginUserId int) (int, error) {
+	var userIdArr []string
 	var msgLogId string
 	if param.SendMode == 2 {
 		param.SendTime = time.Now().Format(qutil.Date_Full_Layout)
@@ -31,6 +32,7 @@ func CustomSendMsgService(param *customMsg, sendStatus int, loginUserName string
 			ok1 := util.JysqlDB.Delete("sendmsg_customer_info", map[string]interface{}{"msg_id": param.Id})
 			var args []interface{}
 			for _, val := range param.UserArr {
+				userIdArr = append(userIdArr, qutil.ObjToString(val["user_id"]))
 				args = append(args, val["user_id"])
 				args = append(args, val["register_phone"])
 				args = append(args, val["call_center_company"])
@@ -82,7 +84,7 @@ func CustomSendMsgService(param *customMsg, sendStatus int, loginUserName string
 					// 执行任务
 					log.Println("创建、修改添加定时任务")
 					log.Println(param.Id, sendTime)
-					task.Task(param.Id, sendTime, param.AndroidUrl, param.IosUrl)
+					task.CustomTask(param.Id, sendTime, param.AndroidUrl, param.IosUrl)
 				})
 			}
 		}
@@ -116,6 +118,7 @@ func CustomSendMsgService(param *customMsg, sendStatus int, loginUserName string
 
 			var args []interface{}
 			for _, val := range param.UserArr {
+				userIdArr = append(userIdArr, qutil.ObjToString(val["user_id"]))
 				args = append(args, val["user_id"])
 				args = append(args, val["register_phone"])
 				args = append(args, val["call_center_company"])
@@ -181,7 +184,6 @@ func CustomSendMsgService(param *customMsg, sendStatus int, loginUserName string
 			"appid":      util.AppId,
 			"msgLogId":   msgLogId,
 		}
-		userIdArr := strings.Split(param.UserIds, ",")
 		for _, val := range userIdArr {
 			userId := val
 			if config.SysConfigs.UserIdMap[userId] != "" {
@@ -239,6 +241,7 @@ func CustomSendMsgService(param *customMsg, sendStatus int, loginUserName string
 			}
 			if i == 100 {
 				//调用消息中台
+				log.Println("发送用户id", userIds)
 				util.MultipleSaveMessage(msg, userIds, userNames)
 				userNames = ""
 				userIds = ""

src/task/task2.go

@@ -34,7 +34,7 @@ type CustomerJson struct {
 func InitTask() {
 	// 获取未执行的消息
 	tim := time.Now().Format(qutil.Date_Full_Layout)
-	msg := util.JysqlDB.SelectBySql("select id,send_usergroup_id,send_usergroup_name,msg_type,title,content,send_time,link,isdel,send_name,send_userid from message_send_log where send_status = 1 and send_time >= ? and isdel = 1", tim)
+	msg := util.JysqlDB.SelectBySql("select * from message_send_log where send_status = 1 and send_time >= ? and isdel = 1", tim)
 	log.Println(msg)
 	if msg != nil {
 		for _, val := range *msg {
@@ -52,7 +52,11 @@ func InitTask() {
 			time.AfterFunc(time.Duration(execTime)*time.Second, func() {
 				// 执行任务
 				//log.Println(msgId, sendTime)
-				Task(msgId, sendTime, androidUrl, iosUrl)
+				if qutil.IntAll(val["sign"]) == 1 {
+					CustomTask(msgId, sendTime, androidUrl, iosUrl)
+				} else {
+					Task(msgId, sendTime, androidUrl, iosUrl)
+				}
 			})
 		}
 	}
@@ -79,6 +83,26 @@ func Task(msgId int, sendTime, androidUrl, iosUrl string) {
 	// 执行完成 更新消息状态
 }
 
+func CustomTask(msgId int, sendTime, androidUrl, iosUrl string) {
+	util.OnTimeSendMap.Delete(strconv.Itoa(msgId) + sendTime)
+	log.Println("开始执行任务", msgId, sendTime, time.Now())
+	//  消息id查库 获取消息信息  库里的发送时间 和创建任务时的时间比较 发送状态为未发送 未删除 判断本次是否需要执行
+	msg := util.JysqlDB.FindOne("message_send_log", map[string]interface{}{"id": msgId, "isdel": 1, "send_status": 1, "send_time": sendTime, "send_mode": 1}, "", "")
+	if msg == nil || len(*msg) <= 0 {
+		log.Println("消息可能已经删除或者修改了时间，本次不执行发送了", msg)
+		return
+	}
+	// 更新消息为发送中
+	ok1 := util.JysqlDB.Update("message_send_log", map[string]interface{}{"id": msgId}, map[string]interface{}{"send_status": 2})
+	//log.Println("发送钱更新消息发送状态为发送中：", ok1)
+	if ok1 {
+		// 执行发送
+		code, err := CustomTaskSaveMsg(msgId, msg, androidUrl, iosUrl)
+		log.Println("定时任务执行结束 ", code, err)
+	}
+	// 执行完成 更新消息状态
+}
+
 func CommonPost(path string, param url.Values) (map[string]interface{}, error) {
 	//param["appId"] = []string{config.JyApiConfig.AppId} //公共appid
 	res, err := http.PostForm(path, param)
@@ -492,6 +516,128 @@ func TaskSaveMsg(msgId int, msg *map[string]interface{}, androidUrl, iosUrl stri
 	return 0, errors.New("发送消息出错")
 }
 
+func CustomTaskSaveMsg(msgId int, msg *map[string]interface{}, androidUrl, iosUrl string) (int, error) {
+	projectIdMap := sync.Map{}
+	orm := util.Tidb.NewSession()
+	err := orm.Begin()
+	if err != nil {
+		log.Println("简历数据库连接出错：", err)
+	}
+	msgs := *msg
+	//根据发送分组的id查询分组下的人
+	st := time.Now()
+	errCount := 0
+	userNames := ""
+	userIds := ""
+	i := 0
+	msgType := strconv.Itoa(qutil.IntAll(msgs["msg_type"]))
+	msg1 := map[string]interface{}{
+		"sendUserId": "qmx",
+		"sendName":   "剑鱼后台",
+		"title":      msgs["title"],
+		"content":    msgs["content"],
+		"msgType":    msgType,
+		"link":       msgs["link"],
+		"appid":      util.AppId,
+		"msgLogId":   strconv.Itoa(msgId),
+	}
+	user := util.JysqlDB.Find("sendmsg_customer_info", map[string]interface{}{"msg_id": msgId}, "user_id", "", -1, -1)
+	if user != nil && len(*user) > 0 {
+		for _, val := range *user {
+			userId := qutil.ObjToString(val["user_id"])
+
+			if config.SysConfigs.UserIdMap[userId] != "" {
+				userId = config.SysConfigs.UserIdMap[userId]
+			}
+			if !util.IsUtf8([]byte(userId)) {
+				log.Println("userId不是UTF8编码", userId)
+				continue
+			}
+			if _, ok := projectIdMap.Load(userId); ok {
+				continue
+			} else {
+				projectIdMap.Store(userId, true)
+			}
+			i++
+			userIds += userId + ","
+			//查询mongo库用户信息
+			userData := &map[string]interface{}{}
+			ok := false
+			var otherPushId, jgPushId, phoneType, name = "", "", "", ""
+			appVersion := ""
+			userData, ok = util.MQFW.FindById("user", userId, `"s_name":1,"s_opushid":1,"s_jpushid":1,"s_appponetype":1,"s_appversion":1`)
+			if userData != nil && len(*userData) > 0 && ok {
+				otherPushId = qutil.ObjToString((*userData)["s_opushid"])
+				jgPushId = qutil.ObjToString((*userData)["s_jpushid"])
+				phoneType = qutil.ObjToString((*userData)["s_appponetype"])
+				name = qutil.ObjToString((*userData)["s_name"])
+				appVersion = qutil.ObjToString((*userData)["s_appversion"])
+			}
+			userNames += name + ","
+			appPushUrl := "/jyapp/frontPage/messageCenter/sess/index"
+			if strings.Contains(phoneType, "iPhone") {
+				if iosUrl != "" {
+					appPushUrl = iosUrl
+				}
+			} else {
+				if androidUrl != "" {
+					appPushUrl = androidUrl
+				}
+			}
+			//实时发送发送时间为当前时间
+			// param.SendTime = time.Now().Format(qutil.Date_Full_Layout)
+			dt := map[string]interface{}{
+				"receiveUserId": userId,
+				"receiveName":   name,
+				"sendUserId":    "qmx",
+				"sendName":      "剑鱼后台",
+				"title":         msgs["title"],
+				"content":       msgs["content"],
+				"msgType":       msgType,
+				"link":          msgs["link"],
+				"appid":         util.AppId,
+			}
+			//推送消息
+			if appVersion > "3.0.3" {
+				go util.AppGrpcPush(dt, otherPushId, jgPushId, phoneType, appPushUrl)
+			}
+			log.Println("用户数量:", i)
+			if i == 100 {
+				//调用中台接口
+				log.Println("定时发送中100个用户开始发送消息")
+				util.MultipleSaveMessage(msg1, userIds, userNames)
+				userNames = ""
+				userIds = ""
+				i = 0
+				log.Println("定时发送中100个用户发送消息完成")
+			}
+		}
+		if i > 0 {
+			//调用中台接口
+			log.Println("定时发送中最后一批用户开始发送消息")
+			util.MultipleSaveMessage(msg1, userIds, userNames)
+			userNames = ""
+			userIds = ""
+			log.Println("定时发送中最后一批用户发送消息完成")
+		}
+		log.Println("定时任务,获取分组及分组下用户耗时：", time.Since(st))
+	}
+
+	sendStatus := 4
+	if errCount > 0 {
+		sendStatus = 3
+	}
+	//更新消息发送状态
+	ok2 := util.JysqlDB.Update("message_send_log", map[string]interface{}{"id": msgId}, map[string]interface{}{"send_status": sendStatus})
+	if ok2 {
+		log.Println("消息id：", msg, "发送消息成功")
+		return 1, nil
+	} else {
+		log.Println("消息id：", msg, "发送消息失败")
+	}
+	return 0, errors.New("发送消息出错")
+}
+
 //根据分组id获取分组下的用户
 func GetGroupUser(groupId string) ([]int, error) {
 	var customerIdArr []int