修改

config/config.go

@@ -0,0 +1,57 @@
+package config
+
+import (
+	"flag"
+	log "github.com/sirupsen/logrus"
+	"gopkg.in/yaml.v3"
+	"io/ioutil"
+	"os"
+)
+
+var (
+	root = flag.String("dir", "D:/gopath/src/edu_api", "设置配置文件的根路径")
+
+	Config = parseYaml()
+)
+
+func parseYaml() *configuration {
+	flag.Parse()
+
+	cfg := new(configuration)
+	cfg, err := cfg.yaml(*root + "/config/config.yaml")
+	if err != nil {
+		log.Info("parse yaml config error:", err.Error())
+	}
+
+	log.Info("cfg is:", cfg.Queue.Addr)
+	return cfg
+}
+
+type configuration struct {
+	Queue queue `json:"queue",yaml:"queue"`
+}
+
+type queue struct {
+	Addr string `json:"addr",yaml:"addr"`
+}
+
+func (cfg *configuration) yaml(f string) (*configuration, error) {
+	file, err := os.Open(f)
+	if err != nil {
+		return nil, err
+	}
+
+	defer file.Close()
+
+	data, err := ioutil.ReadAll(file)
+	if err != nil {
+		return nil, err
+	}
+
+	err = yaml.Unmarshal(data, cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	return cfg, nil
+}

config/config.yaml.example

@@ -0,0 +1,2 @@
+queue:
+   addr: 127.0.0.1:9501

controllers/cashier/cashierController.go

@@ -1,9 +1,11 @@
 package cashier
 
 import (
+	"adx-api.pingcoo.com/config"
 	"edu_api/controllers"
 	"edu_api/middlewares"
 	"errors"
+	"fmt"
 	"github.com/ant0ine/go-json-rest/rest"
 	log "github.com/sirupsen/logrus"
 )
@@ -44,6 +46,8 @@ func (cashier *CashierController) Payment(w rest.ResponseWriter, r *rest.Request
 
 func (cashier *CashierController) PayNotify(w rest.ResponseWriter, r *rest.Request) {
 	var message string
+	fmt.Printf("config.Cfg.Queue.Addr:%s", config.Cfg.Queue.Addr)
+
 	if r.PathParam("type") == "alipay" { //包括 余额、花呗(分期)
 		message = cashier.controller.BaseOrm.PayNotify(r, "alipay")
 	} else if r.PathParam("type") == "wechat_pay" { //包括h5支付、jsapi支付

log/request

@@ -1 +1 @@
-D:/gopath/src/edu_api/log/request.20191226.log
+D:/gopath/src/edu_api/log/request.20191227.log

services/cashierService.go

@@ -7,6 +7,7 @@ import (
 	"crypto/x509"
 	"edu_api/middlewares"
 	"edu_api/models"
+	"edu_api/tasksAndEvents"
 	"edu_api/utils"
 	"encoding/base64"
 	"encoding/hex"
@@ -357,7 +358,7 @@ func afterPayOrder(notifyReq models.NotifyRequestModel, extend interface{}) int
 				return utils.ORDER_INFO_ERROR
 			}
 
-			return updateOrderInfo(vip.ID, "h_orders", value.PaySource, notifyReq)
+			return updateOrderInfo(vip.ID, "h_orders", value.PaySource, value.BranchType, notifyReq)
 		} else if value.BranchType == "invoice" {
 			if err := db.GetDB().Table("h_invoices").Where("id = ? ", value.Id).First(&invoice).Error; err != nil {
 				return utils.ORDER_INFO_ERROR
@@ -374,7 +375,7 @@ func afterPayOrder(notifyReq models.NotifyRequestModel, extend interface{}) int
 				return utils.ORDER_INFO_ERROR
 			}
 
-			return updateOrderInfo(vip.ID, "h_invoices", value.PaySource, notifyReq)
+			return updateOrderInfo(vip.ID, "h_invoices", value.PaySource, value.BranchType, notifyReq)
 		} else if value.BranchType == "vip" {
 			if err := db.GetDB().Table("h_vip_orders").Where("id = ? ", value.Id).First(&vip).Error; err != nil {
 				return utils.ORDER_INFO_ERROR
@@ -391,7 +392,7 @@ func afterPayOrder(notifyReq models.NotifyRequestModel, extend interface{}) int
 				return utils.ORDER_INFO_ERROR
 			}
 
-			return updateOrderInfo(vip.ID, "h_vip_orders", value.PaySource, notifyReq)
+			return updateOrderInfo(vip.ID, "h_vip_orders", value.PaySource, value.BranchType, notifyReq)
 		}
 	}
 
@@ -401,7 +402,7 @@ func afterPayOrder(notifyReq models.NotifyRequestModel, extend interface{}) int
 /**
 更新订单信息
 */
-func updateOrderInfo(id int, table_name, payment_method string, notifyReq models.NotifyRequestModel) int {
+func updateOrderInfo(id int, table_name, payment_method, branch_type string, notifyReq models.NotifyRequestModel) int {
 	updated_at, _ := FormatLocalTime(time.Now())
 
 	if result := db.GetDB().Table(table_name).Where("id = ?", id).Update(map[string]interface{}{
@@ -413,9 +414,13 @@ func updateOrderInfo(id int, table_name, payment_method string, notifyReq models
 		"payment_order_no": notifyReq.TradeNo,
 		"receipt_amount":   notifyReq.ReceiptAmount,
 	}).RowsAffected; result > 0 {
-		//同时发消息
-
 		//异步更新课程信息
+		order_execute := &tasksAndEvents.OrderExecute{OrderId: id, BranchType: branch_type}
+		order_execute.Update()
+
+		//同时发消息
+		paid_success_message := &tasksAndEvents.PaidSuccessMessage{OrderId: id, BranchType: branch_type, PaySource: payment_method, EventType: "pay_order_success"}
+		paid_success_message.Send()
 
 		return utils.ORDER_INFO_OK
 	}

tasksAndEvents/operate.go

@@ -0,0 +1,37 @@
+package tasksAndEvents
+
+import (
+	"bytes"
+	"net"
+	"time"
+)
+
+/**
+定义一个转换消息体的接口
+*/
+type Operate interface {
+	ToBytes() ([]byte, error)
+}
+
+/**
+数据库操作
+*/
+func operateDB(operate Operate) (int, error) {
+	conn, err := net.DialTimeout("tcp", "", 200*time.Millisecond)
+	if err != nil {
+		return 0, err
+	}
+
+	defer conn.Close()
+
+	data, err := operate.ToBytes()
+	if err != nil {
+		return 0, err
+	}
+
+	var buffer bytes.Buffer
+	buffer.Write(data)
+	buffer.WriteString("\n") //这个必须，因为php swoole默认是以\n结束消息的
+
+	return conn.Write(buffer.Bytes())
+}

tasksAndEvents/orderExecute.go

@@ -0,0 +1,36 @@
+package tasksAndEvents
+
+import "encoding/json"
+
+type orderExecute struct {
+	Class  string        `json:"class"`
+	Method string        `json:"method"`
+	Params *OrderExecute `json:"params"`
+}
+
+func (o *orderExecute) ToBytes() ([]byte, error) {
+	o.Class = "OrderExecute"
+	o.Method = "handle"
+
+	data, err := json.Marshal(o)
+	if err != nil {
+		return nil, err
+	}
+
+	return data, nil
+}
+
+type OrderExecute struct {
+	OrderId    int    `json:"order_id"`
+	BranchType string `json:"branch_type"`
+}
+
+/**
+这个主要是更新用户课程的异步操作
+*/
+func (o *OrderExecute) Update() (int, error) {
+	data := new(orderExecute)
+	data.Params = o
+
+	return operateDB(data)
+}

tasksAndEvents/paidSuccessMessage.go

@@ -0,0 +1,38 @@
+package tasksAndEvents
+
+import "encoding/json"
+
+type paidSuccessMessage struct {
+	Class  string              `json:"class"`
+	Method string              `json:"method"`
+	Params *PaidSuccessMessage `json:"params"`
+}
+
+func (o *paidSuccessMessage) ToBytes() ([]byte, error) {
+	o.Class = "OrderExecute"
+	o.Method = "handle"
+
+	data, err := json.Marshal(o)
+	if err != nil {
+		return nil, err
+	}
+
+	return data, nil
+}
+
+type PaidSuccessMessage struct {
+	OrderId    int    `json:"order_id"`
+	BranchType string `json:"branch_type"`
+	PaySource  string `json:"pay_source"`
+	EventType  string `json:"event_type"`
+}
+
+/**
+这个主要是更新用户课程的异步操作
+*/
+func (o *PaidSuccessMessage) Send() (int, error) {
+	data := new(paidSuccessMessage)
+	data.Params = o
+
+	return operateDB(data)
+}