GO 實現高并發高可用分布式系統:Log微服務的實現
本文轉載自微信公眾號「Coding迪斯尼」，作者陳屹。轉載本文請聯系Coding迪斯尼公眾號。
在大數據時代，具備高并發，高可用，理解微服務系統設計的人員需求很大，如果你想從事后臺開發，在JD的描述中最常見的要求就是有所謂的“高并發”系統開發經驗。但我發現在市面上并沒有直接針對“高并發”，“高可用”的教程，你搜到的資料往往都是只言片語，要不就是闡述那些令人摸不著頭腦的理論。但是技術的掌握必須從實踐中來，我找了很久發現很少有指導人動手實踐基于微服務的高并發系統開發，因此我希望結合自己的學習和實踐經驗跟大家分享一下這方面的技術，特別是要強調具體的動手實踐來理解和掌握分布式系統設計的理論和技術。
所謂“微服務”其實沒什么神奇的地方，它只不過是把我們原來聚合在一起的模塊分解成多個獨立的，基于服務器程序存在的形式，假設我們開發的后臺系統分為日志，存儲，業務邏輯，算法邏輯等模塊，以前這些模塊會聚合成一個整體形成一個復雜龐大的應用程序：
這種方式存在很多問題，第一是過多模塊糅合在一起會使得系統設計過于復雜，因為模塊直接存在各種邏輯耦合，這使得隨著時間的推移，系統的開發和維護變得越來越困難。第二是系統越來越脆弱，只要其中一個模塊發送錯誤或奔潰，整個系統可能就會垮塌。第三是可擴展性不強，系統很難通過硬件性能的增強而實現相應擴展。
要實現高并發，高可用，其基本思路就是將模塊拆解，然后讓他們成為獨立運行的服務器程序，各個模塊之間通過消息發送的方式完成配合：
這種模式的好處在于：1，模塊之間解耦合，一個模塊出問題對整個系統影響很小。2，可擴展，高可用，我們可以將模塊部署到不同服務器上，當流量增加，我們只要簡單的增加服務器數量就能使得系統的響應能力實現同等擴展。3，魯棒性增強，由于模塊能備份多個，其中一個模塊出問題，請求可以重定向到其他同樣模塊，于是系統的可靠性能大大增強。
當然任何收益都有對應代價，分布式系統的設計開發相比于原來的聚合性系統會多出很多難點。例如負載均衡，服務發現，模塊協商，共識達成等，分布式算法強調的就是這些問題的解決，但是理論總是抽象難以理解，倘若不能動手實現一個高可用高并發系統，你看多少理論都是霧里看花，越看越糊涂，所以我們必須通過動手實踐來理解和掌握理論，首先我們從最簡單的服務入手，那就是日志服務，我們將使用GO來實現。
首先創建根目錄，可以命名為go_distributed_system，后面所有服務模塊都實現在該目錄下，然后創建子目錄proglog,進去后我們再創建子目錄internel/server/在這里我們實現日志服務的邏輯模塊，首先在internel/server下面執行初始化命令：
- go mod init internal/server
這里開發的模塊會被其他模塊引用，所以我們需要創建mod文件。首先我們需要完成日志系統所需的底層數據結構，創建log.go文件，相應代碼如下：
- package server
- import (
- "fmt"
- "sync"
- )
- type Log struct {
- mu sync.Mutex
- records [] Record
- }
- func NewLog() *Log {
- return &Log{ch : make(chan Record),}
- }
- func(c *Log) Append(record Record) (uint64, error) {
- c.mu.Lock()
- defer c.mu.Unlock()
- record.Offset = uint64(len(c.records))
- c.records = append(c.records, record)
- return record.Offset, nil
- }
- func (c *Log) Read(offset uint64)(Record, error) {
- c.mu.Lock()
- defer c.mu.Unlock()
- if offset >= uint64(len(c.records)) {
- return Record{}, ErrOffsetNotFound
- }
- return c.records[offset], nil
- }
- type Record struct {
- Value []byte `json:"value"`
- Offset uint64 `json:"offset"`
- }
- var ErrOffsetNotFound = fmt.Errorf("offset not found")
由于我們的日志服務將以http服務器程序的方式接收日志讀寫請求，因此多個讀或寫請求會同時執行，所以我們需要對records數組進行互斥操作，因此使用了互斥鎖，在每次讀取records數組前先獲得鎖，這樣能防止服務在同時接收多個讀寫請求時破壞掉數據的一致性。
所有的日志讀寫請求會以http POST 和 GET的方式發起，數據通過json來封裝，所以我們下面將創建一個http服務器對象，新建文件http.go，完成如下代碼：
- package server
- import (
- "encoding/json"
- "net/http"
- "github.com/gorilla/mux"
- )
- func NewHttpServer(addr string) *http.Server {
- httpsrv := newHttpServer()
- r := mux.NewRouter()
- r.HandleFunc("/", httpsrv.handleLogWrite).Methods("POST")
- r.HandleFunc("/", httpsrv.hadnleLogRead).Methods("GET")
- return &http.Server{
- Addr : addr,
- Handler: r,
- }
- }
- type httpServer struct{
- Log *Log
- }
- func newHttpServer() *httpServer {
- return &httpServer {
- Log: NewLog(),
- }
- }
- type WriteRequest struct {
- Record Record `json:"record"`
- }
- type WriteResponse struct {
- Offset uint64 `json:"offset"`
- }
- type ReadRequest struct {
- Offset uint64 `json:"offset"`
- }
- type ReadResponse struct {
- Record Record `json:"record"`
- }
- func (s *httpServer) handleLogWrite(w http.ResponseWriter, r * http.Request) {
- var req WriteRequest
- //服務以json格式接收請求
- err := json.NewDecoder(r.Body).Decode(&req)
- if err != nil {
- http.Error(w, err.Error(), http.StatusBadRequest)
- return
- }
- off, err := s.Log.Append(req.Record)
- if err != nil {
- http.Error(w, err.Error(), http.StatusInternalServerError)
- return
- }
- res := WriteResponse{Offset: off}
- //服務以json格式返回結果
- err = json.NewEncoder(w).Encode(res)
- if err != nil {
- http.Error(w, err.Error(), http.StatusInternalServerError)
- return
- }
- }
- func (s *httpServer) hadnleLogRead(w http.ResponseWriter, r *http.Request) {
- var req ReadRequest
- err := json.NewDecoder(r.Body).Decode(&req)
- if err != nil {
- http.Error(w, err.Error(), http.StatusBadRequest)
- return
- }
- record, err := s.Log.Read(req.Offset)
- if err == ErrOffsetNotFound {
- http.Error(w, err.Error(), http.StatusNotFound)
- return
- }
- if err != nil {
- http.Error(w, err.Error(), http.StatusInternalServerError)
- return
- }
- res := ReadResponse{Record: record}
- err = json.NewEncoder(w).Encode(res)
- if err != nil {
- http.Error(w, err.Error(), http.StatusInternalServerError)
- return
- }
- }
上面代碼顯示出“分布式”，“微服務”的特點。相應的功能代碼以單獨服務器的形式運行，通過網絡來接收服務請求，這對應“分布式”，每個獨立模塊只完成一個特定任務，這就對應“微服務”，由于這種方式可以同時在不同的機器上運行，于是展示了“可擴展性”。
同時服務既然以http 服務器的形式存在，因此服務的請求和返回也要走Http形式，同時數據以Json方式進行封裝。同時實現的邏輯很簡單，但有日志寫請求時，我們把請求解析成Record結構體后加入到隊列末尾，當有讀取日志的請求時，我們獲得客戶端發來的讀取偏移，然后取出對應的記錄，封裝成json格式后返回給客戶。
完成了服務器的代碼后，我們需要將服務器運行起來，為了達到模塊化的目的，我們把服務器的啟動放置在另一個地方，在proglog根目錄下創建cmd/server,在里面添加main.go:
- package main
- import (
- "log"
- "internal/server"
- )
- func main() {
- srv := server.NewHttpServer(":8080")
- log.Fatal(srv.ListenAndServe())
- }
同時為了能夠引用internal/server下面的模塊，我們需要在cmd/server下先通過go mod init cmd/server進行初始化，然后在go.mod文件中添加如下一行：
- replace internal/server => ../../internal/server
然后執行命令 go mod tidy,這樣本地模塊就知道根據給定的目錄轉換去引用模塊，最后使用go run main.go啟動日志服務，現在我們要做的是測試服務器的可用性，我們同樣在目錄下創建server_test.go，然后編寫測試代碼，基本邏輯就是想服務器發送日志寫請求，然后再發送讀請求，并比較讀到的數據是否和我們寫入的數據一致，代碼如下：
- package main
- import(
- "encoding/json"
- "net/http"
- "internal/server"
- "bytes"
- "testing"
- "io/ioutil"
- )
- func TestServerLogWrite(t *testing.T) {
- var tests = []struct {
- request server.WriteRequest
- want_response server.WriteResponse
- } {
- {request: server.WriteRequest{server.Record{[]byte(`this is log request 1`), 0}},
- want_response: server.WriteResponse{Offset: 0, },},
- {request: server.WriteRequest{server.Record{[]byte(`this is log request 2`), 0}},
- want_response: server.WriteResponse{Offset: 1, },},
- {request: server.WriteRequest{server.Record{[]byte(`this is log request 3`), 0}},
- want_response: server.WriteResponse{Offset: 2, },},
- }
- for _, test := range tests {
- //將請求轉換成json格式并post給日志服務
- request := &test.request
- request_json, err := json.Marshal(request)
- if err != nil {
- t.Errorf("convert request to json fail")
- return
- }
- resp, err := http.Post("http://localhost:8080", "application/json",bytes.NewBuffer(request_json))
- defer resp.Body.Close()
- if err != nil {
- t.Errorf("http post request fail: %v", err)
- return
- }
- //解析日志服務返回結果
- body, err := ioutil.ReadAll(resp.Body)
- var response server.WriteResponse
- err = json.Unmarshal([]byte(body), &response)
- if err != nil {
- t.Errorf("Unmarshal write response fail: %v", err)
- }
- //檢測結果是否與預期一致
- if response.Offset != test.want_response.Offset {
- t.Errorf("got offset: %d, but want offset: %d", response.Offset, test.want_response.Offset)
- }
- }
- var read_tests = []struct {
- request server.ReadRequest
- want server.ReadResponse
- } {
- {request: server.ReadRequest{Offset : 0,},
- want: server.ReadResponse{server.Record{[]byte(`this is log request 1`), 0}} },
- {request: server.ReadRequest{Offset : 1,},
- want: server.ReadResponse{server.Record{[]byte(`this is log request 2`), 0}} },
- {request: server.ReadRequest{Offset : 2,},
- want: server.ReadResponse{server.Record{[]byte(`this is log request 3`), 0}} },
- }
- for _, test := range read_tests {
- request := test.request
- request_json , err := json.Marshal(request)
- if err != nil {
- t.Errorf("convert read request to json fail")
- return
- }
- //將請求轉換為json并放入GET請求體
- client := &http.Client{}
- req, err := http.NewRequest(http.MethodGet, "http://localhost:8080", bytes.NewBuffer(request_json))
- req.Header.Set("Content-Type", "application/json")
- resp, err := client.Do(req)
- if err != nil {
- t.Errorf("read request fail: %v", err)
- return
- }
- //解析讀請求返回的結果
- defer resp.Body.Close()
- body, err := ioutil.ReadAll(resp.Body)
- var response server.ReadResponse
- err = json.Unmarshal([]byte(body), &response)
- if err != nil {
- t.Errorf("Unmarshal read response fail: %v", err)
- return
- }
- res := bytes.Compare(response.Record.Value, test.want.Record.Value)
- if res != 0 {
- t.Errorf("got value: %q, but want value : %q", response.Record.Value, test.want.Record.Value)
- }
- }
- }
完成上面代碼后，使用go test運行，結果如下圖所示：
從結果看到，我們的測試能通過，也就是無論是向日志服務提交寫入請求還是讀取請求，所得的結果跟我們預想的一致?？偨Y一下，本節我們設計了一個簡單的JSON/HTTP 日志服務，它能夠接收基于JSON的http寫請求和讀請求，后面我們還會研究基于gPRC技術的微服務開發技術.
代碼獲取
https://github.com/wycl16514/golang_distribute_system_log_service.git
