fix:nsq增加主题

src/config.json

@@ -8,7 +8,7 @@
     "weixinrpc": "127.0.0.1:8083",
     "cacheflag": false,
     "agreement": "http",
-    "webdomain": "https://web-ws.jydev.jianyu360.com",
+    "webdomain": "https://web-wky.jydev.jianyu360.com",
     "redirect": {
         "searchinfo": "/jylab/mainSearch?msg=0",
         "rssset": "/swordfish/newhistorypush?msg=1",
@@ -297,6 +297,7 @@
     "firstBindPop":3,
     "maxBindPop":111111111111115,
     "nsq":"192.168.3.240:4260",
+    "nsq_topic": "jy_event",
     "bindPopRedis":"merge",
     "phoneFilterFlag":true,
     "pcindexCacheTime":7200,
@@ -310,7 +311,7 @@
     "sessionTimeout":168,
     "pcSessionTimeout":168,
     "shareRedisName":"other",
-    "wxWebdomain": "https://web-ws.jydev.jianyu360.com",
+    "wxWebdomain": "https://web-wky.jydev.jianyu360.com",
     "wxOrgUrl": "/page_entbase_mobile/page/invite/inviting.html",
     "pcOrgUrl":"/orgpc/inviting",
     "criticality":2,

src/jfw/front/login.go

@@ -1,7 +1,6 @@
 package front
 
 import (
-	"fmt"
 	"jfw/config"
 	"jfw/public"
 	"log"
@@ -136,8 +135,8 @@ func (l *Login) Login() error {
 						jy.ClearPhoneIdentSession(l.Session())
 						ok, _, userInfo := afterLogin(phone, l.Session())
 						nsqPath, _ := config.Sysconfig["nsq"].(string)
-						fmt.Println(nsqPath)
-						er := jy.Publish(public.Mgo_Log, nsqPath, jy.Jypc_phone_register, _id, jy.Jyweb_node2)
+						nsq_topic, _ := config.Sysconfig["nsq_topic"].(string)
+						er := jy.Publish(public.Mgo_Log, nsqPath, nsq_topic, jy.Jypc_phone_register, _id, jy.Jyweb_node2)
 						if er != nil {
 							log.Println("nsq队列写入失败-->", jy.Jypc_phone_register, _id)
 						}

src/jfw/front/shorturl.go

@@ -553,7 +553,7 @@ func FieldProcessing(obj map[string]interface{}, ssOpenid interface{}, industry,
 		key := fmt.Sprintf("integral_article_%s_%s_%s", article_id, from_userid, userId)
 		if redis.Incr("other", key) == 1 {
 			redis.SetExpire("other", key, 60*60*24)
-			err := jy.Publish(public.Mgo_Log, config.Sysconfig["nsq"].(string), jy.Jyweb_article_open, se.Decode4Hex(from_userid), jy.Jywx_node1)
+			err := jy.Publish(public.Mgo_Log, config.Sysconfig["nsq"].(string), config.Sysconfig["nsq_topic"].(string), jy.Jyweb_article_open, se.Decode4Hex(from_userid), jy.Jywx_node1)
 			if err != nil {
 				log.Println("nsq队列写入失败-->", jy.Jyweb_article_open, se.Decode4Hex(from_userid))
 			}

src/jfw/modules/app/src/app/front/login.go

@@ -204,23 +204,24 @@ func (l *Login) Login() error {
 					if _id != "" {
 						jy.ClearPhoneIdentSession(l.Session())
 						returnSign = afterLogin(data, l.Session(), rid, oid, phoneType, channel, deviceId, true, l.ResponseWriter)
-						addr := config.Sysconfig["nsq"].(string)
+						addr, _ := config.Sysconfig["nsq"].(string)
+						nsq_topic, _ := config.Sysconfig["nsq_topic"].(string)
 						if disWord != "" {
 							redisDis := redis.GetStr("other", "DIS_"+disWord[1:])
 							//邀请新用户 调用队列
 							if redisDisArr := strings.Split(redisDis, "##"); len(redisDisArr) > 1 {
 								fromUserId := redisDisArr[1]
-								err := jy.Publish(public.Mgo_Log, addr, jy.Jywx_subscribe_invite, fromUserId, jy.Jyapp_node1)
+								err := jy.Publish(public.Mgo_Log, addr, nsq_topic, jy.Jywx_subscribe_invite, fromUserId, jy.Jyapp_node1)
 								if err != nil {
 									log.Println("nsq队列写入失败-->", err, jy.Jywx_subscribe_invite, fromUserId)
 								}
-								err = jy.Publish(public.Mgo_Log, addr, jy.Jywx_subscribe_invited, _id, jy.Jyapp_node1)
+								err = jy.Publish(public.Mgo_Log, addr, nsq_topic, jy.Jywx_subscribe_invited, _id, jy.Jyapp_node1)
 								if err != nil {
 									log.Println("nsq队列写入失败-->", err, jy.Jywx_subscribe_invited, _id)
 								}
 							}
 						} else {
-							err := jy.Publish(public.Mgo_Log, addr, jy.Jyapp_phone_register, _id, jy.Jyapp_node1)
+							err := jy.Publish(public.Mgo_Log, addr, nsq_topic, jy.Jyapp_phone_register, _id, jy.Jyapp_node1)
 							if err != nil {
 								log.Println("nsq队列写入失败-->", err, jy.Jyapp_phone_register, _id)
 							}
@@ -457,23 +458,24 @@ func (l *Login) Register() error {
 				}
 				if saveSuccess && saveid != "" {
 					addr := config.Sysconfig["nsq"].(string)
+					topic := config.Sysconfig["nsq_topic"].(string)
 					go func() {
 						if disWord != "" {
 							redisDis := redis.GetStr("other", "DIS_"+disWord[1:])
 							//邀请新用户 调用队列
 							if redisDisArr := strings.Split(redisDis, "##"); len(redisDisArr) > 1 {
 								fromUserId := redisDisArr[1]
-								err := jy.Publish(public.Mgo_Log, addr, jy.Jywx_subscribe_invite, fromUserId, jy.Jyapp_node1)
+								err := jy.Publish(public.Mgo_Log, addr, topic, jy.Jywx_subscribe_invite, fromUserId, jy.Jyapp_node1)
 								if err != nil {
 									log.Println("nsq队列写入失败-->", err, jy.Jywx_subscribe_invite, fromUserId)
 								}
-								err = jy.Publish(public.Mgo_Log, addr, jy.Jywx_subscribe_invited, saveid, jy.Jyapp_node1)
+								err = jy.Publish(public.Mgo_Log, addr, topic, jy.Jywx_subscribe_invited, saveid, jy.Jyapp_node1)
 								if err != nil {
 									log.Println("nsq队列写入失败-->", err, jy.Jywx_subscribe_invited, saveid)
 								}
 							}
 						} else {
-							err := jy.Publish(public.Mgo_Log, addr, jy.Jyapp_phone_register, saveid, jy.Jyapp_node1)
+							err := jy.Publish(public.Mgo_Log, addr, topic, jy.Jyapp_phone_register, saveid, jy.Jyapp_node1)
 							if err != nil {
 								log.Println("nsq队列写入失败-->", err, jy.Jyapp_phone_register, saveid)
 							}
@@ -505,6 +507,7 @@ func (l *Login) Register() error {
 func (l *Login) WxLogin() {
 	// userid := l.GetSession("userId").(string)
 	addr := config.Sysconfig["nsq"].(string)
+	topic := config.Sysconfig["nsq_topic"].(string)
 	disWord := l.GetString("disWord")
 	defer qutil.Catch()
 	returnSign := ""
@@ -606,17 +609,17 @@ func (l *Login) WxLogin() {
 					redisDis := redis.GetStr("other", "DIS_"+disWord[1:])
 					if redisDisArr := strings.Split(redisDis, "##"); len(redisDisArr) > 1 {
 						fromUserId := redisDisArr[1]
-						err := jy.Publish(public.Mgo_Log, addr, jy.Jywx_subscribe_invite, fromUserId, jy.Jyapp_node1)
+						err := jy.Publish(public.Mgo_Log, addr, topic, jy.Jywx_subscribe_invite, fromUserId, jy.Jyapp_node1)
 						if err != nil {
 							log.Println("nsq队列写入失败-->", jy.Jywx_subscribe_invite, fromUserId)
 						}
-						err = jy.Publish(public.Mgo_Log, addr, jy.Jywx_subscribe_invited, _id, jy.Jyapp_node1)
+						err = jy.Publish(public.Mgo_Log, addr, topic, jy.Jywx_subscribe_invited, _id, jy.Jyapp_node1)
 						if err != nil {
 							log.Println("nsq队列写入失败-->", jy.Jywx_subscribe_invited, _id)
 						}
 					}
 				} else {
-					err = jy.Publish(public.Mgo_Log, addr, jy.Jyapp_wx_register, _id, jy.Jyapp_node1)
+					err = jy.Publish(public.Mgo_Log, addr, topic, jy.Jyapp_wx_register, _id, jy.Jyapp_node1)
 					if err != nil {
 						log.Println("nsq队列写入失败-->", jy.Jyapp_wx_register, _id)
 					}

src/jfw/modules/app/src/app/front/shorturl.go

@@ -112,7 +112,7 @@ func (s *Short) Article(stype, id string) error {
 			key := fmt.Sprintf("integral_article_%s_%s_%s", article_id, belongUserId, userId)
 			if redis.Incr("other", key) == 1 {
 				redis.SetExpire("other", key, 60*60*24)
-				err := jy.Publish(public.Mgo_Log, config.Sysconfig["nsq"].(string), jy.Jyweb_article_open, belongUserId, jy.Jyapp_node1)
+				err := jy.Publish(public.Mgo_Log, config.Sysconfig["nsq"].(string), config.Sysconfig["nsq_topic"].(string), jy.Jyweb_article_open, belongUserId, jy.Jyapp_node1)
 				if err != nil {
 					log.Println("nsq队列写入失败-->", jy.Jyweb_article_open, belongUserId)
 				} /* else {

src/jfw/modules/app/src/config.json

@@ -152,6 +152,7 @@
 	},
     "redisSessionLockSize":20,
     "nsq":"192.168.3.240:4260",
+    "nsq_topic": "jy_event",
     "accountMergeOnline":"3333-03-30 00:00:00",
 	"bidSearchOldUserLimit": 1626105600,
     "firstBindPop":3,

src/jfw/modules/common/src/qfw/util/jy/nsq.go

@@ -29,8 +29,7 @@ const (
 
 //e_code 模块
 //e_app 终端
-func Publish(mg MongodbSim, ip, e_code, e_userid, e_app string) error {
-	topicName := "jy_event"
+func Publish(mg MongodbSim, ip, topicName, e_code, e_userid, e_app string) error {
 	if topicName == "" {
 		return errors.New("未找到topicName，请检查tye参数是否正确")
 	}
@@ -51,13 +50,14 @@ func Publish(mg MongodbSim, ip, e_code, e_userid, e_app string) error {
 		fmt.Println("errr2:", err)
 	}
 	producer.Stop()
-	saveLog(mg, *msg)
+	saveLog(mg, topicName, *msg)
 	return err
 }
 
-func saveLog(mg MongodbSim, msg model.Message) {
+func saveLog(mg MongodbSim, topicName string, msg model.Message) {
 	mg.Save("nsq_logs", map[string]interface{}{
 		"body":       msg,
+		"topicName":  topicName,
 		"createtime": time.Now().Unix(),
 		"type":       "producer",
 	})

src/jfw/modules/subscribepay/src/config.json

@@ -108,5 +108,6 @@
 		"中标必听课":true
 	},
 	"smsServiceRpc":"127.0.0.1:932",
-	"nsq":"192.168.3.240:4260"
+	"nsq":"192.168.3.240:4260",
+  "nsq_topic": "jy_event"
 }

src/jfw/modules/subscribepay/src/config/config.go

@@ -72,6 +72,7 @@ type config struct {
 	Product            map[string]bool
 	SmsServiceRpc      string
 	Nsq                string
+	Nsq_Topic          string
 }
 type mgoConf struct {
 	Address           string

src/jfw/modules/subscribepay/src/entity/shareFission.go

@@ -31,11 +31,11 @@ func NewShareFission(userId, shareUid string) *ShareFission {
 //增加权益
 func (this *ShareFission) AddPower(r *http.Request) {
 	if this.ShareUid != "" { //有分享人
-		err := jy.Publish(util.Mgo_log, config.Config.Nsq, jy.Jywx_subscribe_invited, this.UserId, jy.Jywx_node1)
+		err := jy.Publish(util.Mgo_log, config.Config.Nsq, config.Config.Nsq_Topic, jy.Jywx_subscribe_invited, this.UserId, jy.Jywx_node1)
 		if err != nil {
 			log.Println("nsq队列写入失败-->", err, jy.Jywx_subscribe_invited, this.UserId)
 		}
-		err = jy.Publish(util.Mgo_log, config.Config.Nsq, jy.Jywx_subscribe_invite, this.ShareUid, jy.Jywx_node1)
+		err = jy.Publish(util.Mgo_log, config.Config.Nsq, config.Config.Nsq_Topic, jy.Jywx_subscribe_invite, this.ShareUid, jy.Jywx_node1)
 		if err != nil {
 			log.Println("nsq队列写入失败-->", err, jy.Jywx_subscribe_invite, this.ShareUid)
 		}
@@ -48,7 +48,7 @@ func (this *ShareFission) AddPower(r *http.Request) {
 		} else if qu.IsWxBrowser(r) { //微信端绑定手机号
 			e_code = jy.Jywx_subscribe_new
 		}
-		err := jy.Publish(util.Mgo_log, config.Config.Nsq, e_code, this.UserId, jy.Jywx_node1)
+		err := jy.Publish(util.Mgo_log, config.Config.Nsq, config.Config.Nsq_Topic, e_code, this.UserId, jy.Jywx_node1)
 		if err != nil {
 			log.Println("nsq队列写入失败-->", err, e_code, this.UserId)
 		}
@@ -59,11 +59,11 @@ var mobileReg = regexp.MustCompile("(?i)(Android|Mobile|Phone)")
 
 func AddPower(s_m_openid, userId string, r *http.Request) {
 	if shareUid := redis.GetStr(config.Config.ShareRedisName, fmt.Sprintf("rec_%s", s_m_openid)); shareUid != "" {
-		err := jy.Publish(util.Mgo_log, config.Config.Nsq, jy.Jywx_subscribe_invited, userId, jy.Jywx_node1)
+		err := jy.Publish(util.Mgo_log, config.Config.Nsq, config.Config.Nsq_Topic, jy.Jywx_subscribe_invited, userId, jy.Jywx_node1)
 		if err != nil {
 			log.Println("nsq队列写入失败-->", err, jy.Jywx_subscribe_invited, userId)
 		}
-		err = jy.Publish(util.Mgo_log, config.Config.Nsq, jy.Jywx_subscribe_invite, shareUid, jy.Jywx_node1)
+		err = jy.Publish(util.Mgo_log, config.Config.Nsq, config.Config.Nsq_Topic, jy.Jywx_subscribe_invite, shareUid, jy.Jywx_node1)
 		if err != nil {
 			log.Println("nsq队列写入失败-->", err, jy.Jywx_subscribe_invite, shareUid)
 		}