使用golang多線程批量更新數(shù)據(jù)數(shù)據(jù)。

共4個文件

main.go

1. package main 
2.  
3. import ( 
4.     "bufio" 
5.     "fmt" 
6.     _ "github.com/go-sql-driver/mysql" 
7.     "os" 
8.     . "pt" 
9. ) 
10.  
11. func main() { 
12.     args := os.Args 
13.     if len(args) == 2 { 
14.         switch args[1] { 
15.         case "1": 
16.             Cu.Run() 
17.         case "2": 
18.             //平臺負(fù)債表數(shù)據(jù)靜態(tài)化 
19.             Fz.Run() 
20.         case "0": 
21.             os.Exit(0) 
22.         default: 
23.         } 
24.     } 
25.     if len(args) == 1 { 
26.         for { 
27.             fmt.Println("操作目錄: ") 
28.             fmt.Println("1、平臺有效客戶更新（202:13306-platform）。 ") 
29.             fmt.Println("2、平臺負(fù)債數(shù)據(jù)靜態(tài)化（202:13306-platform）。") 
30.             fmt.Println("0、退出。 ") 
31.             inputReader := bufio.NewReader(os.Stdin) 
32.             command, _, _ := inputReader.ReadLine() 
33.             code := string(command) 
34.             switch code { 
35.             case "1": 
36.                 Cu.Run() 
37.             case "2": 
38.                 //平臺負(fù)債表數(shù)據(jù)靜態(tài)化 
39.                 Fz.Run() 
40.             case "0": 
41.                 os.Exit(0) 
42.             default: 
43.                 fmt.Println("default") 
44.             } 
45.             fmt.Println("-------處理完成-------") 
46.         } 
47.     } 
48.  
49. } 

pt/custatus.go

// 客戶屬性自動更新封裝

// 需要公司開啟自動更新并配置客戶過期周期時間

package pt

import (

"fmt"

"log"

"strconv"

"strings"

"time"

)

type CuStatusClass struct {

}

func NewCu() *CuStatusClass {

obj := new(CuStatusClass)

return obj

}

/**

*判斷客戶狀態(tài)

**/

func (obj *CuStatusClass) getState(orders int, practs int, state int) int {

if orders > 0 || practs > 0 {

return 1 //最近有訂單OR有實操 為有效客戶

}

if state == 3 {

return 3 //死檔客戶

}

if state == -1 {

return -1 //無效客戶

}

//默認(rèn)返回為久黨客戶

return 2

}

/*

* 獲取公司關(guān)于有效客戶的配置天數(shù) 默認(rèn)30天

*/

func (obj *CuStatusClass) getConfig(str string) int {

var num int

n := strings.Index(str, "member_config")

if n == -1 {

num = 30

} else {

start := n + 20

end := n + 22

num2 := string([]byte(str)[start:end])

num, _ = strconv.Atoi(num2)

}

return num

}

/**

* 更新一個公司的客戶狀態(tài) (PT) 考慮新建數(shù)據(jù)庫連接 提高效率

* @param {[type]} db *sql.DB [description]

* @param {[type]} c chan int [description]

* @param {[type]} comp_id int [公司ID]

* @param {[type]} num int [有效期天數(shù)]

* @return {[type]} [description]

*/

func (obj *CuStatusClass) updateOneComp(c chan string, comp_id int, num int) {

db := Mydb()

defer db.Close()

end := time.Now().Unix()

start := end - 3600*24*int64(num) //前推num天

sql := "SELECT a.id,a.name,a.status,(SELECT COUNT(*) FROM `order_sale` WHERE `cu_id` = a.id AND `pay_time` > ? AND `pay_time` < ? AND `type` IN (1,2)) AS orders, (SELECT COUNT(*) FROM `practice_order` WHERE `cu_id` = a.id AND `pay_time` > ? AND `pay_time` < ?) AS practs FROM `customer_info` AS a LEFT JOIN `config_membership` AS m ON a.membership_id = m.id WHERE m.`is_member` = 1 AND a.`company_id` = ?"

rows, err := db.Query(sql, start, end, start, end, comp_id)

if err != nil {

log.Println(err)

}

defer rows.Close()

var id int

var orders int

var practs int

var name string

var status int

for rows.Next() {

err := rows.Scan(&id, &name, &status, &orders, &practs)

if err != nil {

log.Fatal(err)

}

new_status := obj.getState(orders, practs, status)

if status != new_status {

stmt, err := db.Prepare("UPDATE `customer_info` SET `status`=? WHERE `id`=?")

defer stmt.Close()

if err != nil {

log.Println(err)

return

}

stmt.Exec(new_status, id)

//返回管道信息寫入

c <- fmt.Sprintf("%d", comp_id) + ":" + fmt.Sprintf("%d", id) + " " + name + " " + fmt.Sprintf("%d", status) + "->" + fmt.Sprintf("%d", new_status)

}

}

err = rows.Err()

if err != nil {

log.Fatal(err)

}

close(c)

}

/**

* 多公司并發(fā)處理 (PT)

* @param {[type]} db *sql.DB [description]

* @return {[type]} [description]

*/

func (obj *CuStatusClass) Run() {

db := Mydb()

defer db.Close()

sql := "SELECT id , auto_cu_status, config FROM `company_info` WHERE `status` = 1"

rows, err := db.Query(sql)

if err != nil {

log.Println(err)

}

defer rows.Close()

chs := make([]chan string, 0) //開多個管道接受消息

var id int

var auto int

var config string

for rows.Next() {

err := rows.Scan(&id, &auto, &config)

if err != nil {

log.Fatal(err)

}

if auto == 1 {

num := obj.getConfig(config) //客戶有效期設(shè)置

c := make(chan string)

chs = append(chs, c)

go obj.updateOneComp(c, id, num)

}

}

err = rows.Err()

if err != nil {

log.Fatal(err)

}

for _, ch := range chs { //多管道寫法

for {

x, ok := <-ch

if ok == false {

break

}

writeResult("cu_status", x)

fmt.Println(x) //消息回收處理 可擴展寫入文件日志

}

}

}

var Cu *CuStatusClass

func init() {

Cu = NewCu()

}