GOは、高度な並列性と可用性を備えた分散システムを実装します。Logマイクロサービスの実装

この記事はWeChatの公開アカウント「Coding Disney」から転載したもので、著者はChen Yiです。記事の転載についてはCoding Disney公式アカウントまでご連絡ください。

ビッグデータの時代では、高い同時実行性、高い可用性、マイクロサービスシステム設計の理解を備えた人材が強く求められています。バックエンド開発に従事したい場合、JD の説明で最も一般的な要件は、いわゆる「高同時実行」システム開発の経験です。しかし、市場には「高同時実行性」と「高可用性」を直接ターゲットにしたチュートリアルがないことがわかりました。見つかる情報は、多くの場合、ほんの数語であったり、混乱を招く理論を説明しているものであったりします。しかし、技術の習得は実践から生まれなければなりません。長い間探していたのですが、マイクロサービスをベースにした高並列システムの開発を実践できるインストラクターが非常に少ないことがわかりました。したがって、私は自分の学習と実践経験に基づいて、特に分散システム設計の理論と技術を理解し習得するための具体的な実践を重視しながら、この技術を皆さんと共有したいと考えています。

いわゆる「マイクロサービス」には魔法のようなものは何もありません。もともと集約していたモジュールを、サーバープログラムの存在に基づいて複数の独立した形式に分解するだけです。開発したバックエンドシステムが、ログ、ストレージ、ビジネスロジック、アルゴリズムロジックなどのモジュールに分割されているとします。以前は、これらのモジュールが 1 つに集約され、複雑で大規模なアプリケーションが形成されていました。

このアプローチには多くの問題があります。 1 つ目は、モジュールをあまりに多く組み合わせると、モジュール間にさまざまな論理的な結合があるためにシステム設計が複雑になりすぎて、時間の経過とともにシステムの開発と保守がますます困難になることです。 2つ目は、システムがますます脆弱になっていることです。モジュールの 1 つがエラーを送信したりクラッシュしたりすると、システム全体が崩壊する可能性があります。 3 つ目は、スケーラビリティが強くなく、ハードウェアの性能向上によってシステムが対応する拡張を実現することが難しいことです。

高い同時実行性と高可用性を実現するための基本的な考え方は、モジュールを分解して独立したサーバープログラムにすることです。モジュールはメッセージを送信することで相互に連携します。

このモデルの利点は次のとおりです。1. モジュールは分離されており、1 つのモジュールに問題が発生してもシステム全体にほとんど影響はありません。 2. スケーラビリティと高可用性。異なるサーバーにモジュールを展開できます。トラフィックが増加した場合は、サーバーの数を増やすだけで、システムの応答性を同様に拡張できます。 3. 堅牢性が向上します。複数のモジュールをバックアップできるため、1 つのモジュールに問題が発生した場合、要求を他の同様のモジュールにリダイレクトすることができ、システムの信頼性が大幅に向上します。

もちろん、いかなる利益にもそれに応じた代償が伴います。分散システムの設計と開発には、元の集約システムよりもはるかに多くの困難が伴います。たとえば、負荷分散、サービス検出、モジュールネゴシエーション、コンセンサスなどです。分散アルゴリズムはこれらの問題の解決を重視しますが、理論は常に抽象的で理解するのが困難です。高可用性と高同時実行性を備えたシステムを実装できない場合、いくら理論を読んでも混乱してしまいます。したがって、実践を通じて理論を理解し、習得する必要があります。まず、最も単純なサービスであるログサービスから始めます。実装にはGOを使用します。

まず、go_distributed_system という名前のルートディレクトリを作成します。以降のすべてのサービスモジュールはこのディレクトリに実装されます。次にサブディレクトリ proglog を作成します。入力後、サブディレクトリ internal/server/ を作成します。ここでは、ログサービスの論理モジュールを実装します。まず、internal/server の下で初期化コマンドを実行します。

 go mod init 内部/サーバー

ここで開発されたモジュールは他のモジュールから参照されるため、mod ファイルを作成する必要があります。まず、ログシステムに必要な基礎データ構造を完成させ、log.go ファイルを作成する必要があります。対応するコードは次のとおりです。

パッケージサーバー
 
輸入 （
 「fmt」  
 「同期」  
 ） 
 
型 Log 構造体 {
    mu sync.Mutex
    記録 [] 記録
} 
 
 NewLog()関数 *Log {
 &Log{ch : make(chan Record),}を返します。
 } 
 
 func(c *Log) Append(record レコード) (uint64, error) {
     c.mu.ロック()
    c.mu.Unlock() を延期する
    レコード.オフセット = uint64(len(c.records))
    c.records = 追加(c.records、レコード)
 record.Offset, nil を返す
} 
 
 func (c *Log) Read (offset uint64)(Record, error) {
    c.mu.ロック()
    c.mu.Unlock() を延期する
    オフセット >= uint64(len(c.records)) の場合 {
レコード{}、ErrOffsetNotFoundを返す
    } 
 
 c.records[オフセット]、nilを返す
} 
 
レコード構造体型{
    値 []byte `json: "value" `
    オフセット uint64 `json: "offset" `
 } 
 
 var ErrOffsetNotFound = fmt.Errorf( "オフセットが見つかりません" )

ログサービスは、ログの読み取りおよび書き込み要求を http サーバープログラムの形式で受信するため、複数の読み取りまたは書き込み要求が同時に実行されるため、レコード配列に対して排他操作を実行する必要があります。したがって、レコード配列の各読み取りの前に、ミューテックスロックを使用してロックを取得します。これにより、複数の読み取りおよび書き込み要求を同時に受信したときに、サービスがデータの一貫性を損なうことを防ぐことができます。

すべてのログの読み取りおよび書き込み要求は http POST および GET の形式で開始され、データは json にカプセル化されるため、以下の http サーバーオブジェクトを作成し、新しいファイル http.go を作成し、次のコードを完成させます。

パッケージサーバー
 
輸入 （
 「エンコーディング/json」  
 「ネット/http」  
 「github.com/gorilla/mux」  
 ） 
 
 func NewHttpServer(addr string) *http.Server {
    httpsrv := 新しいHttpServer()
    r := mux.NewRouter()
    r.HandleFunc( "/" 、 httpsrv.handleLogWrite).Methods( "POST" )
    r.HandleFunc( "/" 、 httpsrv.hadnleLogRead).Methods( "GET" ) 
 
 &http.Server{を返す
        住所:住所、
        ハンドラ: r,
    }
 } 
 
タイプ httpServer 構造体 {
    ログ *ログ
} 
 
関数newHttpServer() *httpServer {
 &httpServer {を返す
        ログ: NewLog(),
    }
 } 
 
 WriteRequest構造体型{
    レコード レコード `json: "record" `
 } 
 
 WriteResponse構造体型{
    オフセット uint64 `json: "offset" `
 } 
 
 ReadRequest構造体型{
    オフセット uint64 `json: "offset" `
 } 
 
 ReadResponse構造体型{
    レコード レコード `json: "record" `
 } 
 
 func (s *httpServer) handleLogWrite(w http.ResponseWriter, r * http.Request) {
    var req 書き込みリクエスト
    //サービスはjson形式でリクエストを受け取ります
    エラー:= json.NewDecoder(r.Body).Decode(&req)
    err != nil の場合 {
        http.Error(w, err.Error(), http.StatusBadRequest)
戻る  
    } 
 
オフ、エラー:= s.Log.Append(req.Record)
    err != nil の場合 {
        エラー:
戻る  
    } 
 
    res := WriteResponse{オフセット:オフ}
    //サービスは結果をjson形式で返します
    エラー = json.NewEncoder(w).Encode(res)
    err != nil の場合 {
        エラー:
戻る  
    }
 } 
 
 func (s *httpServer) hadnleLogRead(w http.ResponseWriter, r *http.Request) {
    var req 読み取りリクエスト
    エラー:= json.NewDecoder(r.Body).Decode(&req)
    err != nil の場合 {
        http.Error(w, err.Error(), http.StatusBadRequest)
戻る  
    } 
 
    レコード、err := s.Log。読み取り(req.Offset)
    err == ErrOffsetNotFound の場合 {
        http.Error(w, err.Error(), http.StatusNotFound)
戻る 
    } 
 
    err != nil の場合 {
        エラー:
戻る  
    } 
 
    res := ReadResponse{レコード: レコード}
    エラー = json.NewEncoder(w).Encode(res)
    err != nil の場合 {
        エラー:
戻る 
    }
 }

上記のコードは、「分散」と「マイクロサービス」の特徴を示しています。対応する機能コードは別のサーバーの形で実行され、ネットワークを介してサービス要求を受信します。これは「分散型」に相当します。それぞれの独立したモジュールは、「マイクロサービス」に対応する特定のタスクのみを完了します。この方法は、異なるマシン上で同時に実行できるため、「スケーラビリティ」を実現します。

同時に、サービスは http サーバーの形式で存在するため、サービス要求と戻りも Http 形式である必要があり、データは Json でカプセル化されます。同時に実装されるロジックは非常にシンプルです。ログ書き込み要求がある場合、要求をレコード構造に解析し、キューの末尾に追加します。ログの読み取り要求があった場合、クライアントから送信された読み取りオフセットを取得し、対応するレコードを取り出して JSON 形式にカプセル化し、顧客に返します。

サーバーコードが完成したら、サーバーを実行する必要があります。モジュール化を実現するために、サーバーの起動を別の場所に配置しました。 proglog ルートディレクトリに cmd/server を作成し、そこに main.go を追加します。

パッケージメイン
 
輸入 （
 "ログ"  
 「内部/サーバー」  
 ） 
 
関数main() {
    srv := server.NewHttpServer( ":8080" )
    ログに記録されます。致命的です(srv.ListenAndServe())
 }

同時に、internal/server の下のモジュールを参照するには、go mod init cmd/server を使用して cmd/server の下で初期化し、go.mod ファイルに次の行を追加する必要があります。

 internal/server を ../../internal/server に置き換えます

次に、go mod tidy コマンドを実行して、ローカルモジュールが指定されたディレクトリ変換に従ってモジュールを参照するようにします。最後に、go run main.go を使用してログサービスを開始します。ここで、サーバーの可用性をテストする必要があります。また、ディレクトリ内にserver_test.goを作成し、テストコードを記述します。基本的なロジックは、サーバーにログ書き込み要求を送信し、次に読み取り要求を送信して、読み取られたデータが書き込んだデータと一致しているかどうかを比較することです。コードは次のとおりです。

パッケージメイン
 
輸入（
 「エンコーディング/json」  
 「ネット/http」  
 「内部/サーバー」  
 「バイト」  
 「テスト」  
 「io/ioutil」  
 ） 
 
関数 TestServerLogWrite(t *testing.T) {
    var テスト = [] 構造体 {
        リクエストサーバー.WriteRequest
        want_response サーバー.WriteResponse
    } {
        {リクエスト: server.WriteRequest{server.Record{[]byte(`これはログリクエスト1です`), 0}},
         want_response: server.WriteResponse{オフセット: 0, },},
         {リクエスト: server.WriteRequest{server.Record{[]byte(`これはログリクエスト2です`), 0}},
         want_response: server.WriteResponse{オフセット: 1, },},
         {リクエスト: server.WriteRequest{server.Record{[]byte(`これはログリクエスト3です`), 0}},
         want_response: server.WriteResponse{オフセット: 2, },},
    } 
 
 _の場合、テスト:=範囲テスト{
        //リクエストをJSON形式に変換し、ログサービスに投稿します
        リクエスト:= &test.request
        request_json、エラー:= json.Marshal(リクエスト)
        err != nil の場合 {
            t.Errorf( "リクエストをJSONに変換できませんでした" )
戻る  
        } 
 
        応答、エラー:= http.Post( "http://localhost:8080" 、 "application/json" 、bytes.NewBuffer(request_json))
        延期resp.Body.Close ()
        err != nil の場合 {
            t.Errorf( "http post リクエスト失敗: %v" , err)
戻る 
        } 
 
        //ログサービスの戻り結果を解析する
        本文、エラー:= ioutil.ReadAll(resp.Body)
        var レスポンス サーバー.WriteResponse
        err = json.Unmarshal([]byte(body), &response)
        err != nil の場合 {
            t.Errorf( "書き込み応答のアンマーシャルに失敗しました: %v" , err)
        } 
 
        // 結果が期待される結果と一致しているかどうかを確認します
        response.Offset != test.want_response.Offset {の場合
            t.Errorf( "オフセット: %d を取得しましたが、必要なオフセット: %d" 、 response.Offset、 test.want_response.Offset)
        } 
 
    } 
 
    var read_tests = []構造体{
        リクエストサーバー.ReadRequest
        server.ReadResponse が必要
    } {
        {リクエスト: server.ReadRequest{オフセット: 0,},
        欲しいもの: server.ReadResponse{server.Record{[]byte(`これはログリクエスト1です`), 0}} },
        {リクエスト: server.ReadRequest{オフセット: 1,},
        欲しいもの: server.ReadResponse{server.Record{[]byte(`これはログリクエスト2です`), 0}} },
        {リクエスト: server.ReadRequest{オフセット: 2,},
        欲しいもの: server.ReadResponse{server.Record{[]byte(`これはログリクエスト3です`), 0}} },
    } 
 
 _の場合、テスト:=範囲read_tests{
        リクエスト:= test.request
        request_json 、エラー:= json.Marshal(リクエスト)
        err != nil の場合 {
            t.Errorf( "読み取り要求を JSON に変換できませんでした" )
戻る  
        } 
 
        //リクエストをjsonに変換し、GETリクエスト本体に挿入します
        クライアント:= &http.Client{}
        要求、エラー:= http.NewRequest(http.MethodGet、 "http://localhost:8080" 、bytes.NewBuffer(request_json))
        必須ヘッダーを設定します( "Content-Type" 、 "application/json" )
        応答、エラー:= client.Do(req)
        err != nil の場合 {
            t.Errorf( "読み取り要求失敗: %v" , err)
戻る  
        } 
 
        //読み取り要求によって返された結果を解析する
        延期resp.Body.Close ()
        本文、エラー:= ioutil.ReadAll(resp.Body)
        var 応答サーバー.ReadResponse
        err = json.Unmarshal([]byte(body), &response)
        err != nil の場合 {
            t.Errorf( "読み取り応答のアンマーシャルに失敗しました: %v" , err)
戻る  
        } 
 
        res := bytes.Compare(response.Record.Value、test.want.Record.Value) を返します。
        res != 0 の場合 {
            t.Errorf( "値: %q を取得しましたが、必要な値は : %q" 、 response.Record.Value、 test.want.Record.Value)
        }
    } 
 
 }

上記のコードが完成したら、go test を使用して実行します。結果は以下のようになります。

結果から、テストが合格したことがわかります。つまり、ログサービスに書き込み要求を送信しても、読み取り要求を送信しても、得られる結果は期待どおりになります。要約すると、このセクションでは、JSON ベースの http 書き込み要求と読み取り要求を受信できるシンプルな JSON/HTTP ログサービスを設計しました。今後はgPRC技術をベースにしたマイクロサービス開発技術についても学習していきます。

コードを取得する

https://github.com/wycl16514/golang_distribute_system_log_service.git

<<: Techo Hubテクノロジーツアー北京駅がオープン、多くの専門家が「デジタル金融イノベーションと実践」の饗宴に参加

>>: 企業向けマルチクラウドコンピューティングのメリットとデメリット

$locvps: "618"、30% 割引、618 をトップアップすると 150 がもらえます、VPS オプション: 香港 cn2\ドイツ cn2\米国 cn2$