grpcexercises/consuldemo/server/main.go

package main

import (
	"context"
	"git.gqnotes.com/guoqiang/grpcexercises/consuldemo/pkg/consul"
	"github.com/hashicorp/consul/api"
	"google.golang.org/grpc/reflection"
	"log"
	"net"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

	"git.gqnotes.com/guoqiang/grpcexercises/consuldemo/pb"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
	"google.golang.org/grpc"
)

func main() {
	// consul地址，此处为本地地址
	consulAddress := "localhost:8500"
	// 服务名称
	serviceName := "service-grpcdemo"

	logger := zap.NewExample()
	// 实现服务端逻辑
	lis, err := net.Listen("tcp", ":5630")

	if err != nil {
		panic(err)
	}

	s := grpc.NewServer()

	pb.RegisterGreetServiceServer(s, &Server{})

	reflection.Register(s)

	logger.Info("grpc server start")

	// 启动相关服务
	var gg errgroup.Group

	gg.Go(func() error {
		logger.Debug("grpc server start")

		go func() {
			if err := s.Serve(lis); err != nil {
				logger.Fatal("grpc server start failed", zap.Error(err))
			}
		}()

		return nil
	})

	gg.Go(func() error {
		if err := consul.InitConsulClient(consulAddress); err != nil {
			return err
		}

		logger.Debug("consul client start")

		return nil
	})

	if err = gg.Wait(); err != nil {
		logger.Fatal("server start failed", zap.Error(err))
		return
	}

	// 注册服务

	// 健康检查
	go func() {
		mux := http.NewServeMux()

		mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
			_, _ = w.Write([]byte("ok"))
		})

		err = http.ListenAndServe(":8081", mux)

		if err != nil {
			logger.Fatal("health check start failed", zap.Error(err))
		}

	}()

	check := &api.AgentServiceCheck{
		Interval:                       "3s",
		Timeout:                        "2s",
		DeregisterCriticalServiceAfter: "60s",
		HTTP:                           "http://127.0.0.1:8081/health",
		Method:                         "GET",
	}

	err = consul.GetConsulClient().Agent().ServiceRegister(&api.AgentServiceRegistration{
		Name:  serviceName,
		Port:  5630,
		Check: check,
		ID:    "1001",
		Tags:  []string{"grpc.port=5630"},
	})

	if err != nil {
		logger.Fatal("register service failed", zap.Error(err))
		return
	}

	// 优雅关闭
	ch := make(chan os.Signal, 1)

	// 监听信号
	signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL, syscall.SIGHUP, syscall.SIGQUIT)

	// 阻塞等待信号
	sig := <-ch

	logger.Info("receive signal", zap.Any("signal", sig))

	// 关闭服务
	s.GracefulStop()

	logger.Debug("server stop")
}

type Server struct {
	pb.UnimplementedGreetServiceServer
}

func (s *Server) Greet(ctx context.Context, req *pb.GreetRequest) (resp *pb.GreetResponse, err error) {
	t0 := time.Now()

	defer func() {
		log.Println("Greet cost:", time.Since(t0))
	}()

	resp = &pb.GreetResponse{
		Result: "hello, " + req.Greeting + "!",
	}

	return
}

func (s *Server) GreetManyTimes(request *pb.GreetRequest, stream pb.GreetService_GreetManyTimesServer) (err error) {
	for i := 0; i < 10; i++ {
		err = stream.Send(&pb.GreetResponse{
			Result: "hello, " + request.Greeting + "!",
		})

		if err != nil {
			return
		}

		time.Sleep(time.Millisecond * 1)
	}

	return
}