Go microservice framework practice
Go kit framework
Start service registration and Discovery Center
The consul used by the service registration and discovery center can be downloaded from the official website https://www.consul.io/docs/agent Consul of the corresponding system.
windows case:
After downloading and installing, it is an exe executable file, which will be executed in the file directory/ consul.exe version displays the version number/ consul.exe agent -dev starts in development mode
linux case:
After downloading, there is an executable file consumption. Add the executable file to sudo MV consumption / usr / local / bin / in the bin directory, and start the consumption agent - Dev in the development mode
Note: the development mode is not applicable to production. Refer to the official documents for the start of production mode.
Service registration and discovery
Use the go kit framework to interact with the service and registry Consul through HTTP.
- Projects created using the go kit framework are generally divided into three layers:
-
Service layer: business logic implementation layer. Practice demo this layer interface implements three service methods:
- Health check method: HealthCheck(), check whether the service registered with the service discovery center (that is, the greeting method below) is normal.
- Greeting method: SayHello(), a method that needs to be registered with the service registration and discovery center.
- Service discovery method: DiscoveryService(ctx context.Context, serviceName string) ([]interface{}, error), take the service name registered in the service registry (such as "SayHello" above) as the parameter, access the service discovery interface provided by the service registration and discovery center, and return the service instance information registered in the service registry.
-
Endpoint layer: for a project that needs to be registered with the service and registry center, not all business logic codes need to be registered. The service layer method encapsulates the business logic code, so the endpoint layer is the further packaging of some methods of the service layer that need to be registered with the service registration and discovery center, which is equivalent to the methods provided by RPC services.
The service method of demo endpoint encapsulation is as follows:
type DiscoveryEndpoints struct { SayHelloEndpoint endpoint.Endpoint DiscoveryEndpoint endpoint.Endpoint HealthCheckEndpoint endpoint.Endpoint }The endpoint layer needs to accept the request and return the response, so it is necessary to construct the request structure and response structure of the service corresponding to each endpoint. The request and response structures of the three endpoints in the demo are as follows:
// SayHello service request structure type SayHelloRequest struct { } // SayHello service response structure type SayHelloResponse struct { Message string `json:"message"` } // Service discovery request structure type DiscoveryRequest struct { ServiceName string } // Service discovery response structure type DiscoveryResponse struct { Instances []interface{} `json:"instances"` Error string `json:"error"` } // Health check request structure type HealthRequest struct{} // Health check response structure type HealthResponse struct { Status bool `json:"status"` } -
transport layer: the entrance of the project to provide external services, and forwards the corresponding request url to the corresponding endpoint.
The project demo uses GitHub of go kit The NewServer method under COM / go Kit / Kit / transport / HTTP needs to pass in the corresponding endpoint, deserialize the request func and serialize the response func of the corresponding endpoint, and other optional parameters.
The processing methods of the three endpoint requests in the project demo are as follows:
options := []kithttp.ServerOption{ kithttp.ServerErrorHandler(transport.NewLogErrorHandler(logger)), kithttp.ServerErrorEncoder(encodeError), } r.Methods("GET").Path("/say-hello").Handler(kithttp.NewServer( endpoints.SayHelloEndpoint, decodeSayHelloRequest, encodeJsonResponse, options..., )) r.Methods("GET").Path("/discovery").Handler(kithttp.NewServer( endpoints.DiscoveryEndpoint, decodeDiscoveryRequest, encodeJsonResponse, )) r.Methods("GET").Path("/health").Handler(kithttp.NewServer( endpoints.HealthCheckEndpoint, decodeHealthCheckRequest, encodeJsonResponse, options..., )) -
The whole practice demo project is registered to the service registration and Discovery Center Consul as a micro service. The project interacts with the service registration center by implementing the client discoveryClient of Consul:
type DiscoveryClient interface { /** * Service registration interface * @param serviceName service name * @param instanceId Service instance Id * @param instancePort Service instance port * @param healthCheckUrl Health check address * @param instanceHost Service instance address * @param meta Service instance metadata */ Register(serviceName, instanceId, healthCheckUrl string, instanceHost string, instancePort int, Meta map[string]string, logger *log.Logger) bool /** * @Description Service logoff interface * @Param instanceId Service instance Id * @return bool **/ DeRegister(instanceId string, logger *log.Logger) bool /** * @Description Discovery service instance interface * @Param serviceName service name * @return []interface{} **/ DiscoverServices(serviceName string, logger *log.Logger) []interface{} } -
The main file flow of the demo project startup entry is as follows:
-
First declare the demo service address and service name
Consumer address, global shi below, errChan (write service error information and service system termination signal)
-
Declare the Service, discover the client and initialize the client. If the initialization fails, close the Service. If the initialization succeeds, declare and initialize the Service using the client
-
Create each endpoint using the initialized Service
-
Create http Handler to pass in the global context, all endpoints and loggers
-
Create a goroutine and start the http server
Before starting, the goroutine registers the demo service to consumer through the consumer client. If the registration fails, write the failure information to errChan and close the service.
Then use the handler created above to create and listen for the service.
-
Create a goroutine monitoring system signal and write the termination signals such as ctrl + c to errChan
-
The main thread takes out the error information from errChan. No error information is blocked. When there is an error information, the service exits and cancels the registration
After the demo service is registered successfully, check the Consul web interface and find multiple service instances: the registered service instance named SayHello:
When the service is stopped, the registration center is as follows:
-
The above implements a custom client that interacts with consul t through HTTP interaction, but the go kit framework has built-in packages that interact with the consumer service registry:
import ( "github.com/go-kit/kit/sd/consul" "github.com/hashicorp/consul/api" "github.com/hashicorp/consul/api/watch" "log" "strconv" "sync" )
Construct kit discover client:
For consumer interaction, you need to use "GitHub. COM / hashicorp / consumer / API"
func NewKitDiscoverClient(consultHost string, consulPort int) (DiscoveryClient, error) {
// Create a consumer through the consumer host and consumer port Client
consulConfig := api.DefaultConfig()
consulConfig.Address = consultHost + ":" + strconv.Itoa(consulPort)
apiClient, err := api.NewClient(consulConfig)
if err != nil {
return nil, err
}
client := consul.NewClient(apiClient)
return &KitDiscoverClient{
Host: consultHost,
Port: consulPort,
config: consulConfig,
client: client,
}, err
}
Registration:
When you use the go kit framework to register the built-in client and consumer, you do not need to customize the service instance information, and use API Agentserviceregistration build service instance metadata:
func (consulClient *KitDiscoverClient) Register(serviceName, instanceId, healthCheckUrl string, instanceHost string, instancePort int, Meta map[string]string, logger *log.Logger) bool {
// 1. Build service instance metadata
serviceRegistration := &api.AgentServiceRegistration{
ID: instanceId,
Name: serviceName,
Address: instanceHost,
Port: instancePort,
Meta: Meta,
Check: &api.AgentServiceCheck{
DeregisterCriticalServiceAfter: "30s",
HTTP: "http://" + instanceHost + ":" + strconv.Itoa(instancePort) + healthCheckUrl,
Interval: "15s",
},
}
// 2. The sending service is registered in consult
err := consulClient.client.Register(serviceRegistration)
if err != nil {
log.Println("Register Service Error")
return false
}
log.Println("Register Service Success!")
return true
}
cancellation:
Logging out also requires API Agentserviceregistration build service instance information:
func (consulClient *KitDiscoverClient) DeRegister(instanceId string, logger *log.Logger) bool {
// Build a metadata structure containing the service instance ID
serviceRegistration := &api.AgentServiceRegistration{
ID: instanceId,
}
// Send service logoff request
err := consulClient.client.Deregister(serviceRegistration)
if err != nil {
logger.Println("Deregister Service Error!")
return false
}
log.Println("Deregister Service Success!")
return true
}
Discovery service instance:
Discover service instances, use atomic locks and dictionaries to cache the service instance information list, and monitor the changes of service instance data under the service name through the Watch mechanism provided by consul to reduce the number of interactions with Consul HTTP:
func (consulClient *KitDiscoverClient) DiscoverServices(serviceName string, logger *log.Logger) []interface{} {
// The service has been monitored and cached
instanceList, ok := consulClient.instancesMap.Load(serviceName)
if ok {
return instanceList.([]interface{})
}
// Apply for lock
consulClient.mutex.Lock()
defer consulClient.mutex.Unlock()
// Check the cache again
instanceList, ok = consulClient.instancesMap.Load(serviceName)
if ok {
return instanceList.([]interface{})
}else {
go func() {
// Use the consumer service instance to monitor the change of the instance list of a service
params := make(map[string]interface{})
params["type"] = "service"
params["service"] = serviceName
plan, _ := watch.Parse(params)
plan.Handler = func(u uint64, i interface{}) {
if i == nil {
return
}
v, ok := i.([]*api.ServiceEntry)
if !ok {
return
}
// No service instance Online
if len(v) == 0 {
consulClient.instancesMap.Store(serviceName, []interface{}{})
}
var healthServices []interface{}
for _, service := range v {
if service.Checks.AggregatedStatus() == api.HealthPassing {
healthServices = append(healthServices, service.Service)
}
}
consulClient.instancesMap.Store(serviceName, healthServices)
}
defer plan.Stop()
plan.Run(consulClient.config.Address)
}()
}
// Request a list of service instances based on the service name
entries, _, err := consulClient.client.Service(serviceName, "", false, nil)
if err != nil {
consulClient.instancesMap.Store(serviceName, []interface{}{})
logger.Println("Discover Service Error!")
return nil
}
instances := make([]interface{}, len(entries))
for i := 0; i < len(instances); i++ {
instances[i] = entries[i].Service
}
consulClient.instancesMap.Store(serviceName, instances)
return instances
}