Женщина-испытатель спросила меня, как использовать gRPC, и я скинул ей эту статью напрямую.

Оригинальная ссылка: Женщина-испытатель спросила меня, как использовать gRPC, и я скинул ей эту статью напрямую.

Предыдущая статьяgRPC, большая похвалаОн взорвался напрямую.Контент в основном включает в себя: простой сервис gRPC, режим потоковой обработки, валидатор, аутентификацию токена и аутентификацию сертификата.

Объем чтения на нескольких платформах достиг нового максимума, и это было рекомендовано на главной странице oschina, а объем чтения достиг 1w+, что уже является пиком моего одиночного чтения.

Кажется, что пока вы пишете внимательно, вы все равно можете что-то получить.

В этой статье мы по-прежнему начинаем с реального боя, в основном представляя режим публикации-подписки gRPC, интерфейс REST и контроль времени ожидания.

Я загружу соответствующий код вGitHub, заинтересованные друзья могут пойти посмотреть или скачать.

опубликовать и подписаться

Публикация-подписка — это распространенный шаблон проектирования, и многие реализации этого шаблона уже существуют в сообществе с открытым исходным кодом. Среди них минималистская реализация pubsub представлена в проекте docker Ниже приведен код локальной публикации и подписки, основанный на пакете pubsub:

package main

import (
	"fmt"
	"strings"
	"time"

	"github.com/moby/moby/pkg/pubsub"
)

func main() {
	p := pubsub.NewPublisher(100*time.Millisecond, 10)

	golang := p.SubscribeTopic(func(v interface{}) bool {
		if key, ok := v.(string); ok {
			if strings.HasPrefix(key, "golang:") {
				return true
			}
		}
		return false
	})
	docker := p.SubscribeTopic(func(v interface{}) bool {
		if key, ok := v.(string); ok {
			if strings.HasPrefix(key, "docker:") {
				return true
			}
		}
		return false
	})

	go p.Publish("hi")
	go p.Publish("golang: https://golang.org")
	go p.Publish("docker: https://www.docker.com/")
	time.Sleep(1)

	go func() {
		fmt.Println("golang topic:", <-golang)
	}()
	go func() {
		fmt.Println("docker topic:", <-docker)
	}()

	<-make(chan bool)
}

Этот код сначала проходитpubsub.NewPublisherсоздал объект, затем передалp.SubscribeTopicреализовать подписку,p.Publishпубликовать новости.

Эффект от исполнения следующий:

docker topic: docker: https://www.docker.com/
golang topic: golang: https://golang.org
fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan receive]:
main.main()
	/Users/zhangyongxin/src/go-example/grpc-example/pubsub/server/pubsub.go:43 +0x1e7
exit status 2

Сообщения о подписке печатаются нормально.

Но возникает тупиковая ошибка, потому что этот оператор<-make(chan bool)вызванный. Но без этого оператора сообщение о подписке не может быть напечатано нормально.

Я не очень хорошо понимаю здесь.Если кто-то знает, пожалуйста, оставьте сообщение и попросите совета.

Затем реализуйте модель публикации-подписки с помощью gRPC и пакета pubsub.

Необходимо реализовать четыре части:

protoдокумент;
Сервер:Он используется для получения запросов на подписку, а также запросов на публикацию и пересылки запросов на публикацию подписчикам;
Подписка клиентов:Используется для подписки на сообщения с сервера и обработки сообщений;
Опубликовать клиент:Используется для отправки сообщений на сервер.

прото-файл

Сначала определите файл proto:

syntax = "proto3";

package proto;
 
message String {
    string value = 1;
}
 
service PubsubService {
    rpc Publish (String) returns (String);
    rpc SubscribeTopic (String) returns (stream String);
    rpc Subscribe (String) returns (stream String);
}

Определите три метода, один для публикацииPublishи две подпискиSubscribeа такжеSubscribeTopic.

Subscribeметод получает все сообщения, аSubscribeTopicв соответствии с конкретнымTopicПолучать сообщения.

Сервер

package main

import (
	"context"
	"fmt"
	"log"
	"net"
	"server/proto"
	"strings"
	"time"

	"github.com/moby/moby/pkg/pubsub"
	"google.golang.org/grpc"
	"google.golang.org/grpc/reflection"
)

type PubsubService struct {
	pub *pubsub.Publisher
}

func (p *PubsubService) Publish(ctx context.Context, arg *proto.String) (*proto.String, error) {
	p.pub.Publish(arg.GetValue())
	return &proto.String{}, nil
}

func (p *PubsubService) SubscribeTopic(arg *proto.String, stream proto.PubsubService_SubscribeTopicServer) error {
	ch := p.pub.SubscribeTopic(func(v interface{}) bool {
		if key, ok := v.(string); ok {
			if strings.HasPrefix(key, arg.GetValue()) {
				return true
			}
		}
		return false
	})

	for v := range ch {
		if err := stream.Send(&proto.String{Value: v.(string)}); nil != err {
			return err
		}
	}
	return nil
}

func (p *PubsubService) Subscribe(arg *proto.String, stream proto.PubsubService_SubscribeServer) error {
	ch := p.pub.Subscribe()

	for v := range ch {
		if err := stream.Send(&proto.String{Value: v.(string)}); nil != err {
			return err
		}
	}
	return nil
}

func NewPubsubService() *PubsubService {
	return &PubsubService{pub: pubsub.NewPublisher(100*time.Millisecond, 10)}
}

func main() {
	lis, err := net.Listen("tcp", ":50051")
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}

	// 简单调用
	server := grpc.NewServer()
	// 注册 grpcurl 所需的 reflection 服务
	reflection.Register(server)
	// 注册业务服务
	proto.RegisterPubsubServiceServer(server, NewPubsubService())

	fmt.Println("grpc server start ...")
	if err := server.Serve(lis); err != nil {
		log.Fatalf("failed to serve: %v", err)
	}
}

По сравнению с предыдущей программой публикации и подписки, здесь*pubsub.PublisherКак структура gRPCPubsubServiceчлен .

Затем, в соответствии с процессом разработки gRPC, реализуются три метода, соответствующие структуре.

Наконец, при регистрации сервиса поставьтеNewPubsubService()Внедрение службы для реализации функции локальной публикации и подписки.

Клиент подписки

package main

import (
	"client/proto"
	"context"
	"fmt"
	"io"
	"log"

	"google.golang.org/grpc"
)

func main() {
	conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()

	client := proto.NewPubsubServiceClient(conn)
	stream, err := client.Subscribe(
		context.Background(), &proto.String{Value: "golang:"},
	)
	if nil != err {
		log.Fatal(err)
	}

	go func() {
		for {
			reply, err := stream.Recv()
			if nil != err {
				if io.EOF == err {
					break
				}
				log.Fatal(err)
			}
			fmt.Println("sub1: ", reply.GetValue())
		}
	}()

	streamTopic, err := client.SubscribeTopic(
		context.Background(), &proto.String{Value: "golang:"},
	)
	if nil != err {
		log.Fatal(err)
	}

	go func() {
		for {
			reply, err := streamTopic.Recv()
			if nil != err {
				if io.EOF == err {
					break
				}
				log.Fatal(err)
			}
			fmt.Println("subTopic: ", reply.GetValue())
		}
	}()

	<-make(chan bool)
}

создать новыйNewPubsubServiceClientобъект, а затем реализоватьclient.Subscribeа такжеclient.SubscribeTopicметодом, а затем непрерывно получать сообщения через горутину.

опубликовать клиент

package main

import (
	"client/proto"
	"context"
	"log"

	"google.golang.org/grpc"
)

func main() {
	conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	client := proto.NewPubsubServiceClient(conn)

	_, err = client.Publish(
		context.Background(), &proto.String{Value: "golang: hello Go"},
	)
	if err != nil {
		log.Fatal(err)
	}

	_, err = client.Publish(
		context.Background(), &proto.String{Value: "docker: hello Docker"},
	)
	if nil != err {
		log.Fatal(err)
	}

}

создать новыйNewPubsubServiceClientобъект, а затем передатьclient.Publishспособ публикации сообщения.

Когда весь код написан, давайте откроем три терминала, чтобы протестировать его:

Терминал 1Запустите сервер на:

go run main.go

Терминал 2Запустите клиент подписки на:

go run sub_client.go

Терминал 3Запустите клиент публикации на:

go run pub_client.go

Таким образом, вТерминал 2Вот соответствующий вывод:

subTopic:  golang: hello Go
sub1:  golang: hello Go
sub1:  docker: hello Docker

Вы также можете открыть еще несколько терминалов подписки, тогда каждый терминал подписки будет иметь одинаковый вывод контента.

Адрес источника: GitHub

REST-интерфейс

gRPC обычно используется для внутрикластерной связи, а если нужно предоставлять внешние сервисы, то большинство из них осуществляется через интерфейс REST. Проект с открытым исходным кодом grpc-gateway предоставляет возможность конвертировать службы gRPC в службы REST, таким образом, вы можете получить прямой доступ к gRPC API.

Но я думаю, на самом деле, его следует использовать реже. Если предоставляется интерфейс REST, гораздо удобнее написать службу HTTP напрямую.

прото-файл

Первым шагом является создание прото-файла:

syntax = "proto3";

package proto;

import "google/api/annotations.proto";

message StringMessage {
  string value = 1;
}

service RestService {
    rpc Get(StringMessage) returns (StringMessage) {
        option (google.api.http) = {
            get: "/get/{value}"
        };
    }
    rpc Post(StringMessage) returns (StringMessage) {
        option (google.api.http) = {
            post: "/post"
            body: "*"
        };
    }
}

Определить службу RESTRestServiceсоответственно реализоватьGETа такжеPOSTметод.

Установите плагин:

go get -u github.com/grpc-ecosystem/grpc-gateway/protoc-gen-grpc-gateway

Сгенерируйте соответствующий код:

protoc -I/usr/local/include -I. \
    -I$GOPATH/pkg/mod \
    -I$GOPATH/pkg/mod/github.com/grpc-ecosystem/grpc-gateway@v1.16.0/third_party/googleapis \
    --grpc-gateway_out=. --go_out=plugins=grpc:.\
    --swagger_out=. \
    helloworld.proto

--grpc-gateway_outПараметр может генерировать соответствующий файл gw,--swagger_outПараметр генерирует соответствующую документацию по API.

Два файла, которые я создал здесь, выглядят следующим образом:

helloworld.pb.gw.go
helloworld.swagger.json

ОТДЫХА службы

package main

import (
	"context"
	"log"
	"net/http"

	"rest/proto"

	"github.com/grpc-ecosystem/grpc-gateway/runtime"
	"google.golang.org/grpc"
)

func main() {
	ctx := context.Background()
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	mux := runtime.NewServeMux()

	err := proto.RegisterRestServiceHandlerFromEndpoint(
		ctx, mux, "localhost:50051",
		[]grpc.DialOption{grpc.WithInsecure()},
	)
	if err != nil {
		log.Fatal(err)
	}

	http.ListenAndServe(":8080", mux)
}

Здесь в основном через реализацию файла gw вRegisterRestServiceHandlerFromEndpointспособ подключения к службе gRPC.

служба gRPC

package main

import (
	"context"
	"net"

	"rest/proto"

	"google.golang.org/grpc"
)

type RestServiceImpl struct{}

func (r *RestServiceImpl) Get(ctx context.Context, message *proto.StringMessage) (*proto.StringMessage, error) {
	return &proto.StringMessage{Value: "Get hi:" + message.Value + "#"}, nil
}

func (r *RestServiceImpl) Post(ctx context.Context, message *proto.StringMessage) (*proto.StringMessage, error) {
	return &proto.StringMessage{Value: "Post hi:" + message.Value + "@"}, nil
}

func main() {
	grpcServer := grpc.NewServer()
	proto.RegisterRestServiceServer(grpcServer, new(RestServiceImpl))
	lis, _ := net.Listen("tcp", ":50051")
	grpcServer.Serve(lis)
}

Служба gRPC реализована так же, как и раньше.

Выше приведен весь код, давайте проверим его сейчас:

Запустите три терминала:

Терминал 1Запустите службу gRPC:

go run grpc_service.go

Терминал 2Запустите службу REST:

go run rest_service.go

Терминал 3Чтобы запросить службу REST:

$ curl localhost:8080/get/gopher
{"value":"Get hi:gopher"}

$ curl localhost:8080/post -X POST --data '{"value":"grpc"}'
{"value":"Post hi:grpc"}

Адрес источника: GitHub

Контроль времени ожидания

В последней части представлен контроль тайм-аута, что очень важно.

Общий API веб-сервиса или Nginx установит период тайм-аута.Если период тайм-аута превышен, если данные не возвращаются, сервер может напрямую вернуть ошибку тайм-аута, или клиент может разорвать соединение.

Без этого тайм-аута это довольно опасно. Все запросы блокируются на стороне сервера, что потребляет много ресурсов, например памяти. Если ресурсы исчерпаны, это может даже привести к сбою всей службы.

Итак, как установить тайм-аут в gRPC? в основном по контекстуcontext.Contextпараметры, в частностиcontext.WithDeadlineфункция.

прото-файл

Создайте простейший прото файл, об этом особо нечего сказать.

syntax = "proto3";

package proto;

// The greeting service definition.
service Greeter {
    // Sends a greeting
    rpc SayHello (HelloRequest) returns (HelloReply) {}
}

// The request message containing the user's name.
message HelloRequest {
    string name = 1;
}

// The response message containing the greetings
message HelloReply {
    string message = 1;
}

клиент

package main

import (
	"client/proto"
	"context"
	"fmt"
	"log"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

func main() {
	// 简单调用
	conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())
	defer conn.Close()

	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Duration(3*time.Second)))
	defer cancel()

	client := proto.NewGreeterClient(conn)
	// 简单调用
	reply, err := client.SayHello(ctx, &proto.HelloRequest{Name: "zzz"})
	if err != nil {
		statusErr, ok := status.FromError(err)
		if ok {
			if statusErr.Code() == codes.DeadlineExceeded {
				log.Fatalln("client.SayHello err: deadline")
			}
		}

		log.Fatalf("client.SayHello err: %v", err)
	}
	fmt.Println(reply.Message)
}

Установите тайм-аут 3 с с помощью следующей функции:

ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Duration(3*time.Second)))
defer cancel()

Ошибки тайм-аута затем обнаруживаются в ошибках ответа.

Сервер

package main

import (
	"context"
	"fmt"
	"log"
	"net"
	"runtime"
	"server/proto"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/reflection"
	"google.golang.org/grpc/status"
)

type greeter struct {
}

func (*greeter) SayHello(ctx context.Context, req *proto.HelloRequest) (*proto.HelloReply, error) {
	data := make(chan *proto.HelloReply, 1)
	go handle(ctx, req, data)
	select {
	case res := <-data:
		return res, nil
	case <-ctx.Done():
		return nil, status.Errorf(codes.Canceled, "Client cancelled, abandoning.")
	}
}

func handle(ctx context.Context, req *proto.HelloRequest, data chan<- *proto.HelloReply) {
	select {
	case <-ctx.Done():
		log.Println(ctx.Err())
		runtime.Goexit() //超时后退出该Go协程
	case <-time.After(4 * time.Second): // 模拟耗时操作
		res := proto.HelloReply{
			Message: "hello " + req.Name,
		}
		// //修改数据库前进行超时判断
		// if ctx.Err() == context.Canceled{
		// 	...
		// 	//如果已经超时，则退出
		// }
		data <- &res
	}
}

func main() {
	lis, err := net.Listen("tcp", ":50051")
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}

	// 简单调用
	server := grpc.NewServer()
	// 注册 grpcurl 所需的 reflection 服务
	reflection.Register(server)
	// 注册业务服务
	proto.RegisterGreeterServer(server, &greeter{})

	fmt.Println("grpc server start ...")
	if err := server.Serve(lis); err != nil {
		log.Fatalf("failed to serve: %v", err)
	}
}

Добавить серверhandleфункция, гдеcase <-time.After(4 * time.Second)Указывает, что соответствующий код будет выполнен через 4 с для имитации запроса тайм-аута.

Если время ожидания клиента превышает 4 с, будет сгенерирована ошибка времени ожидания.

Давайте смоделируем это:

Сервер:

$ go run main.go
grpc server start ...
2021/10/24 22:57:40 context deadline exceeded

Клиент:

$ go run main.go
2021/10/24 22:57:40 client.SayHello err: deadline
exit status 1

Адрес источника: GitHub

Суммировать

В этой статье в основном представлены три части фактического боевого содержания gRPC, а именно:

модель публикации-подписки
REST-интерфейс
Контроль времени ожидания

Лично контроль тайм-аута по-прежнему является самым важным, и в обычном процессе разработки требуется больше внимания.

В сочетании с предыдущей статьей фактический контент gRPC был написан, и все коды могут быть выполнены и загружены вGitHub.

Если у вас есть какие-либо вопросы, пожалуйста, оставьте мне сообщение, и если вы чувствуете себя хорошо, пожалуйста, подпишитесь и перешлите его.

Адрес источника:

GitHub.com/Делайте это с душой/идите…
GitHub.com/Делайте это с душой/идите…

Рекомендуемое чтение:

gRPC, большая похвала
Доступ к службам gRPC из командной строки с помощью grpcurl
Я слышал, что 99% программистов на Go столкнулись с defer.

Ссылаться на:

Разрушение 2010. Талант/продвинутый…
код leading.com/article/946…
nuggets.capable/post/684490…
Блог woohoo.cn на.com/fireworks EA…