gd_service/impl/rcvr.go

// Copyright 2019 getensh.com. All rights reserved.
// Use of this source code is governed by getensh.com.

// package impl implement all interfaces of micro service for vehicle
package impl

import (
	"context"
	"encoding/json"
	"gd_service/impl/service"
	"gd_service/impl/thirdparty_impl/cdbd"
	"gd_service/impl/thirdparty_impl/dy"
	"gd_service/impl/thirdparty_impl/dybd"
	"gd_service/impl/thirdparty_impl/zr"
	"github.com/astaxie/beego/orm"
	"runtime"
	"sync"
	"time"

	"gd_service/apis"
	"gd_service/impl/thirdparty_impl/adm"

	"gd_service/common.in/cache"
	"gd_service/common.in/jsonrpc2"
	"gd_service/common.in/task"
	"gd_service/common.in/utils"
	"go.uber.org/zap"
)

// 具体实现
type Rcvr struct {
}

// 操作pm的公共锁
var mutex sync.Mutex

var pm = map[string]*ParamReuse{}

//短时复用说明:
//1.相同接口相同参数，如果一次调用未完成，又发起另一次调用，则复用第一次调用的结果
//2.第一次是指全局的第一次（服务器集群内该接口的该类参数的第一次调用）
//3.如果本次为第一次调用，需设置本地参数缓存（ParamReuse）和redis参数缓存（ParamReuse）
//4.如果已有本地参数缓存，则等待结果
//5.如果没有本地参数缓存，且没有redis缓存，则本次为第一次，作步骤3操作
//6.如果没有本地参数缓存, 但已有redis缓存，则等待redis结果
//7.本地参数引用计数为0时，清除本地参数缓存
//8.redis参数引用计数为0时，清除redis缓存
type ParamReuse struct {
	//参数引用计数
	Used int

	//响应结果
	Res string

	//是否使用了redis
	Redis bool `json:"-"`

	//参数锁引用计数
	Locked int `json:"-"`

	//参数锁
	Mx sync.Mutex `json:"-"`
}

const (
	ParamExist        = 1
	ParamExistInRedis = 2
	ParamNotExist     = 3
)

func getFuncName() string {
	funcPtr, _, _, _ := runtime.Caller(1)
	return runtime.FuncForPC(funcPtr).Name()
}

func paramLock(req string) {
	mutex.Lock()
	p, ok := pm[req]
	if ok == false {
		pm[req] = &ParamReuse{}
		p = pm[req]
	}
	p.Locked += 1
	mutex.Unlock()

	p.Mx.Lock()
}

func paramUnlock(req string, clean bool) {
	mutex.Lock()
	defer mutex.Unlock()
	if p, ok := pm[req]; ok {
		defer p.Mx.Unlock()
		if p.Locked > 0 {
			p.Locked -= 1
		}
		if clean == false {
			return
		}
		if p.Used > 0 {
			p.Used -= 1
		}
		if p.Locked == 0 && p.Used == 0 {
			if p.Redis {
				cleanParamReqFromRedis(req)
			}
			delete(pm, req)
			if len(pm) == 0 {
				pm = map[string]*ParamReuse{}
			}
		}
	}
}

func setParamReq(req string) int {
	paramLock(req)
	defer paramUnlock(req, false)
	p, _ := pm[req]
	if p.Used == 0 {
		utils.Lock(req)
		defer utils.UnLock(req)

		s, _ := cache.Redis.Get(req)
		p.Used = 1
		p.Redis = true
		if s == "" {
			pr := ParamReuse{
				Used: 1,
			}
			bytes, _ := json.Marshal(pr)
			cache.Redis.SetEx(req, 60, string(bytes))
			return ParamNotExist
		}
		pr := ParamReuse{}
		json.Unmarshal([]byte(s), &pr)
		pr.Used++
		bytes, _ := json.Marshal(pr)
		cache.Redis.SetEx(req, 60, string(bytes))
		return ParamExistInRedis

	}
	p.Used += 1
	return ParamExist
}

func cleanParamReqFromRedis(req string) {
	utils.Lock(req)
	defer utils.UnLock(req)
	s, _ := cache.Redis.Get(req)
	if s == "" {
		return
	}
	pr := ParamReuse{}
	json.Unmarshal([]byte(s), &pr)
	pr.Used -= 1
	if pr.Used == 0 {
		cache.Redis.Del(req)
		return
	}
	bytes, _ := json.Marshal(pr)
	cache.Redis.SetEx(req, 60, string(bytes))
}

func cleanParamReq(req string) {
	paramLock(req)
	paramUnlock(req, true)
}

func setParamRes(req string, res string) {
	paramLock(req)
	defer paramUnlock(req, false)
	p, _ := pm[req]
	p.Res = res
	if p.Redis == false {
		return
	}
	setParamResToRedis(req, res)
}

func setParamResLocal(req string, res string) {
	paramLock(req)
	defer paramUnlock(req, false)
	p, _ := pm[req]
	p.Res = res
}

func setParamResToRedis(req string, res string) {
	utils.Lock(req)
	defer utils.UnLock(req)
	s, _ := cache.Redis.Get(req)
	if s == "" {
		return
	}
	pr := ParamReuse{}
	json.Unmarshal([]byte(s), &pr)
	pr.Res = res
	bytes, _ := json.Marshal(pr)
	cache.Redis.SetEx(req, 60, string(bytes))
}

func getParamRes(req string) string {
	paramLock(req)
	defer paramUnlock(req, false)
	p, _ := pm[req]
	return p.Res
}

func getParamResFromRedis(req string) string {
	s, _ := cache.Redis.Get(req)
	if s == "" {
		return ""
	}
	pr := ParamReuse{}
	json.Unmarshal([]byte(s), &pr)
	return pr.Res
}

func setOrWaitParam(req string) string {
	exist := setParamReq(req)
	if exist == ParamNotExist {
		return ""
	}
	count := 0
	if exist == ParamExist {
		for {
			res := getParamRes(req)
			if res != "" {
				return res
			}
			time.Sleep(50 * time.Millisecond)
			count++
			if count >= 1200 {
				break
			}
		}
	}
	if exist == ParamExistInRedis {
		for {
			res := getParamResFromRedis(req)
			if res != "" {
				setParamResLocal(req, res)
				return res
			}
			time.Sleep(50 * time.Millisecond)
			count++
			if count >= 1200 {
				break
			}
		}

	}
	return ""
}

func (c *Rcvr) Query(ctx context.Context, req *apis.CommonReq, reply *apis.CommonReply) error {
	if false {
		funcName := getFuncName()

		newReq := *req
		newReq.MerchantApiInfo = apis.MerchantApiInfo{}
		reqBytes, _ := json.Marshal(newReq)
		reqString := string(reqBytes) + funcName
		setOrWaitParam(reqString)
		defer cleanParamReq(reqString)

		defer func() {

			setParamRes(reqString, "-")
		}()
	}
	t1 := func() error {
		return service.Query(ctx, req, reply)
	}

	err := task.Do(ctx, t1)
	if err != nil {
		var e jsonrpc2.Error
		merr := json.Unmarshal([]byte(err.Error()), &e)
		if merr != nil {
			reply.ErrCode = int(1001)
			reply.ErrMsg = "服务错误"
		}
		reply.ErrCode = e.Code
		reply.ErrMsg = e.Message
	}
	return nil
}

func SetLogger(logger *zap.Logger) {
	service.SetLogger(logger)
	adm.SetLogger(logger)
	cdbd.SetLogger(logger)
	dy.SetLogger(logger)
	zr.SetLogger(logger)
	dybd.SetLogger(logger)
}

func RegisterOrmModel() {
	orm.RegisterModel(
		new(apis.CdbdData),
		new(apis.DybdX001Data),
		new(apis.DybdX002Data))
}