package commons
import (
"fmt"
log "github.com/sirupsen/logrus"
)
var MonthMap map[string]string = map[string]string{
"January": "01",
"February": "02",
"March": "03",
"April": "04",
"May": "05",
"June": "06",
"July": "07",
"August": "08",
"September": "09",
"October": "10",
"November": "11",
"December": "12",
}
var MonthIntMap map[string]int = map[string]int{
"January": 1,
"February": 2,
"March": 3,
"April": 4,
"May": 5,
"June": 6,
"July": 7,
"August": 8,
"September": 9,
"October": 10,
"November": 11,
"December": 12,
}
func FailOnError(err error, msg string) {
if err != nil {
log.Fatalf("%s: %s", msg, err)
panic(fmt.Sprintf("%s: %s", msg, err))
}
}
//往前多少个月
func GetUnixToOldTime(i int) int64 {
currentMonth := commons.MonthIntMap[time.Now().Month().String()]
oldMonth := currentMonth - i
t := time.Date(time.Now().Year(),time.Month(oldMonth),1,0,0,0,0,time.Local)
return t.Unix()
}
func HttpDo(url string, token string, param string) (string, error) {
if url == "" || token == "" || param == "" {
return "", errors.New("url,token,param不正确")
}
client := &http.Client{}
fmt.Println("url:",url)
fmt.Println("token:",token)
fmt.Println("param:",param)
req, err := http.NewRequest("POST", url, bytes.NewBuffer([]byte(param)))
if err != nil {
return "", err
}
req.Header.Set("token", token)
req.Header.Set("Content-Type", "application/json;charset=UTF-8")
resp, err := client.Do(req)
if err != nil {
return "", err
}
defer resp.Body.Close()
if resp.StatusCode == http.StatusOK {
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return "", err
}
return string(body), nil
} else {
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return "", err
}
return string(body), errors.New("http请求错误")
}
}
func Md5(str string) string {
md5Ctx := md5.New()
md5Ctx.Write([]byte(str))
cipherStr := md5Ctx.Sum(nil)
return strings.ToUpper(hex.EncodeToString(cipherStr))
}
// 全国文化市场网络游戏发展趋势 产品访问量
func (this *OnlineGame) ReqGameProductPV() (map[string][]commons.BaseIntIntData, error) {
var startTime int64 = utils.GetUnixToOldTime(12)
var endTime int64 = utils.GetNowUnix()
param := commons.StrMd5{Appkey: commons.ApiBigapiAppkey, Time: time.Now().Unix()*1000, StartTime: startTime * 1000, EndTime: endTime * 1000}
parambyte, _ := json.Marshal(param)
token := utils.Md5("/ccmdss/gameProductPV"+string(parambyte)+commons.ApiWebapiScrect)
val, err := utils.HttpDo(commons.ApiCcmdssUrl + "/ccmdss/gameProductPV",token,string(parambyte))
fmt.Println(string(val))
if err != nil {
log.Errorf("全国文化市场网络游戏发展趋势 产品访问量:%s", err)
return nil, err
}
valueMap := make(map[string][]commons.BaseIntIntData)
err = json.Unmarshal([]byte(val), &valueMap)
if err != nil {
log.Errorf("全国文化市场网络游戏发展趋势 产品访问量,数据转结构体错误:%s", err)
return nil, err
}
return valueMap, err
}
//产品访问量 一年
func (ds *DecisionService) DecisionServiceTraffic() {
og := models.OnlineGame{}
rtogp, err := og.ReqGameProductPV()
returna := make(map[string][]commons.BaseIntIntData)
if err != nil {
log.Errorf("业务 产品访问量 ,查询失败：%s", err)
return
}
baseInt64List := []int64{}
for i:= 11;i >= 0;i-- {
baseInt64List = append(baseInt64List, utils.GetUnixToOldTime(i)*1000)
}
for key,value := range rtogp{
baseInt64BaseIntIntData := []commons.BaseIntIntData{}
for _,value := range baseInt64List{
baseInt64BaseIntIntData = append(baseInt64BaseIntIntData,commons.BaseIntIntData{value,0})
}
for _,timeValue := range value{
returna[key] = baseInt64BaseIntIntData
for index,time := range baseInt64BaseIntIntData {
if time.Name == timeValue.Name{
baseInt64BaseIntIntData[index].Value = timeValue .Value
}
continue
}
}
}
returnTus := []commons.TrafficUserScaleUserScaleUserScale{}
for key,value := range returna{
reList := []commons.BaseIntData{}
for _,value := range value{
ft := time.Unix(value.Name/1000,0)
reList = append(reList, commons.BaseIntData{ft.Format("01"),value.Value})
}
returnTus = append(returnTus,commons.TrafficUserScaleUserScaleUserScale{key,reList} )
}
s := connectors.DecisionSaveTraffic(returnTus)
if s.Code != commons.SUCCESS {
log.Errorf("全国文化市场 产品访问量 ,保存Redis失败：%s", err)
return
}
strByte, _ := json.Marshal(returnTus)
connectors.Send("{\"decision\":{\"traffic\":" + string(strByte) + "}}")
fmt.Println("{\"decision\":{\"traffic\":" + string(strByte) + "}}")
log.Debugln("全国文化市场 产品访问量 ,发送MQTT成功：%s", err)
fmt.Println("ok")
}
package connectors
import (
"../commons"
"../plugins"
"strings"
"github.com/streadway/amqp"
log "github.com/sirupsen/logrus"
"github.com/surgemq/message"
"github.com/surgemq/surgemq/service"
"strconv"
)
func Send(msg string) {
defer func() {
if err := recover(); err != nil {
log.Errorf("MQ发送存储消息错误 %s", err)
}
}()
switch strings.ToUpper(commons.MqSend) {
case "MQTT":
mqttSend(msg)
case "RABBIT":
rabbitMqSend(msg)
default:
rabbitMqSend(msg)
}
}
func mqttSend(data string) {
c := &service.Client{}
msg := message.NewConnectMessage()
msg.SetWillQos(commons.MqttConfig.MqttClient.Qos)
msg.SetVersion(4)
msg.SetCleanSession(true)
msg.SetClientId([]byte(commons.MqttConfig.MqttClient.ClientId))
msg.SetKeepAlive(commons.MqttConfig.MqttClient.KeepAlive)
msg.SetWillTopic([]byte(commons.MqttConfig.MqttClient.Topic))
msg.SetWillMessage([]byte(data))
msg.SetUsername([]byte(commons.MqttConfig.MqttClient.Username))
msg.SetPassword([]byte(commons.MqttConfig.MqttClient.Password))
mqttAddr := "tcp://:1883"
if commons.MqttConfig.MqttServer.MqttPort > 0 {
mqttAddr = "tcp://:" + strconv.Itoa(commons.MqttConfig.MqttClient.MqttPort)
}
// Connects to the remote server at 127.0.0.1 port 1883
err := c.Connect(mqttAddr, msg)
commons.FailOnError(err, "MQTT客户端发送数据失败")
defer c.Disconnect()
}
func RabbitStart() (*amqp.Connection, error) {
rabbitConfig := commons.RabbitConfig
conn, err := amqp.Dial("amqp://" + rabbitConfig.Username + ":" + rabbitConfig.Password + "@" + rabbitConfig.Host + ":" + strconv.Itoa(rabbitConfig.Port) + "/" + rabbitConfig.VHost)
return conn, err
}
func rabbitMqSend(msg string) {
rabbitConfig := commons.RabbitConfig
conn, err := plugins.RabbitStart()
commons.FailOnError(err, "RabbitMQ连接失败")
defer conn.Close()
ch, err := conn.Channel()
commons.FailOnError(err, "RabbitMQ管道打开失败")
defer ch.Close()
err = ch.ExchangeDeclare(
rabbitConfig.Exchange, // name
"fanout", // type
true, // durable
false, // auto-deleted
false, // internal
false, // no-wait
nil, // arguments
)
commons.FailOnError(err, "RabbitMQ的exchange定义失败")
err = ch.Publish(
rabbitConfig.Exchange, // exchange
"mcprc.#", // routing key
false, // mandatory
false, // immediate
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(msg),
})
commons.FailOnError(err, "RabbitMQ发送消息失败")
}
type TrafficUserScaleUserScaleUserScale struct {
Name string `json:"name"`
Value []BaseIntData `json:"value"`
}
type BaseIntData struct {
Name string `json:"name"`
Value int `json:"value"`
}
type BaseIntIntData struct {
Name int64 `json:"name"`
Value int `json:"value"`
}
网友评论