yyfbaloot/servers/micros/common/common.go

package common

import (
	"fmt"
	"math/rand"
	"net"
	"time"

	"bet24.com/log"
	"bet24.com/servers/monitor"
	"bet24.com/utils"
	consulapi "github.com/hashicorp/consul/api"
	"github.com/rcrowley/go-metrics"
	"github.com/rpcxio/rpcx-consul/serverplugin"
	"github.com/smallnest/rpcx/server"
)

const Service_base_path = "bet24.com"
const Default_Consul_Addr = "localhost:8500"
const port_monitor_delta = 1000

var _port int
var _server *server.Server
var serviceId string

var consul_client *consulapi.Client

func RunService(name string, rcvr interface{}, consulAddr string) {
	run(name, rcvr, "8.141.80.205:8500", true, 0, fmt.Sprintf("log/%s", name))
}

func RunServiceWithPort(name string, rcvr interface{}, consulAddr string, port int, logFilePath string) {
	run(name, rcvr, "localhost:8500", true, port, logFilePath)
}

func RunWithoutChangeTitle(name string, rcvr interface{}, consulAddr string) *server.Server {
	return run(name, rcvr, "localhost:8500", false, 0, "")
}

func run(name string, rcvr interface{}, consulAddr string, changeTitle bool, port int, logFilePath string) *server.Server {
	//utils.SetErrorFile(fmt.Sprintf("log/%s/err.log", name), fmt.Sprintf("%s starting", name))
	rand.Seed(time.Now().UnixNano())
	if port == 0 {
		port = getRandomPort()
	}

	serviceId = fmt.Sprintf("%s_%s_%d", name, getLocalIp(), port)
	local_server := server.NewServer()
	if !addRegistryPlugin(local_server, name, consulAddr, port) {
		return nil
	}
	local_server.RegisterName(name, rcvr, "")

	if changeTitle {
		monitorPort := port + port_monitor_delta
		if port == 5000 {
			monitorPort = 5100
		}
		monitor.Run(monitorPort, logFilePath)
		utils.SetConsoleTitle(fmt.Sprintf("[S]%s running in port[%d] monitor[%d]", name, port, monitorPort))
		_server = local_server
		local_server.Serve("tcp", getAddress(port))
		_port = port
	} else {
		go local_server.Serve("tcp", getAddress(port))
		log.Release("side service [%s] running in port[%d]", name, port)
	}

	return local_server
}

func GetServer() *server.Server {
	return _server
}

func DeregisterService(name string) {
	if consul_client != nil {
		consul_client.Agent().ServiceDeregister(serviceId)
	}
	if _server != nil {
		_server.UnregisterAll()
		utils.SetConsoleTitle(fmt.Sprintf("Service[%s] stopping in port[%d]", name, _port))
	} else {
		log.Release("DeregisterService instance not found")
	}
}

func getRandomPort() int {
	var port int
	tryMax := 100
	for i := 0; i < tryMax; i++ {
		port = 6000 + rand.Intn(1000)
		if !utils.CheckPortInUse(port) {
			return port
		}
	}
	log.Release("getRandomPort failed")
	port = 6999
	return port
}

func GetLocalIp() string {
	return getLocalIp()
}

func GetServicePort() int {
	return _port
}

func getLocalIp() string {
	addrs, err := net.InterfaceAddrs()
	if err != nil {
		fmt.Println(err)
		return ""
	}
	//fmt.Println(addrs)
	for _, address := range addrs {
		// 检查ip地址判断是否回环地址
		if ipnet, ok := address.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
			if ipnet.IP.To4() != nil {
				return ipnet.IP.String()
			}
		}
	}
	return ""
}

func getServiceAddress(port int) string {
	return fmt.Sprintf("tcp@%s:%d", getLocalIp(), port)
}

func getAddress(port int) string {
	return fmt.Sprintf(":%d", port)
}

func addRegistryPlugin(s *server.Server, name string, consulAddr string, port int) bool {
	r := &serverplugin.ConsulRegisterPlugin{
		ServiceAddress: getServiceAddress(port),
		ConsulServers:  []string{consulAddr},
		BasePath:       Service_base_path,
		Metrics:        metrics.NewRegistry(),
		UpdateInterval: time.Second * 10,
	}
	err := r.Start()
	if err != nil {
		log.Release("%v", err)
		return false
	}
	s.Plugins.Add(r)
	registerConsul(consulAddr, name, port)
	return true
}

func registerConsul(consulAddress string, name string, port int) {
	config := consulapi.DefaultConfig()
	config.Address = consulAddress
	client, err := consulapi.NewClient(config)
	if err != nil {
		fmt.Println("consul client error : ", err)
		return
	}

	// 创建注册到consul的服务到
	registration := new(consulapi.AgentServiceRegistration)
	registration.Address = getLocalIp()
	registration.ID = serviceId
	registration.Name = name
	registration.Port = port

	// 增加consul健康检查回调函数
	check := new(consulapi.AgentServiceCheck)
	check.TCP = fmt.Sprintf("%s:%d", registration.Address, registration.Port)
	check.Timeout = "10s"                        //超时
	check.Interval = "5s"                        //健康检查频率
	check.DeregisterCriticalServiceAfter = "30s" // 故障检查失败30s后 consul自动将注册服务删除
	registration.Check = check

	client.Agent().ServiceRegister(registration)
	consul_client = client
}