EggPlanet
/
egg_mq_consume
3
0
Fork 0
Code Issues 0 Pull Requests 0 Releases 0 Wiki Activity
Browse Source

update

gim
dengbiao 4 days ago
parent
4d0e336156
commit
dfe178172f
6 changed files with 123 additions and 69 deletions
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +27
    -27
      app/db/db_user.go
  2. +2
    -2
      consume/init.go
  3. +16
    -0
      consume/user_register_for_my_my_fans.go
  4. +21
    -1
      consume/user_register_for_my_my_recommender.go
  5. +20
    -2
      consume/user_register_for_official_consume.go
  6. +37
    -37
      etc/cfg.yml

+ 27
- 27
app/db/db_user.go View File

@@ -34,7 +34,7 @@ func UserisExistByMobile(Db *xorm.Engine, n string) (bool, error) {
// UserInByUIDByLevel is In查询 以及是否是有效用户
func UserInByUIDByLevel(Db *xorm.Engine, ids []int, levelID interface{}) (*[]model.User, error) {
var m []model.User
if err := Db.In("uid", ids).Where("level = ?", levelID).
if err := Db.In("id", ids).Where("level = ?", levelID).
Find(&m); err != nil {
return nil, logx.Warn(err)
}
@@ -44,7 +44,7 @@ func UserInByUIDByLevel(Db *xorm.Engine, ids []int, levelID interface{}) (*[]mod
// UserFindByMobile search user by mobile
func UserFindByMobile(Db *xorm.Engine, mobile string) (*model.User, error) {
var m model.User
if has, err := Db.Where("phone LIKE ? AND delete_at = 0", "%"+mobile).
if has, err := Db.Where("phone LIKE ? ", "%"+mobile).
Get(&m); err != nil || has == false {
return nil, logx.Warn(err)
}
@@ -54,7 +54,7 @@ func UserFindByMobile(Db *xorm.Engine, mobile string) (*model.User, error) {
// UserFindExistByMobile search user by mobile
func UserFindExistByMobile(Db *xorm.Engine, mobile string) (*model.User, bool, error) {
var m model.User
has, err := Db.Where("(phone = ? OR uid = ?) AND delete_at = 0", mobile, mobile).Get(&m)
has, err := Db.Where("(phone = ? OR id = ?) ", mobile, mobile).Get(&m)
if err != nil {
logx.Infof("UserFindExistByMobile err")
return nil, false, logx.Warn(err)
@@ -65,7 +65,7 @@ func UserFindExistByMobile(Db *xorm.Engine, mobile string) (*model.User, bool, e
// UserFindByMobile search user by mobile
func UserFindByMobileAll(Db *xorm.Engine, mobile string) (*model.User, error) {
var m model.User
if has, err := Db.Where("(phone = ? OR uid = ?)", mobile, mobile).
if has, err := Db.Where("(phone = ? OR id = ?)", mobile, mobile).
Get(&m); err != nil || has == false {
return nil, logx.Warn(err)
}
@@ -110,13 +110,13 @@ func UsersFindByNickNameLike(Db *xorm.Engine, nickname string) (*[]model.User, e
func UsersInByIds(Db *xorm.Engine, ids []int, limit, start int) (*[]model.User, error) {
var m []model.User
if limit == 0 && start == 0 {
if err := Db.In("uid", ids).
if err := Db.In("id", ids).
Find(&m); err != nil {
return nil, logx.Warn(err)
}
return &m, nil
}
if err := Db.In("uid", ids).Limit(limit, start).
if err := Db.In("id", ids).Limit(limit, start).
Find(&m); err != nil {
return nil, logx.Warn(err)
}
@@ -127,13 +127,13 @@ func UsersInByIds(Db *xorm.Engine, ids []int, limit, start int) (*[]model.User,
func UsersInByIdsWhereLv(Db *xorm.Engine, ids []int, lv interface{}, limit, start int) (*[]model.User, error) {
var m []model.User
if limit == 0 && start == 0 {
if err := Db.Where("level = ?", lv).In("uid", ids).
if err := Db.Where("level = ?", lv).In("id", ids).
Find(&m); err != nil {
return nil, logx.Warn(err)
}
return &m, nil
}
if err := Db.Where("level = ?", lv).In("uid", ids).Limit(limit, start).
if err := Db.Where("level = ?", lv).In("id", ids).Limit(limit, start).
Find(&m); err != nil {
return nil, logx.Warn(err)
}
@@ -144,13 +144,13 @@ func UsersInByIdsWhereLv(Db *xorm.Engine, ids []int, lv interface{}, limit, star
func UsersInByIdsByAscWhereLv(Db *xorm.Engine, ids []int, lv interface{}, limit, start int, c string) (*[]model.User, error) {
var m []model.User
if limit == 0 && start == 0 {
if err := Db.Where("level = ?", lv).In("uid", ids).Asc(c).
if err := Db.Where("level = ?", lv).In("id", ids).Asc(c).
Find(&m); err != nil {
return nil, logx.Warn(err)
}
return &m, nil
}
if err := Db.Where("level = ?", lv).In("uid", ids).Asc(c).Limit(limit, start).
if err := Db.Where("level = ?", lv).In("id", ids).Asc(c).Limit(limit, start).
Find(&m); err != nil {
return nil, logx.Warn(err)
}
@@ -161,13 +161,13 @@ func UsersInByIdsByAscWhereLv(Db *xorm.Engine, ids []int, lv interface{}, limit,
func UsersInByIdsByDescWhereLv(Db *xorm.Engine, ids []int, lv interface{}, limit, start int, c string) (*[]model.User, error) {
var m []model.User
if limit == 0 && start == 0 {
if err := Db.Where("level = ?", lv).In("uid", ids).Desc(c).
if err := Db.Where("level = ?", lv).In("id", ids).Desc(c).
Find(&m); err != nil {
return nil, logx.Warn(err)
}
return &m, nil
}
if err := Db.Where("level = ?", lv).In("uid", ids).Desc(c).Limit(limit, start).
if err := Db.Where("level = ?", lv).In("id", ids).Desc(c).Limit(limit, start).
Find(&m); err != nil {
return nil, logx.Warn(err)
}
@@ -187,7 +187,7 @@ func UserFindByArkidUserName(Db *xorm.Engine, name string) (*model.User, error)
// UserFindByID is find user byid
func UserFindByID(Db *xorm.Engine, id interface{}) (*model.User, error) {
var m model.User
if _, err := Db.Where("uid = ?", id).
if _, err := Db.Where("id = ?", id).
Get(&m); err != nil {
return nil, logx.Warn(err)
}
@@ -196,7 +196,7 @@ func UserFindByID(Db *xorm.Engine, id interface{}) (*model.User, error) {

func UserFindByIDs(Db *xorm.Engine, uids []int) (*[]model.User, error) {
var m []model.User
if err := Db.In("uid", uids).Find(&m); err != nil {
if err := Db.In("id", uids).Find(&m); err != nil {
return nil, logx.Warn(err)
}
return &m, nil
@@ -206,13 +206,13 @@ func UserFindByIDs(Db *xorm.Engine, uids []int) (*[]model.User, error) {
func UsersInByIdsByDesc(Db *xorm.Engine, ids []int, limit, start int, c string) (*[]model.User, error) {
var m []model.User
if limit == 0 && start == 0 {
if err := Db.In("uid", ids).Desc(c).
if err := Db.In("id", ids).Desc(c).
Find(&m); err != nil {
return nil, logx.Warn(err)
}
return &m, nil
}
if err := Db.In("uid", ids).Desc(c).Limit(limit, start).
if err := Db.In("id", ids).Desc(c).Limit(limit, start).
Find(&m); err != nil {
return nil, logx.Warn(err)
}
@@ -223,13 +223,13 @@ func UsersInByIdsByDesc(Db *xorm.Engine, ids []int, limit, start int, c string)
func UsersInByIdsByAsc(Db *xorm.Engine, ids []int, limit, start int, c string) (*[]model.User, error) {
var m []model.User
if limit == 0 && start == 0 {
if err := Db.In("uid", ids).Asc(c).
if err := Db.In("id", ids).Asc(c).
Find(&m); err != nil {
return nil, logx.Warn(err)
}
return &m, nil
}
if err := Db.In("uid", ids).Asc(c).Limit(limit, start).
if err := Db.In("id", ids).Asc(c).Limit(limit, start).
Find(&m); err != nil {
return nil, logx.Warn(err)
}
@@ -248,7 +248,7 @@ func UserInsert(Db *xorm.Engine, user *model.User) (int64, error) {
// UserIsExistByMobile is mobile exist
func UserIsExistByMobile(Db *xorm.Engine, mobile string) (bool, error) {
//fmt.Println(mobile)
has, err := Db.Where("phone = ? OR uid = ?", mobile, mobile).Exist(&model.User{})
has, err := Db.Where("phone = ? OR id = ?", mobile, mobile).Exist(&model.User{})
fmt.Println(has, mobile)
if err != nil {
return false, err
@@ -258,7 +258,7 @@ func UserIsExistByMobile(Db *xorm.Engine, mobile string) (bool, error) {

// UserIsExistByID is mobile exist by id
func UserIsExistByID(Db *xorm.Engine, id string) (bool, error) {
has, err := Db.Where("uid = ?", id).Exist(&model.User{})
has, err := Db.Where("id = ?", id).Exist(&model.User{})
if err != nil {
return false, err
}
@@ -266,15 +266,15 @@ func UserIsExistByID(Db *xorm.Engine, id string) (bool, error) {
}

// UserUpdate is update user
func UserUpdate(Db *xorm.Engine, uid interface{}, user *model.User, forceColums ...string) (int64, error) {
func UserUpdate(Db *xorm.Engine, id interface{}, user *model.User, forceColums ...string) (int64, error) {
var (
affected int64
err error
)
if forceColums != nil {
affected, err = Db.Where("uid=?", uid).Cols(forceColums...).Update(user)
affected, err = Db.Where("id=?", id).Cols(forceColums...).Update(user)
} else {
affected, err = Db.Where("uid=?", uid).Update(user)
affected, err = Db.Where("id=?", id).Update(user)
}
if err != nil {
return 0, err
@@ -287,12 +287,12 @@ func UpdateUserFinValid() {
}

// UserDelete is delete user
func UserDelete(Db *xorm.Engine, uid interface{}) (int64, error) {
return Db.Where("uid = ?", uid).Delete(model.User{})
func UserDelete(Db *xorm.Engine, id interface{}) (int64, error) {
return Db.Where("id = ?", id).Delete(model.User{})
}

func UserDeleteWithSess(sess *xorm.Session, uid interface{}) (int64, error) {
return sess.Where("uid = ?", uid).Delete(model.User{})
func UserDeleteWithSess(sess *xorm.Session, id interface{}) (int64, error) {
return sess.Where("id = ?", id).Delete(model.User{})
}

func UserFindByLevel(eg *xorm.Engine, level int) []model.User {


+ 2
- 2
consume/init.go View File

@@ -17,9 +17,9 @@ func Init() {

// 增加消费任务队列
func initConsumes() {
jobs[consumeMd.UserRegisterConsumeForMyFansFunName] = UserRegisterConsumeForMyFans
//jobs[consumeMd.UserRegisterConsumeForMyFansFunName] = UserRegisterConsumeForMyFans
jobs[consumeMd.UserRegisterConsumeForMyRecommenderFunName] = UserRegisterConsumeForMyRecommender
jobs[consumeMd.UserRegisterConsumeForOfficialFunName] = UserRegisterConsumeForOfficial
//jobs[consumeMd.UserRegisterConsumeForOfficialFunName] = UserRegisterConsumeForOfficial
}

func Run() {


+ 16
- 0
consume/user_register_for_my_my_fans.go View File

@@ -8,6 +8,7 @@ import (
"applet/app/db/model"
md2 "applet/app/md"
utils2 "applet/app/utils"
"applet/app/utils/cache"
"applet/app/utils/logx"
utils "applet/app/utils/rpc"
"applet/consume/md"
@@ -21,6 +22,8 @@ import (
"time"
)

const UserRegisterConsumeForMyFansKey = "UserRegisterConsumeForMyFans" // 用户访问记录 uid index

func UserRegisterConsumeForMyFans(queue md.MqQueue) {
fmt.Println(">>>>>>>>>>>>UserRegisterConsumeForMyFans>>>>>>>>>>>>")
ch, err := rabbit.Cfg.Pool.GetChannel()
@@ -72,6 +75,18 @@ func handleUserRegisterConsumeForMyFans(msgData []byte) error {
if err != nil {
return err
}
//TODO::判断是否已操作过
get, err1 := cache.GetBit(UserRegisterConsumeForMyFansKey, msg.Uid)
if err1 != nil || get == 0 {
_, err1 = cache.SetBit(UserRegisterConsumeForMyFansKey, msg.Uid, 1)
if err1 != nil {
return err1
}
}
if get == 1 {
return nil
}

//2、查找用户对应im系统中的数据
gimUser, err := db2.UserGetOneByParams(db.DbIm, map[string]interface{}{
"key": "phone_number",
@@ -81,6 +96,7 @@ func handleUserRegisterConsumeForMyFans(msgData []byte) error {
return err
}
if gimUser == nil {
fmt.Println("===========================", msg.Phone)
user, err := db.UserFindByMobile(db.Db, strconv.FormatInt(msg.Phone, 10))
if err != nil {
return err


+ 21
- 1
consume/user_register_for_my_my_recommender.go View File

@@ -8,6 +8,7 @@ import (
model2 "applet/app/db/model"
md2 "applet/app/md"
utils2 "applet/app/utils"
"applet/app/utils/cache"
"applet/app/utils/logx"
utils "applet/app/utils/rpc"
"applet/consume/md"
@@ -21,6 +22,8 @@ import (
"time"
)

const UserRegisterConsumeForMyRecommenderKey = "UserRegisterConsumeForMyRecommender" // 用户访问记录 uid index

func UserRegisterConsumeForMyRecommender(queue md.MqQueue) {
fmt.Println(">>>>>>>>>>>>UserRegisterConsumeForMyRecommender>>>>>>>>>>>>")
ch, err := rabbit.Cfg.Pool.GetChannel()
@@ -71,6 +74,19 @@ func handleUserRegisterConsumeForMyRecommender(msgData []byte) error {
if err != nil {
return err
}

//TODO::判断是否已操作过
get, err1 := cache.GetBit(UserRegisterConsumeForMyRecommenderKey, msg.Uid)
if err1 != nil || get == 0 {
_, err1 = cache.SetBit(UserRegisterConsumeForMyRecommenderKey, msg.Uid, 1)
if err1 != nil {
return err1
}
}
if get == 1 {
return nil
}

//2、查找用户对应im系统中的数据
gimUser, err := db2.UserGetOneByParams(db.DbIm, map[string]interface{}{
"key": "phone_number",
@@ -81,6 +97,7 @@ func handleUserRegisterConsumeForMyRecommender(msgData []byte) error {
}

if gimUser == nil {
fmt.Println("===========================", msg.Phone)
user, err := db.UserFindByMobile(db.Db, strconv.FormatInt(msg.Phone, 10))
if err != nil {
return err
@@ -172,7 +189,8 @@ func handleUserRegisterConsumeForMyRecommender(msgData []byte) error {
return err
}
tempFansGroup, err := db.ImGroupGetOneByParams(db.Db, 2, map[string]interface{}{
"group_id": resp.GroupId,
"key": "group_id",
"value": resp.GroupId,
})
if err != nil {
return err
@@ -192,6 +210,8 @@ func handleUserRegisterConsumeForMyRecommender(msgData []byte) error {
}

//加入群
fmt.Println(fmt.Sprintf("=======GroupId======%d==============", fansGroup.GroupId))
fmt.Println(fmt.Sprintf("=======gimUserId======%d==============", gimUser.Id))
_, err = utils.GetLogicExtClient(cfg.ImLogicRpc.URL, cfg.ImLogicRpc.PORT).AddGroupMembers(
utils.GetCtx("", strconv.FormatInt(userGroup.UserId, 10), ""),
&pb.AddGroupMembersReq{


+ 20
- 2
consume/user_register_for_official_consume.go View File

@@ -8,6 +8,7 @@ import (
"applet/app/db/model"
md2 "applet/app/md"
utils2 "applet/app/utils"
"applet/app/utils/cache"
"applet/app/utils/logx"
utils "applet/app/utils/rpc"
"applet/consume/md"
@@ -21,6 +22,8 @@ import (
"time"
)

const UserRegisterConsumeForOfficialKey = "UserRegisterConsumeForOfficial" // 用户访问记录 uid index

func UserRegisterConsumeForOfficial(queue md.MqQueue) {
fmt.Println(">>>>>>>>>>>>UserRegisterConsumeForOfficial>>>>>>>>>>>>")
ch, err := rabbit.Cfg.Pool.GetChannel()
@@ -71,6 +74,18 @@ func handleUserRegisterConsumeForOfficial(msgData []byte) error {
if err != nil {
return err
}
//TODO::判断是否已操作过
get, err1 := cache.GetBit(UserRegisterConsumeForOfficialKey, msg.Uid)
if err1 != nil || get == 0 {
_, err1 = cache.SetBit(UserRegisterConsumeForOfficialKey, msg.Uid, 1)
if err1 != nil {
return err1
}
}
if get == 1 {
return nil
}

//2、查找用户对应im系统中的数据
gimUser, err := gim.UserGetOneByParams(db.DbIm, map[string]interface{}{
"key": "phone_number",
@@ -151,7 +166,7 @@ func handleUserRegisterConsumeForOfficial(msgData []byte) error {
now := time.Now()
_, err = db.ImGroupInsert(db.Db, &model.ImGroup{
Kind: 1,
Uid: int(msg.Uid),
Uid: int(1),
GroupId: int(resp.GroupId),
IsFull: 0,
Name: groupName,
@@ -174,12 +189,15 @@ func handleUserRegisterConsumeForOfficial(msgData []byte) error {
}

//加入群
_, err = utils.GetLogicExtClient(cfg.ImLogicRpc.URL, cfg.ImLogicRpc.PORT).AddGroupMembers(
_, err11 := utils.GetLogicExtClient(cfg.ImLogicRpc.URL, cfg.ImLogicRpc.PORT).AddGroupMembers(
utils.GetCtx("", strconv.FormatInt(userGroup.UserId, 10), ""),
&pb.AddGroupMembersReq{
GroupId: int64(officialGroup.GroupId),
UserIds: []int64{gimUser.Id},
})
if err11 != nil {
return err11
}
}
return nil
}

+ 37
- 37
etc/cfg.yml View File

@@ -8,26 +8,26 @@ local: true
redis_addr: '127.0.0.1:6379'
redis_password: ''

db:
host: '119.23.182.117:3306'
name: 'egg'
user: 'root'
psw: 'Fnuo123com@'
show_log: true
max_lifetime: 30
max_open_conns: 100
max_idle_conns: 100
path: 'tmp/%s.log'
#db:
# host: 'advertisement-db.rwlb.rds.aliyuncs.com:3306'
# host: '119.23.182.117:3306'
# name: 'egg'
# user: 'canal'
# psw: 'DengBiao@1997'
# user: 'root'
# psw: 'Fnuo123com@'
# show_log: true
# max_lifetime: 30
# max_open_conns: 100
# max_idle_conns: 100
# path: 'tmp/%s.log'
db:
host: 'advertisement-db.rwlb.rds.aliyuncs.com:3306'
name: 'egg'
user: 'canal'
psw: 'Fnuo123@com'
show_log: true
max_lifetime: 30
max_open_conns: 100
max_idle_conns: 100
path: 'tmp/%s.log'

db_back:
host: '119.23.182.117:3306'
@@ -45,28 +45,28 @@ db_back:
# show_log: true
# path: 'tmp/%s.log'

im_db:
host: '119.23.182.117:3306'
name: 'egg-im'
user: 'root'
psw: 'Fnuo123com@'
show_log: true
max_lifetime: 30
max_open_conns: 100
max_idle_conns: 100
path: 'tmp/%s.log'

#im_db:
# host: 'advertisement-db.rwlb.rds.aliyuncs.com:3306'
# host: '119.23.182.117:3306'
# name: 'egg-im'
# user: 'canal'
# psw: 'DengBiao@1997'
# user: 'root'
# psw: 'Fnuo123com@'
# show_log: true
# max_lifetime: 30
# max_open_conns: 100
# max_idle_conns: 100
# path: 'tmp/%s.log'

im_db:
host: 'advertisement-db.rwlb.rds.aliyuncs.com:3306'
name: 'egg-im'
user: 'canal'
psw: 'Fnuo123@com'
show_log: true
max_lifetime: 30
max_open_conns: 100
max_idle_conns: 100
path: 'tmp/%s.log'

# 日志
log:
app_name: 'applet'
@@ -81,16 +81,16 @@ log:
file_name: 'debug.log'

# 连接RabbitMq
mq:
host: '120.77.153.180'
port: '5672'
user: 'guest'
pwd: 'guest'
#mq:
# host: '116.62.62.35'
# host: '120.77.153.180'
# port: '5672'
# user: 'zhios'
# pwd: 'ZHIoscnfnuo123'
# user: 'guest'
# pwd: 'guest'
mq:
host: '116.62.62.35'
port: '5672'
user: 'zhios'
pwd: 'ZHIoscnfnuo123'

#es:
# url: 'http://120.55.48.175:9200'
@@ -103,9 +103,9 @@ es:
pwd: 'se4BxqYyGhHyPWPA3w8Q'

im_business_rpc:
url: im-rpc-business.izhim.com
url: im-rpc-business.ddstaros.cn
port: 1005

im_logic_rpc:
url: im-rpc-logic.izhim.com
url: im-rpc-logic.ddstaros.cn
port: 1008

Write Preview
Loading…
Cancel
Save