Merge pull request #13 from aberic/master

fix some bug

CODE_OF_CONDUCT.md

@@ -0,0 +1,76 @@
+# Contributor Covenant Code of Conduct
+
+## Our Pledge
+
+In the interest of fostering an open and welcoming environment, we as
+contributors and maintainers pledge to making participation in our project and
+our community a harassment-free experience for everyone, regardless of age, body
+size, disability, ethnicity, sex characteristics, gender identity and expression,
+level of experience, education, socio-economic status, nationality, personal
+appearance, race, religion, or sexual identity and orientation.
+
+## Our Standards
+
+Examples of behavior that contributes to creating a positive environment
+include:
+
+* Using welcoming and inclusive language
+* Being respectful of differing viewpoints and experiences
+* Gracefully accepting constructive criticism
+* Focusing on what is best for the community
+* Showing empathy towards other community members
+
+Examples of unacceptable behavior by participants include:
+
+* The use of sexualized language or imagery and unwelcome sexual attention or
+ advances
+* Trolling, insulting/derogatory comments, and personal or political attacks
+* Public or private harassment
+* Publishing others' private information, such as a physical or electronic
+ address, without explicit permission
+* Other conduct which could reasonably be considered inappropriate in a
+ professional setting
+
+## Our Responsibilities
+
+Project maintainers are responsible for clarifying the standards of acceptable
+behavior and are expected to take appropriate and fair corrective action in
+response to any instances of unacceptable behavior.
+
+Project maintainers have the right and responsibility to remove, edit, or
+reject comments, commits, code, wiki edits, issues, and other contributions
+that are not aligned to this Code of Conduct, or to ban temporarily or
+permanently any contributor for other behaviors that they deem inappropriate,
+threatening, offensive, or harmful.
+
+## Scope
+
+This Code of Conduct applies both within project spaces and in public spaces
+when an individual is representing the project or its community. Examples of
+representing a project or community include using an official project e-mail
+address, posting via an official social media account, or acting as an appointed
+representative at an online or offline event. Representation of a project may be
+further defined and clarified by project maintainers.
+
+## Enforcement
+
+Instances of abusive, harassing, or otherwise unacceptable behavior may be
+reported by contacting the project team at aberic@qq.com. All
+complaints will be reviewed and investigated and will result in a response that
+is deemed necessary and appropriate to the circumstances. The project team is
+obligated to maintain confidentiality with regard to the reporter of an incident.
+Further details of specific enforcement policies may be posted separately.
+
+Project maintainers who do not follow or enforce the Code of Conduct in good
+faith may face temporary or permanent repercussions as determined by other
+members of the project's leadership.
+
+## Attribution
+
+This Code of Conduct is adapted from the [Contributor Covenant][homepage], version 1.4,
+available at https://www.contributor-covenant.org/version/1/4/code-of-conduct.html
+
+[homepage]: https://www.contributor-covenant.org
+
+For answers to common questions about this code of conduct, see
+https://www.contributor-covenant.org/faq

bow/bow.go

@@ -22,6 +22,7 @@ import (
 	"github.com/ennoo/rivet/utils/log"
 	"github.com/gin-gonic/gin"
 	"go.uber.org/zap"
+	"net/http"
 	"sync"
 )
 
@@ -90,8 +91,8 @@ func RunBow(context *gin.Context, serviceName string) {
 	routeService, ok := instance.AllWay[serviceName]
 	if !ok {
 		err := fmt.Errorf("routeService not fount")
-		fmt.Println("not found ", serviceName)
 		log.Shunt.Error(err.Error(), zap.String("serviceName", serviceName))
+		context.JSON(http.StatusOK, err.Error())
 	} else {
 		request.SyncPoolGetRequest().Call(context, context.Request.Method, routeService.OutRemote, routeService.OutURI)
 	}
@@ -102,8 +103,8 @@ func RunBowCallback(context *gin.Context, serviceName string, f func() *response
 	routeService, ok := instance.AllWay[serviceName]
 	if !ok {
 		err := fmt.Errorf("service not fount")
-		fmt.Println("not found ", serviceName)
 		log.Shunt.Error(err.Error(), zap.String("serviceName", serviceName))
+		context.JSON(http.StatusOK, err.Error())
 	} else {
 		request.SyncPoolGetRequest().Callback(context, context.Request.Method, routeService.OutRemote, routeService.OutURI, f)
 	}

examples/routine/routine2.go

@@ -0,0 +1,56 @@
+/*
+ * Copyright (c) 2019. ENNOO - All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package main
+
+import (
+	"fmt"
+	"time"
+)
+
+func main() {
+
+	c := make(chan int, 2)
+	defer close(c)
+	//提前将队列放满
+	c <- 1
+	c <- 2
+	fmt.Println("开始尝试执行")
+	go cross(c)
+	process(c)
+
+}
+
+func process(c chan int) {
+
+	fmt.Println("被限流阻塞")
+	c <- 1 //channel已满，将阻塞，直到成功放入channel  **
+	fmt.Println("已放行，执行process")
+
+}
+
+func cross(c chan int) {
+	a := 1
+	b := 5
+	for b >= a {
+		fmt.Println("阻塞", b, "秒")
+		time.Sleep(time.Second)
+		b -= 1
+	}
+
+	fmt.Println("释放一个通行证")
+	<-c //取出元素，则chan可以继续放入数据，将唤醒**行代码
+
+}

examples/shunt1/shunt1.go

@@ -36,9 +36,9 @@ func main() {
 	//	ServiceName: env.GetEnvDefault("SERVICE_NAME", "shunt1"),
 	//})
 	rivet.UseDiscovery(discovery.ComponentConsul, "127.0.0.1:8500", "shunt", "127.0.0.1", 8083)
-	rivet.Shunt().Register("test", &shunt.RoundRobinBalance{Position: 0})
-	rivet.Shunt().Register("test1", &shunt.RandomBalance{})
-	rivet.Shunt().Register("test2", &shunt.HashBalance{Key: []string{}})
+	rivet.Shunt().Register("test", shunt.Round)
+	rivet.Shunt().Register("test1", shunt.Random)
+	rivet.Shunt().Register("test2", shunt.Hash)
 	//addAddress()
 	rivet.ListenAndServe(&rivet.ListenServe{
 		Engine:      rivet.SetupRouter(testShunt1),
@@ -56,7 +56,7 @@ func testShunt1(engine *gin.Engine) {
 func shunt3(context *gin.Context) {
 	rivet.Response().Do(context, func(result *response.Result) {
 		serviceName := context.Param("serviceName")
-		rivet.Shunt().Register(serviceName, &shunt.RoundRobinBalance{Position: 0})
+		rivet.Shunt().Register(serviceName, shunt.Round)
 		result.SaySuccess(context, "test2")
 	})
 }

rivet/init.go

@@ -91,7 +91,7 @@ func UseDiscovery(component, url, serviceName, hostname string, port int) {
 			log.Rivet.Info("use discovery service {}" + discovery.ComponentConsul)
 			ud = true
 			if request.LB {
-				consul.Enroll(url, serviceID, ServiceName(), hostname, port)
+				go consul.Enroll(url, serviceID, ServiceName(), hostname, port)
 			} else {
 				go scheduled.ConsulEnroll(url, serviceID, ServiceName(), hostname, port)
 			}

shunt/hash.go

@@ -22,19 +22,10 @@ import (
 	"math/rand"
 )
 
-// HashBalance 负载均衡 hash 策略实体
-type HashBalance struct {
-	Key []string
-}
-
-// Run 负载均衡 round 策略实现
-func (hash *HashBalance) Run(serviceName string) (add *server.Service, err error) {
+// RunHash 负载均衡 hash 策略实现
+func RunHash(serviceName string) (add *server.Service, err error) {
 	services := server.ServiceGroup()[serviceName].Services
 	defKey := fmt.Sprintf("%d", rand.Int())
-	if len(hash.Key) > 0 {
-		defKey = hash.Key[0]
-	}
-
 	lens := len(services)
 	if lens == 0 {
 		err = fmt.Errorf("no balance")

shunt/random.go

@@ -21,12 +21,8 @@ import (
 	"math/rand"
 )
 
-// RandomBalance 负载均衡 random 策略实体
-type RandomBalance struct {
-}
-
-// Run 负载均衡 round 策略实现
-func (random *RandomBalance) Run(serviceName string) (add *server.Service, err error) {
+// RunRandom 负载均衡 random 策略实现
+func RunRandom(serviceName string) (add *server.Service, err error) {
 	services := server.ServiceGroup()[serviceName].Services
 	if len(services) == 0 {
 		err = errors.New("no instance")

shunt/round.go

@@ -20,24 +20,41 @@ import (
 	"github.com/ennoo/rivet/server"
 )
 
+var roundRobinBalances map[string]*RoundRobinBalance
+
 // RoundRobinBalance 负载均衡 round 策略实体
 type RoundRobinBalance struct {
-	Position int
+	serviceName string
+	rrbCh       chan int
+}
+
+// generaCount 自增生成一个0~65535的数，到达65535则重0开始计数
+func generaCount() chan int {
+	var ch = make(chan int)
+	go func() {
+		for i := 0; ; i++ {
+			ch <- i // 等待索要数据
+			if i == 65535 {
+				i = 0
+			}
+		}
+	}()
+	return ch
 }
 
-// Run 负载均衡 round 策略实现
-func (round *RoundRobinBalance) Run(serviceName string) (add *server.Service, err error) {
+// RunRound 负载均衡 round 策略实现
+func RunRound(serviceName string) (service *server.Service, err error) {
 	services := server.ServiceGroup()[serviceName].Services
-	if len(services) == 0 {
+	var lens int
+	if lens = len(services); nil == services || lens == 0 {
 		err = errors.New("no instance")
 		return
 	}
-
-	lens := len(services)
-	if round.Position >= lens {
-		round.Position = 0
+	roundRobinBalance := roundRobinBalances[serviceName]
+	var position int
+	if position = <-roundRobinBalance.rrbCh; position >= lens {
+		position = position % lens
 	}
-	add = services[round.Position]
-	round.Position++
+	service = services[position]
 	return
 }

shunt/shunt.go

@@ -23,46 +23,68 @@ import (
 	"sync"
 )
 
-var instance *Shunt
-var once sync.Once
+const (
+	// Random 负载均衡 random 策略
+	Random = iota
+	// Round 负载均衡 round 策略
+	Round
+	// Hash 负载均衡 hash 策略
+	Hash
+)
+
+var (
+	instance *Shunt
+	once     sync.Once
+)
 
 // GetShuntInstance 获取负载管理对象 Shunt 单例
 func GetShuntInstance() *Shunt {
 	once.Do(func() {
-		instance = &Shunt{AllWay: make(map[string]Way)}
+		instance = &Shunt{AllWay: make(map[string]int)}
 	})
 	return instance
 }
 
-// Way 负载均衡方式接口
-type Way interface {
-	// Run 负载均衡算法
-	Run(string) (*server.Service, error)
-}
-
 // Shunt 负载入口对象
 type Shunt struct {
-	AllWay map[string]Way
+	AllWay map[string]int
 }
 
 // Register 注册新的负载方式
-func (s *Shunt) Register(serviceName string, way Way) {
+func (s *Shunt) Register(serviceName string, way int) {
+	switch way {
+	case Round:
+		if nil == roundRobinBalances {
+			roundRobinBalances = make(map[string]*RoundRobinBalance)
+		}
+		roundRobinBalances[serviceName] = &RoundRobinBalance{
+			serviceName: serviceName,
+			rrbCh:       generaCount(),
+		}
+	}
 	instance.AllWay[serviceName] = way
 }
 
 // RunShunt 开启负载
-func RunShunt(serviceName string) (*server.Service, error) {
+func RunShunt(serviceName string) (service *server.Service, err error) {
 	way, ok := instance.AllWay[serviceName]
 	if !ok {
 		err := fmt.Errorf("service not fount")
 		fmt.Println("not found ", serviceName)
 		log.Shunt.Error(err.Error(), zap.String("serviceName", serviceName))
 		return nil, err
 	}
-	service, err := way.Run(serviceName)
+	switch way {
+	case Random:
+		service, err = RunRandom(serviceName)
+	case Round:
+		service, err = RunRound(serviceName)
+	case Hash:
+		service, err = RunHash(serviceName)
+	}
 	if err != nil {
 		err = fmt.Errorf(" %s erros", serviceName)
-		return nil, err
+		return
 	}
-	return service, nil
+	return
 }