建站
咨詢
售后
建站咨詢
新聞中心
這里有您想知道的互聯(lián)網(wǎng)營(yíng)銷解決方案
Go實(shí)現(xiàn)分布式高可用后臺(tái):使用gRPC實(shí)現(xiàn)日志微服務(wù)
第一步還是要定義proto文件，修改proglog/api/v1下面的log.proto文件：
為朔州等地區(qū)用戶提供了全套網(wǎng)頁(yè)設(shè)計(jì)制作服務(wù)，及朔州網(wǎng)站建設(shè)行業(yè)解決方案。主營(yíng)業(yè)務(wù)為成都網(wǎng)站制作、成都網(wǎng)站設(shè)計(jì)、朔州網(wǎng)站設(shè)計(jì)，以傳統(tǒng)方式定制建設(shè)網(wǎng)站，并提供域名空間備案等一條龍服務(wù)，秉承以專業(yè)、用心的態(tài)度為用戶提供真誠(chéng)的服務(wù)。我們深信只要達(dá)到每一位用戶的要求，就會(huì)得到認(rèn)可，從而選擇與我們長(zhǎng)期合作。這樣，我們也可以走得更遠(yuǎn)！
syntax = "proto3";
package log.v1;
option go_package = "api/log_v1";
service Log {
rpc Produce(ProduceRequest) returns (ProduceResponse) {}
rpc Consume(ConsumeRequest) returns (ConsumeResponse) {}
rpc ConsumeStream(ConsumeRequest) returns (stream ConsumeResponse){}
rpc ProduceStream(stream ProduceRequest) returns (stream ProduceResponse) {}
}
message Record {
bytes value = 1;
uint64 offset = 2;
}
message ProduceRequest {
Record record = 1;
}
message ProduceResponse {
uint64 offset = 1;
}
message ConsumeRequest {
uint64 offset = 1;
}
message ConsumeResponse {
Record record = 2;
}
代碼的邏輯跟前面幾節(jié)我們嘗試使用gRPC時(shí)的proto文件定義邏輯沒(méi)什么不同，Produce接口是客戶端向服務(wù)端提交一條日志信息，Consume是客戶端向服務(wù)端提交日志編號(hào)，然后服務(wù)端返回日志信息，ConsumeStream是客戶端向服務(wù)端提交一連串的日志編號(hào)，然后服務(wù)端返回一連串的日志信息，ProduceStream是客戶端向服務(wù)端提交一連串的日志信息，然后服務(wù)端返回日志添加后對(duì)應(yīng)的編號(hào)。完成上面proto文件定義后，將它編譯出對(duì)應(yīng)的pb.go文件，這些文件會(huì)放置在api/v1/api_log_v1目錄下，然后我們看看服務(wù)端的邏輯設(shè)計(jì)。
在internal/server下新建server.go文件，首先我們添加依賴模塊，同時(shí)生成gRPC服務(wù)器對(duì)象，并注冊(cè)我們要實(shí)現(xiàn)的接口:
package server
import (
"context"
api "api/v1/api/log_v1"
"google.golang.org/grpc"
)
type commitLog interface {
Append(*api.Record) (uint64, error)
Read(uint64) (*api.Record, error)
}
type Config struct { //實(shí)現(xiàn)依賴注入
CommitLog commitLog
}
var _ api.LogServer = (*grpcServer)(nil) //gRPC服務(wù)器對(duì)象
func NewGRPCServer(config *Config) (*grpc.Server, error) {
gsrv := grpc.NewServer()
srv, err := newgrpcServer(config)
if err != nil {
return nil, err
}
api.RegisterLogServer(gsrv, srv)
return gsrv, nil
}
type grpcServer struct {
api.UnimplementedLogServer
*Config
}
func newgrpcServer(config *Config)(srv *grpcServer, err error) {
srv = &grpcServer { //grpcServer會(huì)實(shí)現(xiàn)proto文件里面定義的接口
Config: config,
}
return srv, nil
}
在上面代碼中有一點(diǎn)需要注意，那就是它使用了常用的設(shè)計(jì)模式叫依賴注入，我們的服務(wù)需要使用到日志模塊提供的功能，但是我們這里只需要知道日志模塊提供的接口，也就是Append和Read，我們不需要關(guān)心它的具體實(shí)現(xiàn)，這樣我們就能實(shí)現(xiàn)邏輯上的解耦合，在啟動(dòng)我們的服務(wù)程序時(shí)，只需要調(diào)用者將實(shí)現(xiàn)了commitLog接口的實(shí)例傳給我們即可，至于接口的實(shí)現(xiàn)細(xì)節(jié)我們不需要關(guān)心，通過(guò)依賴注入這種設(shè)計(jì)模式能夠使得系統(tǒng)設(shè)計(jì)的復(fù)雜度降低，靈活性提升。
接下來(lái)就是對(duì)四個(gè)服務(wù)接口的實(shí)現(xiàn)，其邏輯跟我們前兩節(jié)做的沒(méi)有什么區(qū)別:
func (s *grpcServer) Produce(ctx context.Context,
req *api.ProduceRequest) (*api.ProduceResponse, error){
//收到客戶端發(fā)來(lái)的日志添加請(qǐng)求，然后調(diào)用日志模塊Append接口進(jìn)行添加
offset, err := s.CommitLog.Append(req.Record)
if err != nil {
return nil, err
}
//添加完成后返回日志編號(hào)
return &api.ProduceResponse{Offset: offset}, nil
}
func (s *grpcServer) Consume(ctx context.Context,
req *api.ConsumeRequest)(*api.ConsumeResponse, error) {
//收到客戶端發(fā)來(lái)的日志編號(hào)，返回日志內(nèi)容
record, err := s.CommitLog.Read(req.Offset)
if err != nil {
return nil, err
}
return &api.ConsumeResponse{Record: record}, nil
}
func (s *grpcServer) ProduceStream (stream api.Log_ProduceStreamServer) error {
for {
//客戶端發(fā)來(lái)一系列日志數(shù)據(jù)，服務(wù)端通過(guò)Recv()依次收取，然后將日志進(jìn)行添加
req, err := stream.Recv()
if err != nil {
return err
}
res, err := s.Produce(stream.Context(), req)
if err != nil {
return err
}
if err = stream.Send(res); err != nil {
return err
}
}
}
func (s *grpcServer) ConsumeStream(req *api.ConsumeRequest, stream api.Log_ConsumeStreamServer) error {
for {
//客戶端發(fā)來(lái)一系列日志編號(hào)，服務(wù)端返回一系列與編號(hào)對(duì)應(yīng)的日志內(nèi)容
select {
case <-stream.Context().Done():
//進(jìn)入這里表明客戶端終端了連接
return nil
default:
res, err := s.Consume(stream.Context(), req)
switch err.(type){
case nil:
case api.ErrorOffsetOutOfRange:
continue
default:
return err
}
//將獲得的日志信息發(fā)送給客戶端
if err = stream.Send(res); err != nil {
return err
}
req.Offset++
}
}
}
上面代碼的實(shí)現(xiàn)邏輯與我們前面描述的一模一樣，因此沒(méi)有多少可以探究的地方，最后我們測(cè)試一下上面代碼的實(shí)現(xiàn)，新建server_test.go，添加內(nèi)容如下：
package server
import (
"context"
"io/ioutil"
"net"
"testing"
"github.com/stretchr/testify/require"
api "api/v1/api/log_v1"
"internal/log"
"google.golang.org/grpc"
)
func TestServer(t *testing.T) {
for scenario, fn := range map[string]func(
t *testing.T,
client api.LogClient,
config *Config,
) {
"produce/consume a meesage to/from the log success": testProduceConsume ,
"produce/consume stream success": testProduceConsumeStream,
"consume past log boundary fails: ": testConsumePastBoundary,
} {
t.Run(scenario, func(t *testing.T) {
//在運(yùn)行測(cè)試用例前先創(chuàng)建服務(wù)端對(duì)象
client, config, teardown := setupTest(t, nil)
defer teardown() //關(guān)閉服務(wù)端
fn(t, client, config)
})
}
}
func setupTest(t *testing.T, fn func(*Config)) (client api.LogClient, cfg*Config, teardown func()) {
t.Helper()
//生成tcp連接，使用0意味著使用隨機(jī)端口
l, err := net.Listen("tcp", ":0")
require.NoError(t, err)
clientOptions := []grpc.DialOption{grpc.WithInsecure()}
cc, err := grpc.Dial(l.Addr().String() , clientOptions...)
require.NoError(t, err)
dir, err := ioutil.TempDir("", "server-test")
require.NoError(t, err)
clog, err := log.NewLog(dir, log.Config{})
require.NoError(t, err)
cfg = &Config{
CommitLog: clog,
}
if fn != nil {
fn(cfg)
}
//創(chuàng)建服務(wù)端對(duì)象
server, err := NewGRPCServer(cfg)
require.NoError(t ,err)
go func() {
//啟動(dòng)服務(wù)端
server.Serve(l)
}()
//創(chuàng)建客戶端對(duì)象
client = api.NewLogClient(cc)
return client, cfg, func() {
server.Stop()
cc.Close()
l.Close()
clog.Remove()
}
}
func testProduceConsume(t *testing.T, client api.LogClient, config*Config) {
ctx := context.Background()
want := &api.Record{
Value: []byte("hello world"),
}
//客戶端提交一條日志，然后拿到日志編號(hào)后再用于請(qǐng)求日志內(nèi)容,檢驗(yàn)服務(wù)端返回的日志內(nèi)容與提交的是否一致
produce, err := client.Produce(ctx, &api.ProduceRequest{
Record: want,
})
require.NoError(t, err)
consume, err := client.Consume(ctx, &api.ConsumeRequest{
Offset: produce.Offset,
})
require.NoError(t, err)
require.Equal(t, want.Value, consume.Record.Value)
require.Equal(t, want.Offset, consume.Record.Offset)
}
func testConsumePastBoundary(t *testing.T, client api.LogClient, config *Config) {
ctx := context.Background()
produce, err := client.Produce(ctx, &api.ProduceRequest{
Record: &api.Record {
Value: []byte("hello world"),
},
})
//使用不存在的日志編號(hào)進(jìn)行請(qǐng)求，服務(wù)端應(yīng)該返回相應(yīng)錯(cuò)誤
require.NoError(t, err)
consume, err := client.Consume(ctx, &api.ConsumeRequest{
Offset: produce.Offset + 1,
})
if consume != nil {
t.Fatal("consume not nil")
}
got := grpc.Code(err)
want := grpc.Code(api.ErrorOffsetOutOfRange{}.GRPCStatus().Err())
if got != want {
t.Fatalf("got err: %v, want %v", got, want)
}
}
func testProduceConsumeStream(t *testing.T, client api.LogClient, config *Config) {
ctx := context.Background()
records := []*api.Record{{
Value: []byte("first message"),
Offset: 0,
},
{
Value: []byte("second message"),
Offset: 0,
},
}
//客戶端向服務(wù)端提交多個(gè)日志，獲得多個(gè)日志編號(hào)，然后再提交獲得的編號(hào)，從而讓服務(wù)端返回一系列日志數(shù)據(jù)
//接著比對(duì)服務(wù)端返回的日志內(nèi)容和服務(wù)端是否一致
{
stream, err := client.ProduceStream(ctx)
require.NoError(t, err)
for offset, record := range records {
err = stream.Send(&api.ProduceRequest{
Record: record,
})
require.NoError(t, err)
res, err := stream.Recv()
require.NoError(t, err)
if res.Offset != uint64(offset) {
t.Fatalf("got offset: %d, want: %d", res.Offset, offset,)
}
}
}
{
stream, err := client.ConsumeStream(ctx, &api.ConsumeRequest{Offset: 0},)
require.NoError(t, err)
for i, record := range records{
res, err := stream.Recv()
require.NoError(t, err)
require.Equal(t, res.Record, &api.Record{
Value: record.Value,
Offset: uint64(i),
})
}
}
}
測(cè)試代碼的邏輯通過(guò)注釋就能理解，在測(cè)試用例中，客戶端的創(chuàng)建，數(shù)據(jù)的發(fā)送和接收跟我們前面描述的沒(méi)什么區(qū)別，由此我們依靠gRPC框架就完成了日志服務(wù)，下一節(jié)我們看看gRPC框架提供的數(shù)據(jù)安全功能。
網(wǎng)頁(yè)題目：Go實(shí)現(xiàn)分布式高可用后臺(tái):使用gRPC實(shí)現(xiàn)日志微服務(wù)
本文地址：http://m.fisionsoft.com.cn/article/cccjghi.html
