Go 微服务，第11部分：Hystrix和Resilience

> git clone https://github.com/callistaenterprise/goblog.git
> git checkout P11

func getQuote() (model.Quote, error) {
 body, err := cb.CallUsingCircuitBreaker("quotes-service", "http://quotes-service:8080/api/quote?strength=4", "GET")
    // Code handling response or err below, omitted for clarity
    ...
 }

func CallUsingCircuitBreaker(breakerName string, url string, method string) ([]byte, error) {
        output := make(chan []byte, 1)  // Declare the channel where the hystrix goroutine will put success responses.
        errors := hystrix.Go(breakerName,    // Pass the name of the circuit breaker as first parameter.
            // 2nd parameter, the inlined func to run inside the breaker.
            func() error {
                    // Create the request. Omitted err handling for brevity
                    req, _ := http.NewRequest(method, url, nil)
                    // For hystrix, forward the err from the retrier. It's nil if successful.  
                    return callWithRetries(req, output)
            }, 
            // 3rd parameter, the fallback func. In this case, we just do a bit of logging and return the error.
            func(err error) error {
                    logrus.Errorf("In fallback function for breaker %v, error: %v", breakerName, err.Error())
                    circuit, _, _ := hystrix.GetCircuit(breakerName)
                    logrus.Errorf("Circuit state is: %v", circuit.IsOpen())
                    return err
        })
        // Response and error handling. If the call was successful, the output channel gets the response. Otherwise,
        // the errors channel gives us the error.
        select {
        case out := <-output:
                logrus.Debugf("Call in breaker %v successful", breakerName)
                return out, nil
        case err := <-errors:
                return nil, err
        }
}

func callWithRetries(req *http.Request, output chan []byte) error {
    // Create a retrier with constant backoff, RETRIES number of attempts (3) with a 100ms sleep between retries.
    r := retrier.New(retrier.ConstantBackoff(RETRIES, 100 * time.Millisecond), nil)
    // This counter is just for getting some logging for showcasing, remove in production code.
    attempt := 0
    // Retrier works similar to hystrix, we pass the actual work (doing the HTTP request) in a func.
    err := r.Run(func() error {
            attempt++
            // Do HTTP request and handle response. If successful, pass resp.Body over output channel,
            // otherwise, do a bit of error logging and return the err.
            resp, err := Client.Do(req)
            if err == nil && resp.StatusCode < 299 {
                    responseBody, err := ioutil.ReadAll(resp.Body)
                    if err == nil {
                            output <- responseBody
                            return nil
                    }
                    return err
            } else if err == nil {
                    err = fmt.Errorf("Status was %v", resp.StatusCode)
            }
            logrus.Errorf("Retrier failed, attempt %v", attempt)
            return err
    })
    return err
}

func TestCallUsingResilienceLastSucceeds(t *testing.T) {
        defer gock.Off()
        buildGockMatcherTimes(500, 2)        // First two requests respond with 500 Server Error
        body := []byte("Some response")
        buildGockMatcherWithBody(200, string(body))   // Next (3rd) request respond with 200 OK
        hystrix.Flush()     // Reset circuit breaker state
        Convey("Given a Call request", t, func() {
                Convey("When", func() {
                        // Call single time (will become three requests given that we retry thrice)
                        bytes, err := CallUsingCircuitBreaker("TEST", "http://quotes-service", "GET")
                        Convey("Then", func() {
                                // Assert no error and expected response
                                So(err, ShouldBeNil)
                                So(bytes, ShouldNotBeNil)
                                So(string(bytes), ShouldEqual, string(body))
                        })
                })
        })
}

ERRO[2017-09-03T10:26:28.106] Retrier failed, attempt 1                    
ERRO[2017-09-03T10:26:28.208] Retrier failed, attempt 2                    
DEBU[2017-09-03T10:26:28.414] Call in breaker TEST successful

hystrix.ConfigureCommand("quotes-service", hystrix.CommandConfig{
    SleepWindow:            5000,
RequestVolumeThreshold: 10,
})

hystrix.command.[circuit name].[config property] = [value]

hystrix.command.quotes-service.SleepWindow: 5000

> go run *.go -zuul=false

{"name":"Person_6","servedBy":"10.255.0.19","quote":{"quote":"To be or not to be","ipAddress":"10.0.0.22"},"imageUrl":"http://imageservice:7777/file/cake.jpg"} 
{"name":"Person_23","servedBy":"10.255.0.21","quote":{"quote":"You, too, Brutus?","ipAddress":"10.0.0.25"},"imageUrl":"http://imageservice:7777/file/cake.jpg"}

> docker service scale quotes-service=0

{name":"Person_23","servedBy":"10.255.0.19","quote":{"quote":"May the source be with you, always.","ipAddress":"circuit-breaker"},"imageUrl":"http://imageservice:7777/file/cake.jpg"}

http://quotes-service:8080/api/quote?strength=4

func publishDiscoveryToken(amqpClient messaging.IMessagingClient) {
        // Get hold of our IP adress (reads it from /etc/hosts) and build a discovery token.
        ip, _ := util.ResolveIpFromHostsFile()
        token := DiscoveryToken{
                State:   "UP",
                Address: ip,
        }
        bytes, _ := json.Marshal(token)
        // Enter an eternal loop in a new goroutine that sends the UP token every
        // 30 seconds to the "discovery" channel.
        go func() {
                for {
                        amqpClient.PublishOnQueue(bytes, "discovery")
                        time.Sleep(time.Second * 30)
                }
        }()
}

# Hystrix Dashboard
docker build -t someprefix/hystrix support/monitor-dashboard
docker service rm hystrix
docker service create --constraint node.role==manager --replicas 1 -p 7979:7979 --name hystrix --network my_network --update-delay 10s --with-registry-auth  --update-parallelism 1 someprefix/hystrix
# Turbine
docker service rm turbine
docker service create --constraint node.role==manager --replicas 1 -p 8282:8282 --name turbine --network my_network --update-delay 10s --with-registry-auth  --update-parallelism 1 eriklupander/turbine