Skip to content

孙龙 / golang_asynctask

Go to a project
Commit afb42837 by 孙龙
Options

up

parent 52c34ec4
Inline Side-by-side
Showing with 516 additions and 10 deletions
README.md
......@@ -30,9 +30,17 @@ go run ./cmd/crm/recv_comuser/main.go -config=./config/dev/ -logdir=./logs/
go run ./cmd/crm/sync_comuser_to_erp/main.go -config=./config/dev/ -logdir=./logs/
go run ./cmd/crm/sendmsg/main.go -config=./config/dev/ -logdir=./logs/
go run ./cmd/crm/cron_sendmsg/main.go -config=./config/dev/ -logdir=./logs/
#备忘录提醒
go run ./cmd/crm/cron_crmschedulememo/main.go -config=./config/dev/ -logdir=./logs/
#小组池释放任务
#若已下单，任务标记为已处理，不释放；请求erp查询是否下单
go run ./cmd/crm/cron_crm_todotask/main.go -config=./config/dev/ -logdir=./logs/ --action=Corder
#若60天没有关闭任务，则任务状态为已超时，该客户释放回小组池
go run ./cmd/crm/cron_crm_todotask/main.go -config=./config/dev/ -logdir=./logs/ --action=TimeoutRelease
#任务超时前3天于上午9：00邮件提醒，邮件抄送上级组长、主管
go run ./cmd/crm/cron_crm_todotask/main.go -config=./config/dev/ -logdir=./logs/ --action=RemindTimeout
##项目的目录结构
├── app 项目
......
app/common/config/cron_crm_todotask/mysql.go 0 → 100644
package cron_crm_todotask
import (
"github.com/ichunt2019/cfg/lib"
)
type BaseDatabase struct {
DataSourceName string
MaxIdleCons int
MaxOpenCons int
Prefix string
}
type GroupDatabase struct {
DataSourceName []string
MaxIdleCons int
MaxOpenCons int
Prefix string
}
//单、主数据 master
func BuildDatabaseList() (DatabaseList map[string]BaseDatabase) {
return map[string]BaseDatabase{
"crm": {
DataSourceName: lib.Instance("db").GetString("crm.data_source_name"),
Prefix: lib.Instance("db").GetString("crm.table_prefix"),
MaxIdleCons:lib.Instance("db").GetInt("crm.max_idle_conn"),
MaxOpenCons:lib.Instance("db").GetInt("crm.max_open_conn"),
},
//"cms": {
// DataSourceName: lib.Instance("db").GetString("cms.data_source_name"),
// Prefix: lib.Instance("db").GetString("cms.table_prefix"),
// MaxIdleCons:lib.Instance("db").GetInt("cms.max_idle_conn"),
// MaxOpenCons:lib.Instance("db").GetInt("cms.max_open_conn"),
//},
}
}
//主从数据 master slave
func BuildGroupDatabaseList() (DatabaseList map[string]GroupDatabase) {
return map[string]GroupDatabase{
//"liexin": {
// DataSourceName:lib.Instance("db").GetStringSlice("liexin.data_source_name"),
// Prefix: lib.Instance("db").GetString("liexin.table_prefix"),
// MaxIdleCons:lib.Instance("db").GetInt("liexin.max_idle_conn"),
// MaxOpenCons:lib.Instance("db").GetInt("liexin.max_open_conn"),
//},
}
}
app/common/config/cron_crm_todotask/redis.go 0 → 100644
package cron_crm_todotask
import (
"github.com/ichunt2019/cfg/lib"
"time"
)
type RedisGroupDatabase struct {
MasterHost string
Password string //master密码
SlaveHost []string
ReadonlyPassword string //从服务器密码
DialTimeout time.Duration //连接超时
MinIdleConns int //空间链接
ReadTimeout time.Duration //读超时
WriteTimeout time.Duration //写超时
}
//多数据库配置
func BuildRedisGroupConfigs() ( map[string]RedisGroupDatabase) {
//fmt.Println("6666666666")
//fmt.Println(lib.Instance("redis").GetStringSlice("api.slave.host"))
return map[string]RedisGroupDatabase{
"api": {
MasterHost: lib.Instance("redis").GetString("api.master.host"),
Password:lib.Instance("redis").GetString("api.master.password"),
SlaveHost: lib.Instance("redis").GetStringSlice("api.slave.host"),
ReadonlyPassword:lib.Instance("redis").GetString("api.slave.password"),
DialTimeout: time.Duration(lib.Instance("redis").GetInt("api.dial_timeout"))*time.Second,
ReadTimeout: time.Duration(lib.Instance("redis").GetInt("api.read_timeout"))*time.Second,
WriteTimeout: time.Duration(lib.Instance("redis").GetInt("api.write_timeout"))*time.Second,
MinIdleConns:lib.Instance("redis").GetInt("api.min_idle_conns"),
},
"sku": {
MasterHost: lib.Instance("redis").GetString("sku.master.host"),
Password:lib.Instance("redis").GetString("sku.master.password"),
SlaveHost: lib.Instance("redis").GetStringSlice("sku.slave.host"),
ReadonlyPassword:lib.Instance("redis").GetString("sku.slave.password"),
DialTimeout: time.Duration(lib.Instance("redis").GetInt("sku.dial_timeout"))*time.Second,
ReadTimeout: time.Duration(lib.Instance("redis").GetInt("sku.read_timeout"))*time.Second,
WriteTimeout: time.Duration(lib.Instance("redis").GetInt("sku.write_timeout"))*time.Second,
MinIdleConns:lib.Instance("redis").GetInt("sku.min_idle_conns"),
},
}
}
app/dao/crm/cron_crm_todotask/dao.go 0 → 100644
package cron_crm_todotask
import (
"fmt"
"golang-asynctask/app/common/config/cron_crm_schedulememo"
cron_crm_todotask "golang-asynctask/app/common/config/cron_crm_todotask"
"sync"
"time"
"xorm.io/xorm"
"github.com/go-redis/redis/v7"
_ "github.com/go-sql-driver/mysql"
"github.com/ichunt2019/cfg/lib"
redisPool "github.com/ichunt2019/go-redis-pool"
)
var (
once sync.Once
Dao *dao
)
type dao struct {
db map[string]*xorm.Engine //非主从mysql数据库 连接池
dbGroup map[string]*xorm.EngineGroup //mysql主从 连接池
redisGroup map[string]*redisPool.Pool //redis 主从 连接池
}
//获取db实例
func (self *dao) GetDb(databases string) *xorm.Engine {
return self.db[databases]
}
//获取主从db实例
//获取主从db实例
//func (self *dao) GetDbGroup(databases string) *xorm.EngineGroup {
// return self.dbGroup[databases]
//}
//获取主从db实例
//func (self *dao) GetRedisDbGroup(databases string) *redisPool.Pool {
// return self.redisGroup[databases]
//}
func mysqlSetup(d *dao) *dao {
var (
err error
)
DatabaseList := cron_crm_todotask.BuildDatabaseList()
GroupDatabaseList := cron_crm_todotask.BuildGroupDatabaseList()
if len(DatabaseList) > 0 {
for conName, db := range DatabaseList {
d.db[conName], err = xorm.NewEngine("mysql", db.DataSourceName)
if err != nil {
panic(err)
}
err = d.db[conName].Ping()
if err != nil {
panic(fmt.Sprintf("数据库连接失败 ，数据库连接信息：%s , 错误信息：%s",db.DataSourceName,err.Error()))
}
//日志打印SQL
ShowSql := lib.Instance("db").GetBool("xorm.ShowSQL")
d.db[conName].ShowSQL(ShowSql)
//设置连接池的空闲数大小
d.db[conName].SetMaxIdleConns(db.MaxIdleCons)
//设置最大打开连接数
d.db[conName].SetMaxOpenConns(db.MaxOpenCons)
}
}
if len(GroupDatabaseList) > 0 {
for conName, db := range GroupDatabaseList {
d.dbGroup[conName], err = xorm.NewEngineGroup("mysql", db.DataSourceName)
if err != nil {
panic(err)
}
//日志打印SQL
ShowSql := lib.Instance("db").GetBool("xorm.ShowSQL")
d.dbGroup[conName].ShowSQL(ShowSql)
//设置连接池的空闲数大小
d.dbGroup[conName].SetMaxIdleConns(db.MaxIdleCons)
//设置最大打开连接数
d.dbGroup[conName].SetMaxOpenConns(db.MaxOpenCons)
}
}
return d
}
func redisSetup(d *dao) *dao {
var err error
redisGroupList := cron_crm_schedulememo.BuildRedisGroupConfigs()
//fmt.Println(redisGroupList)
for redisServerName, redisInfo := range redisGroupList {
d.redisGroup[redisServerName], err = redisPool.NewHA(&redisPool.HAConfig{
Master: redisInfo.MasterHost,
Slaves: redisInfo.SlaveHost,
Password: redisInfo.Password,
ReadonlyPassword: redisInfo.ReadonlyPassword,
Options: &redis.Options{
DialTimeout: redisInfo.DialTimeout, //连接超时
MinIdleConns: redisInfo.MinIdleConns, //空闲链接数
ReadTimeout: redisInfo.ReadTimeout,
WriteTimeout: redisInfo.WriteTimeout,
},
AutoEjectHost: true,//是否弹出故障主机
ServerFailureLimit: 3,//达到失败次数时弹出
ServerRetryTimeout: 5 * time.Second,//在“ServerRetryTimeout”之后重试弹出的主机`
MinServerNum: 0,//保留min服务器 针对从服务器
})
if err != nil {
panic(err)
}
}
return d
}
func Init() {
Dao = &dao{}
once.Do(func() {
//单、主数据 master
Dao.db = make(map[string]*xorm.Engine, 0)
////主从数据 master slave
//Dao.dbGroup = make(map[string]*xorm.EngineGroup, 0)
////redis连接池 支持主从 master slave
//Dao.redisGroup = make(map[string]*redisPool.Pool, 0)
Dao = mysqlSetup(Dao)
//Dao = redisSetup(Dao)
})
}
app/service/crm/cron_crm_todotask/service.go 0 → 100644
package cron_crm_todotask
import (
"fmt"
"github.com/tidwall/gjson"
"golang-asynctask/app/dao/crm/cron_crm_todotask"
comfunc "golang-asynctask/util/lib"
xlog "github.com/ichunt2019/lxLog/log"
cfg "github.com/ichunt2019/cfg/lib"
"net/http"
"net/url"
"strconv"
"sync"
"time"
)
//若已下单，任务标记为已处理，不释放；
func Corder(){
sql := "select id,com_name,create_time,end_time from lie_schedule where status = 1"
res,err := cron_crm_todotask.Dao.GetDb("crm").QueryString(sql);
if err != nil{
xlog.Instance("cron_crm_todotask").Error("获取自住领取的任务列表失败,err:%s",err);
}
for _,item := range res{
if item["com_name"] != ""{
createTime ,err := strconv.ParseInt(item["create_time"],10,64)
if err != nil{
xlog.Instance("cron_crm_todotask").Error("创建时间转换失败，数据：%s ,err: %s",item,err);
}
bizDate := time.Unix(createTime,0).Format("20060102")
isCorder := checkCustomerCOrder(item["com_name"],bizDate)
if !isCorder {
continue
}
id,err := strconv.ParseInt(item["id"],10,64)
if err != nil{
xlog.Instance("cron_crm_todotask").Error("创建时间转换失败，数据：%s ,err: %s",item,err);
continue
}
//客户已下单 修改任务为已处理
err = updateScheduleByStatus(id,2)
if err != nil{
xlog.Instance("cron_crm_todotask").Error("创建时间转换失败，数据：%s ,err: %s",item,err);
continue
}
}
}
}
//若过期的任务，该客户释放回小组池
func TimeoutRelease(){
sql := "SELECT id,end_time,status,handler_id FROM `lie_schedule` where status = 1 and end_time <= unix_timestamp(now()) "
res,err := cron_crm_todotask.Dao.GetDb("crm").QueryString(sql);
if err != nil{
xlog.Instance("cron_crm_todotask").Error("若过期的任务,err:%s",err);
}
//fmt.Println(res)
for _,v := range res{
isok := releaseUser(v["id"],v["handler_id"])
if !isok{
continue
}
id , err := strconv.ParseInt(v["id"], 10, 64)
if err != nil{
continue;
}
sql := "UPDATE lie_schedule SET status = -1 where id = ?"
_,err = cron_crm_todotask.Dao.GetDb("crm").Exec(sql,id)
if err != nil{
continue;
}
}
}
/*
释放用户
*/
func releaseUser(id string,handler_id string) bool{
ichuntapiDomain := cfg.Instance("config").GetString("crmDomain")
push_url := ichuntapiDomain+"/api/releaseUser"
urlParams := url.Values{}
header := http.Header{}
header.Set("api-key",cfg.Instance("config").GetString("apiKey"))
urlParams.Add("handler_id",handler_id)
urlParams.Add("id",id)
response,returnData,err := comfunc.HttpPOST(push_url,urlParams,0,header,"")
//fmt.Println(string(returnData))
if err == nil && response.StatusCode == 200{
err_code := gjson.Parse(string(returnData)).Get("errcode").Int()
msg := gjson.Parse(string(returnData)).Get("errmsg").String()
if err_code != 0{
xlog.Instance("cron_crm_todotask").Error("任务超时发送邮件提醒失败,参数：[%s,%s],错误:[%s]",id,handler_id,msg);
return false
}
return true
}
return false
}
//任务超时前3天于上午9：00邮件提醒，邮件抄送上级组长、主管
func RemindTimeout(){
var (
wg sync.WaitGroup
err error
)
currentTime := time.Now().Format("2006-01-02")
fmt.Println(currentTime)
start_time_str := currentTime + " 00:00:00"
end_time_str := currentTime + " 23:59:59"
start_time,_ :=time.Parse("2006-01-02 15:04:05",start_time_str);
end_time,_ :=time.Parse("2006-01-02 15:04:05",end_time_str);
sql := "select id,handler_id,handler_name from lie_schedule where status = 1 and end_time >=? and end_time <=?"
reult,err := cron_crm_todotask.Dao.GetDb("crm").SQL(sql,start_time.Unix()+1*24*3600,end_time.Unix()+3*24*3600).QueryString()
if err != nil{
xlog.Instance("cron_crm_todotask").Error("err:%s",err);
return
}
//fmt.Println(reult)
//
//fmt.Println(len(reult))
for _,v :=range reult{
wg.Add(1)
go func(wg *sync.WaitGroup,sale_id string , id string) {
_ = seneEmailByRemindTimeout(wg,sale_id,id)
}(&wg,v["handler_id"],v["id"])
}
wg.Wait()
}
/*
检查客户是否下单
*/
func checkCustomerCOrder(customerName string,bizDate string) (bool){
ichuntapiDomain := cfg.Instance("config").GetString("ichuntapiDomain")
push_url := ichuntapiDomain+"/Server/Consume/getPurOrderStatusByCustomer"
urlParams := url.Values{}
header := http.Header{}
urlParams.Add("customerName",customerName)
urlParams.Add("bizDate",bizDate)
response,returnData,err := comfunc.HttpPOST(push_url,urlParams,0,header,"")
if err == nil && response.StatusCode == 200{
err_code := gjson.Parse(string(returnData)).Get("err_code").Int()
msg := gjson.Parse(string(returnData)).Get("err_msg").String()
if err_code != 0{
xlog.Instance("cron_crm_todotask").Error("检查客户是否下单,参数：[%s,%s],错误:[%s]",customerName,bizDate,msg);
return false
}
return true
}
return false
}
/*
任务超时发送邮件提醒
*/
func seneEmailByRemindTimeout(wg *sync.WaitGroup,sale_id string , id string) bool{
defer wg.Done()
ichuntapiDomain := cfg.Instance("config").GetString("crmDomain")
push_url := ichuntapiDomain+"/api/scheduleTimeOutRemind"
urlParams := url.Values{}
header := http.Header{}
header.Set("api-key",cfg.Instance("config").GetString("apiKey"))
urlParams.Add("sale_id",sale_id)
urlParams.Add("id",id)
response,returnData,err := comfunc.HttpPOST(push_url,urlParams,0,header,"")
//fmt.Println(string(returnData))
if err == nil && response.StatusCode == 200{
err_code := gjson.Parse(string(returnData)).Get("err_code").Int()
msg := gjson.Parse(string(returnData)).Get("err_msg").String()
if err_code != 0{
xlog.Instance("cron_crm_todotask").Error("任务超时发送邮件提醒失败,参数：[%s,%s],错误:[%s]",sale_id,id,msg);
return false
}
return true
}
return false
}
/*
修改待办信息表 任务为已处理
*/
func updateScheduleByStatus(id int64,status int) (err error){
sql := "update lie_schedule set status = ? where id = ?"
_,err =cron_crm_todotask.Dao.GetDb("crm").Exec(sql,status,id)
return
}
\ No newline at end of file
boot/cron_crm_todotask/boot.go 0 → 100644
package boot
import (
"github.com/ichunt2019/cfg/lib"
cron_crm_todotask "golang-asynctask/app/dao/crm/cron_crm_todotask"
xlog "github.com/ichunt2019/lxLog/log"
)
func Init(configPath string,logPath string)(err error){
err = lib.Init(configPath)
if err != nil{
panic(err)
}
xlog.Init(logPath,"cron_crm_todotask")
//初始化数据库
cron_crm_todotask.Init()
return
}
cmd/crm/cron_crm_todotask/README.md 0 → 100644
#crm
任务脚本：
若已下单，任务标记为已处理，不释放；
若60天没有关闭任务，则任务状态为已超时，该客户释放回小组池，任务超时前3天于上午9：00邮件提醒，邮件抄送上级组长、主管
\ No newline at end of file
cmd/crm/cron_crm_todotask/main.go 0 → 100644
package main
import (
"flag"
xlog "github.com/ichunt2019/lxLog/log"
"golang-asynctask/app/service/crm/cron_crm_todotask"
boot "golang-asynctask/boot/cron_crm_todotask"
)
var (
configPath string
logPath string
action string
)
//crm释放任务 小组池
func main(){
defer func(){
xlog.Instance("cron_crm_todotask").Close()
}()
flag.StringVar(&configPath, "config", "./config/dev/", "配置文件")
flag.StringVar(&logPath, "logdir", "./logs/", "日志文件存储目录")
flag.StringVar(&action, "action", "", "其它参数 区分需要执行的任务名")
flag.Parse()
_ = boot.Init(configPath,logPath)
switch action{
case "Corder":
//若已下单，任务标记为已处理，不释放；
cron_crm_todotask.Corder()
case "TimeoutRelease":
//若60天没有关闭任务，则任务状态为已超时，该客户释放回小组池
cron_crm_todotask.TimeoutRelease()
case "RemindTimeout":
//任务超时前3天于上午9：00邮件提醒，邮件抄送上级组长、主管
cron_crm_todotask.RemindTimeout()
}
}
cmd/crm/cron_crmschedulememo/main.go
......@@ -20,7 +20,7 @@ func main(){
flag.StringVar(&configPath, "config", "./config/dev/", "配置文件")
flag.StringVar(&logPath, "logdir", "./logs/", "日志文件存储目录")
flag.Parse()
boot.Init(configPath,logPath)
_ = boot.Init(configPath,logPath)
//关闭已超时的
cron_crm_schedulememo.CloseTimeoutMemo()
//备忘录 提醒
......
config/dev/config.toml
......@@ -8,6 +8,7 @@ sendMsgByChangeInvoiceCom = "http://crm.liexin.net/api/sendMsgByChangeInvoiceCom
apiKey = "crm a1b2c3d4e5f6g7h8i9jk"
#crm发送备忘录提醒
crmDomain = "http://crm.liexin.net/"
ichuntapiDomain = "http://api.liexin.com"
#异步任务转发url
asyncTaskforward = "http://192.168.1.168:8700/callback"
......
config/dev/db.toml
......@@ -9,7 +9,7 @@ ShowSQL = true
max_conn_life_time = 100
[cms]
data_source_name = "ichuntcms:ichuntcms#zsyM@tcp(192.168.1.235:3306)/ichuntcms?charset=utf8&parseTime=true&loc=Asia%2FShanghai"
data_source_name = "ichuntcms:ichuntcms#zsyM@tcp(192.168.1.252:3306)/ichuntcms?charset=utf8&parseTime=true&loc=Asia%2FShanghai"
max_open_conn = 0
max_idle_conn = 0
table_prefix = "lie_"
......@@ -18,7 +18,7 @@ ShowSQL = true
[liexin]
data_source_name = "liexin:liexin#zsyM@tcp(192.168.2.232:3306)/liexin?charset=utf8&parseTime=true&loc=Asia%2FShanghai"
data_source_name = "liexin:liexin#zsyM@tcp(192.168.2.252:3306)/liexin?charset=utf8&parseTime=true&loc=Asia%2FShanghai"
max_open_conn = 0
max_idle_conn = 0
table_prefix = "lie_"
......@@ -32,14 +32,14 @@ ShowSQL = true
table_prefix = ""
max_conn_life_time = 100
data_source_name = [
"micro_service:lie_micro_service#zsyM@tcp(192.168.2.232:3306)/lie_micro_service?charset=utf8&parseTime=true&loc=Asia%2FChongqing",
"micro_service:lie_micro_service#zsyM@tcp(192.168.2.232:3306)/lie_micro_service?charset=utf8&parseTime=true&loc=Asia%2FChongqing",
"micro_service:lie_micro_service#zsyM@tcp(192.168.2.232:3306)/lie_micro_service?charset=utf8&parseTime=true&loc=Asia%2FChongqing",
"micro_service:lie_micro_service#zsyM@tcp(192.168.2.252:3306)/lie_micro_service?charset=utf8&parseTime=true&loc=Asia%2FChongqing",
"micro_service:lie_micro_service#zsyM@tcp(192.168.2.252:3306)/lie_micro_service?charset=utf8&parseTime=true&loc=Asia%2FChongqing",
"micro_service:lie_micro_service#zsyM@tcp(192.168.2.252:3306)/lie_micro_service?charset=utf8&parseTime=true&loc=Asia%2FChongqing",
]
[micro_service]
dns = "micro_service:lie_micro_service#zsyM@tcp(192.168.2.232:3306)/lie_micro_service?charset=utf8&parseTime=true&loc=Asia%2FChongqing"
dns = "micro_service:lie_micro_service#zsyM@tcp(192.168.2.252:3306)/lie_micro_service?charset=utf8&parseTime=true&loc=Asia%2FChongqing"
max_open_conn = 20
max_idle_conn = 10
table_prefix = ""
......@@ -47,14 +47,14 @@ ShowSQL = true
[sku]
[sku.sku_0]
dns = "micro_service:lie_micro_service#zsyM@tcp(192.168.2.232:3306)/lie_micro_service?charset=utf8&parseTime=true&loc=Asia%2FChongqing"
dns = "micro_service:lie_micro_service#zsyM@tcp(192.168.2.252:3306)/lie_micro_service?charset=utf8&parseTime=true&loc=Asia%2FChongqing"
max_open_conn = 20
max_idle_conn = 10
table_prefix = ""
max_conn_life_time = 100
[sku.sku_1]
dns = "micro_service:lie_micro_service#zsyM@tcp(192.168.2.232:3306)/lie_micro_service?charset=utf8&parseTime=true&loc=Asia%2FChongqing"
dns = "micro_service:lie_micro_service#zsyM@tcp(192.168.2.252:3306)/lie_micro_service?charset=utf8&parseTime=true&loc=Asia%2FChongqing"
max_open_conn = 20
max_idle_conn = 10
table_prefix = ""
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment