go_rely_warehouse
/
zyos_go_mq_consume
4
0
Fork 0
Code Issues 0 Pull Requests 0 Releases 0 Wiki Activity
Browse Source

初始化

three
jiaoboxiang 2 years ago
parent
ce25b5a0cd
commit
227404d3ab
10 changed files with 207 additions and 409 deletions
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +11
    -0
      app/cfg/cfg_app.go
  2. +9
    -0
      app/cfg/init_cfg.go
  3. +28
    -0
      app/cfg/init_rabbitmq.go
  4. +3
    -1
      go.mod
  5. +46
    -42
      main.go
  6. +77
    -0
      processor/mq_processor.go
  7. +0
    -175
      test/aces_test.go
  8. +0
    -35
      test/toke_test.go
  9. +33
    -0
      test/work_queue_test.go
  10. +0
    -156
      test/zhimeng_api.go

+ 11
- 0
app/cfg/cfg_app.go View File

@@ -10,7 +10,11 @@ type Config struct {
CurlDebug bool `yaml:"curldebug"`
SrvAddr string `yaml:"srv_addr"`
RedisAddr string `yaml:"redis_addr"`
MqWorkQueueName string `yaml:"mq_work_queue_name"`
RabbitMqAddr string `yaml:"rabbitMq_addr"`
RabbitMqAddrTest string `yaml:"rabbitMq_addr_test"`
DB DBCfg `yaml:"db"`
MQ MQCfg `yaml:"mq"`
Log LogCfg `yaml:"log"`
ArkID ArkIDCfg `yaml:"arkid"`
Admin AdminCfg `yaml:"admin"`
@@ -62,6 +66,13 @@ type DBCfg struct {
Path string `yaml:"path"` //日志文件存放路径
}

type MQCfg struct {
Host string `yaml:"host"`
Port string `yaml:"port"`
User string `yaml:"user"`
Pwd string `yaml:"pwd"`
}

//日志配置结构体
type LogCfg struct {
AppName string `yaml:"app_name" `


+ 9
- 0
app/cfg/init_cfg.go View File

@@ -15,6 +15,7 @@ var (
SrvAddr string
RedisAddr string
DB *DBCfg
MQ *MQCfg
Log *LogCfg
ArkID *ArkIDCfg
Admin *AdminCfg
@@ -22,6 +23,8 @@ var (
WxappletFilepath *WxappletFilepathCfg
Local bool
AppComm *AppCommCfg
MqWorkQueueName string
RabbitMqAddr string
)

//初始化配置文件，将cfg.yml读入到内存
@@ -57,4 +60,10 @@ func InitCfg() {
Official = &conf.Official
WxappletFilepath = &conf.WxappletFilepath
AppComm = &conf.AppComm
MqWorkQueueName = conf.MqWorkQueueName
RabbitMqAddr = conf.RabbitMqAddrTest
if conf.Prd {
RabbitMqAddr = conf.RabbitMqAddr
}
MQ = &conf.MQ
}

+ 28
- 0
app/cfg/init_rabbitmq.go View File

@@ -0,0 +1,28 @@
package cfg

import (
"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
"encoding/json"
"os"
"strings"
"time"
)

func InitMq() {
data, _ := json.Marshal(MQ)
filePutContents("init_rabbit_mq", string(data))
err := rabbit.Init(MQ.Host, MQ.Port, MQ.User, MQ.Pwd)
if err != nil {
filePutContents("init_rabbit_mq", err.Error())
return
}
}

func filePutContents(fileName string, content string) {
fd, _ := os.OpenFile("./tmp/"+fileName+".log", os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
fd_time := time.Now().Format("2006-01-02 15:04:05")
fd_content := strings.Join([]string{"[", fd_time, "] ", content, "\n"}, "")
buf := []byte(fd_content)
fd.Write(buf)
fd.Close()
}

+ 3
- 1
go.mod View File

@@ -3,6 +3,7 @@ module applet
go 1.15

require (
code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git v0.0.3
github.com/afex/hystrix-go v0.0.0-20180502004556-fa1af6a1f4f5
github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751
github.com/boombuler/barcode v1.0.1
@@ -25,7 +26,7 @@ require (
github.com/gorilla/sessions v1.2.1 // indirect
github.com/iGoogle-ink/gopay v1.5.36
github.com/iGoogle-ink/gotil v1.0.20
github.com/json-iterator/go v1.1.10 // indirect
github.com/json-iterator/go v1.1.10
github.com/leodido/go-urn v1.2.1 // indirect
github.com/mailru/easyjson v0.7.7 // indirect
github.com/makiuchi-d/gozxing v0.0.0-20210324052758-57132e828831
@@ -33,6 +34,7 @@ require (
github.com/qiniu/api.v7/v7 v7.8.2
github.com/robfig/cron/v3 v3.0.1
github.com/sony/sonyflake v1.0.0
github.com/streadway/amqp v1.0.0
github.com/swaggo/swag v1.7.0
github.com/syyongx/php2go v0.9.4
github.com/tidwall/gjson v1.7.4


+ 46
- 42
main.go View File

@@ -1,19 +1,17 @@
package main

import (
"context"
"applet/app/cfg"
"applet/app/db"
"applet/app/utils"
"applet/app/utils/logx"
"applet/processor"
"errors"
"fmt"
"log"
"net/http"
"os"
"os/signal"
"strings"
"syscall"
"time"

"applet/app/cfg"
"applet/app/db"
"applet/app/router"
"applet/app/utils"
)

//系统初始化
@@ -21,6 +19,7 @@ func init() {
cfg.InitCfg() //配置初始化
cfg.InitLog() //日志初始化
cfg.InitCache() //缓存初始化
cfg.InitMq() //队列初始化
if cfg.Debug { //判断是否是debug
if err := db.InitDB(cfg.DB); err != nil { //主数据库初始化
panic(err)
@@ -44,45 +43,50 @@ func init() {
// @name MasterID
// @BasePath /
func main() {
// 启动获取所有品牌
//go taoke.GetAllBrand()
r := router.Init() //创建路由
// arkid.Init()
srv := &http.Server{ //设置http服务参数
Addr: cfg.SrvAddr, //指定ip和端口
Handler: r, //指定路由
}
// 读取默认站长的推广位 并写进redis
// master, err := db.UserProfileFindByID(,"1")

// if err != nil {
// panic(err)
// }
if cfg.CurlDebug {
utils.CurlDebug = true
}
//
// if has := cache.SetJson(svc.SysCfgGet(nil, "app_name")+"_default_pid_user", master, 0); has != true {
// panic(errors.New("设置默认pid缓存失败"))
// }
// Initializing the server in a goroutine so that it won't block the graceful shutdown handling below
go func() { //协程启动监听http服务
fmt.Println("Listening and serving HTTP on " + cfg.SrvAddr)
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Fatalf("listen: %s\n", err)
}
}()
// graceful shutdown
//r := router.Init() //创建路由
//srv := &http.Server{ //设置http服务参数
// Addr: cfg.SrvAddr, //指定ip和端口
// Handler: r, //指定路由
//}
//go func() { //协程启动监听队列
// fmt.Println("Listening and serving HTTP on " + cfg.SrvAddr)
// if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
// log.Fatalf("listen: %s\n", err)
// }
//}()
//退出go守护进程
//quit := make(chan os.Signal)
//signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
//<-quit
//log.Println("Shutting down server...")
//ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
//defer cancel()
//if err := srv.Shutdown(ctx); err != nil {
// log.Fatal("Server forced to shutdown:", err)
//}
//log.Println("Server exiting")

//获取需要监听多少个队列
if cfg.RabbitMqAddr == "" {
panic(errors.New("请填写MQ参数"))
}
forever := make(chan bool)
for _, workQueueName := range strings.Split(cfg.MqWorkQueueName, ",") {
if workQueueName == "" {
continue
}
fmt.Printf("监听%s队列", workQueueName)
fmt.Println()
go processor.WorkReceive(workQueueName)
}
//processor.TestWorkSend()
<-forever
quit := make(chan os.Signal)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
<-quit
log.Println("Shutting down server...")
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := srv.Shutdown(ctx); err != nil {
log.Fatal("Server forced to shutdown:", err)
}
log.Println("Server exiting")
_ = logx.Info("Server exiting...")

}

+ 77
- 0
processor/mq_processor.go View File

@@ -0,0 +1,77 @@
package processor

import (
"applet/app/utils"
"applet/app/utils/logx"
"bytes"
"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
"encoding/json"
jsoniter "github.com/json-iterator/go"
"github.com/streadway/amqp"
"log"
"strings"
)

var Json = jsoniter.ConfigCompatibleWithStandardLibrary

type Message struct {
MessageType string `json:"message_type"`
Data int `json:"data"`
}

func WorkReceive(name string) {
ch, err := rabbit.Cfg.Pool.GetChannel()
if err != nil {
logx.Error(err)
return
}
defer ch.Release()
//接收消息时，指定
msgs := ch.Consume(name)
for msg := range msgs {
var message2 Message
jsonStr := string(msg.Body)
jsonStr = strings.Trim(jsonStr, "\"")
jsonStr = strings.ReplaceAll(jsonStr, "\\", "")
utils.Unserialize([]byte(jsonStr), &message2)
switch message2.MessageType {
case "test":
go func(msg *amqp.Delivery) {
log.Printf("recevie1 Received a message: %s", msg.Body)
msg.Ack(true)
}(&msg)
}

}
}

func TestWorkSend() {
// 推入rabbitMq
ch, err := rabbit.Cfg.Pool.GetChannel()
if err != nil {
logx.Error(err)
}
defer ch.Release()
var message struct {
MessageType string `json:"message_type"`
Data int `json:"data"`
}
message.MessageType = "test"
message.Data = 1
for message.Data < 2 {
ch.Publish("test_work_queue_processor", utils.SerializeStr(message), "")
message.Data += 1
//time.Sleep(time.Second * 5)
}
}

// 去除json中的转义字符
func disableEscapeHtml(data interface{}) (string, error) {
bf := bytes.NewBuffer([]byte{})
jsonEncoder := json.NewEncoder(bf)
jsonEncoder.SetEscapeHTML(true)
if err := jsonEncoder.Encode(data); err != nil {
return "", err
}
return bf.String(), nil
}

+ 0
- 175
test/aces_test.go View File

@@ -1,175 +0,0 @@
package test

import (
"applet/app/db"
"applet/app/md"
"applet/app/svc"
"applet/app/utils"
"encoding/base64"
"encoding/json"
"fmt"
"testing"
)

func TestAesCrypt_Encrypt(t *testing.T) {
var aesCrypt = utils.AesCrypt{
Key: []byte("e{&[^<wpliI$AgKs:>Ft(.~g]1eR-]VO"),
Iv: []byte("ZV`7<5X]/2brS@sz"),
}

var text = `{"uid":"82","applyOrder":"821607392542143106","db":{"db_host":"119.23.182.117","db_port":"3306","db_name":"fnuoos_template","db_username":"root","db_password":"Fnuo123com@"}}`
result, err := aesCrypt.Encrypt([]byte(text))
if err != nil {
fmt.Println(err)
return
}

pass64 := base64.StdEncoding.EncodeToString(result)
fmt.Println(pass64)
}

func TestAesCrypt_Decrypt(t *testing.T) {
var aesCrypt = utils.AesCrypt{
Key: []byte("e{&[^<wpliI$AgKs:>Ft(.~g]1eR-]VO"),
Iv: []byte("ZV`7<5X]/2brS@sz"),
}

pass64 := "JD0RXX1YbZPWKeNiVKsq0jQ1Bfnbln3fIMcmJkovU5gUCf329y9ZdqECWe4OKpoOk25/hPNaBH9VwellhIQhpw=="
bytesPass, err := base64.StdEncoding.DecodeString(pass64)
if err != nil {
fmt.Println(err)
return
}

plainText, err := aesCrypt.Decrypt(bytesPass)
if err != nil {
fmt.Println(err)
return
}

fmt.Println(string(plainText))
}

func Test_Vi(t *testing.T) {
fmt.Println("123")
fmt.Println([]byte("ZV`7<5X]/2brS@sz"))
}

func Test_CombineData(t *testing.T) {
data :=
`{
"Uid": 21699,
"Lv": 0,
"NewLv": 0,
"LevelWeight": -1,
"Profit": 7.13,
"SubsidyFee": 0,
"ProfitList": [
{
"cid": "0",
"val": 7.13
},
{
"cid": "1",
"val": 10
},
{
"cid": "19",
"val": 120
},
{
"cid": "20",
"val": 0
},
{
"cid": "21",
"val": 0
}
],
"SubsidyFeeList": null,
"OwnbuyReturnType": 0,
"Diff": 0,
"ParentUser": {
"Uid": 553,
"Lv": 8,
"NewLv": 0,
"LevelWeight": 2,
"Profit": 0,
"SubsidyFee": 0,
"ProfitList": [
{
"cid": "0",
"val": 0
}
],
"SubsidyFeeList": null,
"OwnbuyReturnType": 0,
"Diff": 1,
"ParentUser": {
"Uid": 21699,
"Lv": 0,
"NewLv": 0,
"LevelWeight": -1,
"Profit": 7.13,
"SubsidyFee": 0,
"ProfitList": [
{
"cid": "0",
"val": 7.13
},
{
"cid": "1",
"val": 10
},
{
"cid": "19",
"val": 120
},
{
"cid": "20",
"val": 0
},
{
"cid": "21",
"val": 0
}
],
"SubsidyFeeList": null,
"OwnbuyReturnType": 0,
"Diff": 0,
"ParentUser": {
"Uid": 553,
"Lv": 8,
"NewLv": 0,
"LevelWeight": 2,
"Profit": 0,
"SubsidyFee": 0,
"ProfitList": [
{
"cid": "0",
"val": 0
}
],
"SubsidyFeeList": null,
"OwnbuyReturnType": 0,
"Diff": 1,
"ParentUser": null
}
}
}
}`
bytes := []byte(data)
var lvUser md.LvUser
if err := json.Unmarshal(bytes, &lvUser); err != nil {
fmt.Println(err)
return
}
dataSlice := svc.CombineVirtualCoinRelateData(&lvUser, 249534162504132595, "mall_goods", 0)
for _, item := range dataSlice {
fmt.Printf("%#v\n", item)
}
err := db.DbInsertBatch(db.DBs["123456"], dataSlice)
if err != nil {
fmt.Println(err)
}
}

+ 0
- 35
test/toke_test.go View File

@@ -1,35 +0,0 @@
package test

import (
"applet/app/db"
"applet/app/lib/taoke"
"applet/app/svc"
"fmt"
"testing"
)

func TestBrandList(t *testing.T) {
tkBrandStruct, err := taoke.BrandList("1", "20")
if err != nil {
t.Error(err)
}
for _, brandStruct := range *tkBrandStruct {
fmt.Println(brandStruct)
}

}

func TestSuperCategory(t *testing.T) {
category, err := taoke.SuperCategory()
if err != nil {
t.Error(err)
}
for _, brandStruct := range *category {
fmt.Println(brandStruct)
}
}

func TestAddBrandDB(t *testing.T) {
engine := db.DBs["123456"]
svc.AddBrandDB(engine)
}

+ 33
- 0
test/work_queue_test.go View File

@@ -0,0 +1,33 @@
package test

import (
"applet/app/utils"
"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
"testing"
"time"
)

const WorkQueueName = "test_work_queue"

/**
TODO:: 公平分发模式下的发送端和接收端(一个发动端，多个接受端)
*/
func TestWorkSend(t *testing.T) {
// 推入rabbitMq
ch, err := rabbit.Cfg.Pool.GetChannel()
if err != nil {
t.Error(err)
}
defer ch.Release()
var message struct {
MessageType string `json:"message_type"`
Data int `json:"data"`
}
message.MessageType = "test"
message.Data = 1
for message.Data < 10 {
ch.Publish(WorkQueueName+"_processor", utils.SerializeStr(message), "")
message.Data += 1
time.Sleep(time.Second * 5)
}
}

+ 0
- 156
test/zhimeng_api.go View File

@@ -1,156 +0,0 @@
package test

import (
"fmt"
"github.com/gocolly/colly"
"github.com/gocolly/colly/extensions"
"github.com/tidwall/gjson"
"net/http"
"regexp"
"strings"
)

/*
目前可用接口
[商品查询]https://www.showdoc.com.cn/59349170678610?page_id=339616554551473
[商品详情]https://www.showdoc.com.cn/59349170678610?page_id=339687047645094

*/

// Response is SDK Response
type Response struct {
Msg string `json:"msg"`
Success int `json:"success"`
Data interface{} `json:"data"`
}

func main() {
// // JD
// postData := map[string]string{"keyword": "联想", "p": "1", "size": "10"}
// fmt.Println(postData["time"])
// res, _ := zhimeng.Send("jd", "getgoods", postData)
// fmt.Println(string(res))
// p := Response{}
// json.Unmarshal(res, &p)
// fmt.Println(p)
// // VIP
// postData = map[string]string{"keyword": "联想", "p": "1", "size": "10", "order": "0"}
// fmt.Println(postData["time"])
// res, _ = zhimeng.Send("wph", "seach_goods", postData)
// fmt.Println(string(res))
// p = Response{}
// json.Unmarshal(res, &p)
// fmt.Println(p)
// // PDD
// postData = map[string]string{"keyword": "联想", "p": "1", "size": "10", "sort": "goods_price asc"}
// res, _ = zhimeng.Send("pdd", "getgoods", postData)
// fmt.Println(string(res))
// p = Response{}
// json.Unmarshal(res, &p)
// fmt.Println(p)
for i := 0; i < 1000; i++ {
fmt.Println(i)
scrapPDD()
}
}

func scrapJD() {
c := colly.NewCollector(func(collector *colly.Collector) {
extensions.RandomUserAgent(collector)
})
c.OnResponse(func(r *colly.Response) {
re, _ := regexp.Compile(`[(]//[^\s]*[)]`)
body := r.Body
fmt.Println(string(body))
urls := re.FindAllString(string(body), -1)
fmt.Println(urls)
for _, url := range urls {
url = strip(url, "()")
url = "https:" + url
fmt.Println(url)
}
})
c.Visit("https://wqsitem.jd.com/detail/100008309360_d100008309360_normal.html")
}

func scrapPDD() {
var cookies = []*http.Cookie{}
var mapcookies = make(map[string]string)
url := fmt.Sprintf("https://mobile.yangkeduo.com/goods.html?goods_id=%s", "156632692649")
cs := "api_uid=CiHUKl9DZKpL6QBVK4qWAg==; _nano_fp=Xpdbl0PyX5Pxn0TynT_DTGXbst0kz5cjzGAQDnBR; ua=Mozilla%2F5.0%20(Windows%20NT%2010.0%3B%20Win64%3B%20x64)%20AppleWebKit%2F537.36%20(KHTML%2C%20like%20Gecko)%20Chrome%2F84.0.4147.135%20Safari%2F537.36; webp=1; quick_entrance_click_record=20200824%2C1; PDDAccessToken=XRC6FNX7FRBL6AJRMRBRN4CDG2PZXO3YJZYHFUA4O2PLDAWVYXHA1125821; pdd_user_id=9622705741400; pdd_user_uin=F27EAZ4V5S7EGEVMCJI2P7RFLE_GEXDA; chat_config={'host_whitelist':['.yangkeduo.com','.pinduoduo.com','.10010.com/queen/tencent/pinduoduo-fill.html','.ha.10086.cn/pay/card-sale!toforward.action','wap.ha.10086.cn','m.10010.com']}; pdd_vds=gaLMNqmfGfyYEpyYiZGWopaCicNHbXGWtDNcOZnWLqiDNfLHOXnZaqtCLDiX"
csList := strings.Split(cs, ";")
for _, c := range csList {
s := strings.Trim(c, " ")
sList := strings.SplitN(s, "=", 2)

mapcookies[sList[len(sList)-len(sList)]] = sList[(len(sList) - len(sList) + 1)]

}
fmt.Println(mapcookies)
for key, value := range mapcookies {
if key == "ua" {
continue
}
cookies = append(cookies, &http.Cookie{Name: key, Value: value})
}
c := colly.NewCollector(
colly.UserAgent("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.135 Safari/537.36"),
)

c.OnResponse(func(r *colly.Response) {
re, _ := regexp.Compile(`window.rawData=.*}`)
body := r.Body
fmt.Println(string(body))
result := re.FindString(string(body))
// fmt.Println(result)
result = strings.SplitN(result, "=", 2)[1]
// fmt.Println(result)
value := gjson.Get(result, "store.initDataObj.goods.detailGallery")
// fmt.Println(value)
list := value.Array()
imageList := []string{}
for _, v := range list {
nv := gjson.Get(v.String(), "url")
imageList = append(imageList, nv.String())
}
fmt.Println(imageList)
ck := c.Cookies("https://mobile.yangkeduo.com")
fmt.Println(ck)
cookies = ck
})

c.SetCookies("https://mobile.yangkeduo.com", cookies)

c.Visit(url)
}

func strip(ss string, charss string) string {
s, chars := []rune(ss), []rune(charss)
length := len(s)
max := len(s) - 1
l, r := true, true //标记当左端或者右端找到正常字符后就停止继续寻找
start, end := 0, max
tmpEnd := 0
charset := make(map[rune]bool) //创建字符集，也就是唯一的字符，方便后面判断是否存在
for i := 0; i < len(chars); i++ {
charset[chars[i]] = true
}
for i := 0; i < length; i++ {
if _, exist := charset[s[i]]; l && !exist {
start = i
l = false
}
tmpEnd = max - i
if _, exist := charset[s[tmpEnd]]; r && !exist {
end = tmpEnd
r = false
}
if !l && !r {
break
}
}
if l && r { // 如果左端和右端都没找到正常字符，那么表示该字符串没有正常字符
return ""
}
return string(s[start : end+1])
}

Write Preview
Loading…
Cancel
Save