GO实现高并发高可用分布式系统:Log微服务的实现

GO 实现高并发高可用分布式系统:Log微服务的实现

作者：陈屹 2022-01-10 19:45:40

开发

架构

分布式 在大数据时代，具备高并发，高可用，理解微服务系统设计的人员需求很大，如果你想从事后台开发，在JD的描述中最常见的要求就是有所谓的“高并发”系统开发经验。

创新互联建站10多年企业网站制作服务;为您提供网站建设,网站制作,网页设计及高端网站定制服务,企业网站制作及推广,对成都酒楼设计等多个方面拥有多年的网站营销经验的网站建设公司。

本文转载自微信公众号「Coding迪斯尼」，作者陈屹。转载本文请联系Coding迪斯尼公众号。

在大数据时代，具备高并发，高可用，理解微服务系统设计的人员需求很大，如果你想从事后台开发，在JD的描述中最常见的要求就是有所谓的“高并发”系统开发经验。但我发现在市面上并没有直接针对“高并发”，“高可用”的教程，你搜到的资料往往都是只言片语，要不就是阐述那些令人摸不着头脑的理论。但是技术的掌握必须从实践中来，我找了很久发现很少有指导人动手实践基于微服务的高并发系统开发，因此我希望结合自己的学习和实践经验跟大家分享一下这方面的技术，特别是要强调具体的动手实践来理解和掌握分布式系统设计的理论和技术。

所谓“微服务”其实没什么神奇的地方，它只不过是把我们原来聚合在一起的模块分解成多个独立的，基于服务器程序存在的形式，假设我们开发的后台系统分为日志，存储，业务逻辑，算法逻辑等模块，以前这些模块会聚合成一个整体形成一个复杂庞大的应用程序：

这种方式存在很多问题，第一是过多模块糅合在一起会使得系统设计过于复杂，因为模块直接存在各种逻辑耦合，这使得随着时间的推移，系统的开发和维护变得越来越困难。第二是系统越来越脆弱，只要其中一个模块发送错误或奔溃，整个系统可能就会垮塌。第三是可扩展性不强，系统很难通过硬件性能的增强而实现相应扩展。

要实现高并发，高可用，其基本思路就是将模块拆解，然后让他们成为独立运行的服务器程序，各个模块之间通过消息发送的方式完成配合：

这种模式的好处在于：1，模块之间解耦合，一个模块出问题对整个系统影响很小。2，可扩展，高可用，我们可以将模块部署到不同服务器上，当流量增加，我们只要简单的增加服务器数量就能使得系统的响应能力实现同等扩展。3，鲁棒性增强，由于模块能备份多个，其中一个模块出问题，请求可以重定向到其他同样模块，于是系统的可靠性能大大增强。

当然任何收益都有对应代价，分布式系统的设计开发相比于原来的聚合性系统会多出很多难点。例如负载均衡，服务发现，模块协商，共识达成等，分布式算法强调的就是这些问题的解决，但是理论总是抽象难以理解，倘若不能动手实现一个高可用高并发系统，你看多少理论都是雾里看花，越看越糊涂，所以我们必须通过动手实践来理解和掌握理论，首先我们从最简单的服务入手，那就是日志服务，我们将使用GO来实现。

首先创建根目录，可以命名为go_distributed_system，后面所有服务模块都实现在该目录下，然后创建子目录proglog,进去后我们再创建子目录internel/server/在这里我们实现日志服务的逻辑模块，首先在internel/server下面执行初始化命令：

  
 
 
 
    
   
  
  
  
  1. go mod init internal/server
    2.

这里开发的模块会被其他模块引用，所以我们需要创建mod文件。首先我们需要完成日志系统所需的底层数据结构，创建log.go文件，相应代码如下：

  
 
 
 
    
   
  
  
  
  1. package server
    2. 
   
  
  
  
  2. 
   
  
  
  
  3. import (
    4. 
   
  
  
  
  4.     "fmt"
    5. 
   
  
  
  
  5.     "sync"
    6. 
   
  
  
  
  6. )
    7. 
   
  
  
  
  7. 
   
  
  
  
  8. type Log struct {
    9. 
   
  
  
  
  9.     mu sync.Mutex
    10. 
   
  
  
  
  10.     records [] Record 
    11. 
   
  
  
  
  11. }
    12. 
   
  
  
  
  12. 
   
  
  
  
  13. func NewLog() *Log {
    14. 
   
  
  
  
  14.     return &Log{ch : make(chan Record),} 
    15. 
   
  
  
  
  15. }
    16. 
   
  
  
  
  16. 
   
  
  
  
  17. func(c *Log) Append(record Record) (uint64, error) {
    18. 
   
  
  
  
  18.      c.mu.Lock()
    19. 
   
  
  
  
  19.     defer c.mu.Unlock()
    20. 
   
  
  
  
  20.     record.Offset = uint64(len(c.records))
    21. 
   
  
  
  
  21.     c.records = append(c.records, record)
    22. 
   
  
  
  
  22.     return record.Offset, nil 
    23. 
   
  
  
  
  23. }
    24. 
   
  
  
  
  24. 
   
  
  
  
  25. func (c *Log) Read(offset uint64)(Record, error) {
    26. 
   
  
  
  
  26.     c.mu.Lock()
    27. 
   
  
  
  
  27.     defer c.mu.Unlock()
    28. 
   
  
  
  
  28.     if offset >= uint64(len(c.records)) {
    29. 
   
  
  
  
  29.         return Record{}, ErrOffsetNotFound 
    30. 
   
  
  
  
  30.     }
    31. 
   
  
  
  
  31. 
   
  
  
  
  32.     return c.records[offset], nil 
    33. 
   
  
  
  
  33. }
    34. 
   
  
  
  
  34. 
   
  
  
  
  35. type Record struct {
    36. 
   
  
  
  
  36.     Value []byte `json:"value"`
    37. 
   
  
  
  
  37.     Offset uint64 `json:"offset"`
    38. 
   
  
  
  
  38. }
    39. 
   
  
  
  
  39. 
   
  
  
  
  40. var ErrOffsetNotFound = fmt.Errorf("offset not found")
    41.

由于我们的日志服务将以http服务器程序的方式接收日志读写请求，因此多个读或写请求会同时执行，所以我们需要对records数组进行互斥操作，因此使用了互斥锁，在每次读取records数组前先获得锁，这样能防止服务在同时接收多个读写请求时破坏掉数据的一致性。

所有的日志读写请求会以http POST 和 GET的方式发起，数据通过json来封装，所以我们下面将创建一个http服务器对象，新建文件http.go，完成如下代码：

  
 
 
 
    
   
  
  
  
  1. package server 
    2. 
   
  
  
  
  2. 
   
  
  
  
  3. import (
    4. 
   
  
  
  
  4.     "encoding/json"
    5. 
   
  
  
  
  5.     "net/http"
    6. 
   
  
  
  
  6.     "github.com/gorilla/mux"
    7. 
   
  
  
  
  7. )
    8. 
   
  
  
  
  8. 
   
  
  
  
  9. func NewHttpServer(addr string) *http.Server {
    10. 
   
  
  
  
  10.     httpsrv := newHttpServer()
    11. 
   
  
  
  
  11.     r := mux.NewRouter()
    12. 
   
  
  
  
  12.     r.HandleFunc("/", httpsrv.handleLogWrite).Methods("POST")
    13. 
   
  
  
  
  13.     r.HandleFunc("/", httpsrv.hadnleLogRead).Methods("GET")
    14. 
   
  
  
  
  14. 
   
  
  
  
  15.     return &http.Server{
    16. 
   
  
  
  
  16.         Addr : addr,
    17. 
   
  
  
  
  17.         Handler: r,
    18. 
   
  
  
  
  18.     }
    19. 
   
  
  
  
  19. }
    20. 
   
  
  
  
  20. 
   
  
  
  
  21. type httpServer struct{
    22. 
   
  
  
  
  22.     Log *Log 
    23. 
   
  
  
  
  23. }
    24. 
   
  
  
  
  24. 
   
  
  
  
  25. func newHttpServer() *httpServer {
    26. 
   
  
  
  
  26.     return &httpServer {
    27. 
   
  
  
  
  27.         Log: NewLog(),
    28. 
   
  
  
  
  28.     }
    29. 
   
  
  
  
  29. }
    30. 
   
  
  
  
  30. 
   
  
  
  
  31. type WriteRequest struct {
    32. 
   
  
  
  
  32.     Record Record `json:"record"`
    33. 
   
  
  
  
  33. }
    34. 
   
  
  
  
  34. 
   
  
  
  
  35. type WriteResponse struct {
    36. 
   
  
  
  
  36.     Offset uint64 `json:"offset"`
    37. 
   
  
  
  
  37. }
    38. 
   
  
  
  
  38. 
   
  
  
  
  39. type ReadRequest struct {
    40. 
   
  
  
  
  40.     Offset uint64 `json:"offset"`
    41. 
   
  
  
  
  41. }
    42. 
   
  
  
  
  42. 
   
  
  
  
  43. type ReadResponse struct {
    44. 
   
  
  
  
  44.     Record Record `json:"record"`
    45. 
   
  
  
  
  45. }
    46. 
   
  
  
  
  46. 
   
  
  
  
  47. func (s *httpServer) handleLogWrite(w http.ResponseWriter, r * http.Request) {
    48. 
   
  
  
  
  48.     var req WriteRequest 
    49. 
   
  
  
  
  49.     //服务以json格式接收请求
    50. 
   
  
  
  
  50.     err := json.NewDecoder(r.Body).Decode(&req)
    51. 
   
  
  
  
  51.     if err != nil {
    52. 
   
  
  
  
  52.         http.Error(w, err.Error(), http.StatusBadRequest)
    53. 
   
  
  
  
  53.         return 
    54. 
   
  
  
  
  54.     }
    55. 
   
  
  
  
  55. 
   
  
  
  
  56.     off, err := s.Log.Append(req.Record)
    57. 
   
  
  
  
  57.     if err != nil {
    58. 
   
  
  
  
  58.         http.Error(w, err.Error(), http.StatusInternalServerError)
    59. 
   
  
  
  
  59.         return 
    60. 
   
  
  
  
  60.     }
    61. 
   
  
  
  
  61. 
   
  
  
  
  62.     res := WriteResponse{Offset: off}
    63. 
   
  
  
  
  63.     //服务以json格式返回结果
    64. 
   
  
  
  
  64.     err = json.NewEncoder(w).Encode(res)
    65. 
   
  
  
  
  65.     if err != nil {
    66. 
   
  
  
  
  66.         http.Error(w, err.Error(), http.StatusInternalServerError)
    67. 
   
  
  
  
  67.         return 
    68. 
   
  
  
  
  68.     }
    69. 
   
  
  
  
  69. }
    70. 
   
  
  
  
  70. 
   
  
  
  
  71. func (s *httpServer) hadnleLogRead(w http.ResponseWriter, r *http.Request) {
    72. 
   
  
  
  
  72.     var req ReadRequest 
    73. 
   
  
  
  
  73.     err := json.NewDecoder(r.Body).Decode(&req)
    74. 
   
  
  
  
  74.     if err != nil {
    75. 
   
  
  
  
  75.         http.Error(w, err.Error(), http.StatusBadRequest)
    76. 
   
  
  
  
  76.         return 
    77. 
   
  
  
  
  77.     }
    78. 
   
  
  
  
  78. 
   
  
  
  
  79.     record, err := s.Log.Read(req.Offset)
    80. 
   
  
  
  
  80.     if err == ErrOffsetNotFound {
    81. 
   
  
  
  
  81.         http.Error(w, err.Error(), http.StatusNotFound)
    82. 
   
  
  
  
  82.         return
    83. 
   
  
  
  
  83.     }
    84. 
   
  
  
  
  84. 
   
  
  
  
  85.     if err != nil {
    86. 
   
  
  
  
  86.         http.Error(w, err.Error(), http.StatusInternalServerError)
    87. 
   
  
  
  
  87.         return 
    88. 
   
  
  
  
  88.     }
    89. 
   
  
  
  
  89. 
   
  
  
  
  90.     res := ReadResponse{Record: record}
    91. 
   
  
  
  
  91.     err = json.NewEncoder(w).Encode(res)
    92. 
   
  
  
  
  92.     if err != nil {
    93. 
   
  
  
  
  93.         http.Error(w, err.Error(), http.StatusInternalServerError)
    94. 
   
  
  
  
  94.         return
    95. 
   
  
  
  
  95.     }
    96. 
   
  
  
  
  96. }
    97.

上面代码显示出“分布式”，“微服务”的特点。相应的功能代码以单独服务器的形式运行，通过网络来接收服务请求，这对应“分布式”，每个独立模块只完成一个特定任务，这就对应“微服务”，由于这种方式可以同时在不同的机器上运行，于是展示了“可扩展性”。

同时服务既然以http 服务器的形式存在，因此服务的请求和返回也要走Http形式，同时数据以Json方式进行封装。同时实现的逻辑很简单，但有日志写请求时，我们把请求解析成Record结构体后加入到队列末尾，当有读取日志的请求时，我们获得客户端发来的读取偏移，然后取出对应的记录，封装成json格式后返回给客户。

完成了服务器的代码后，我们需要将服务器运行起来，为了达到模块化的目的，我们把服务器的启动放置在另一个地方，在proglog根目录下创建cmd/server,在里面添加main.go:

  
 
 
 
    
   
  
  
  
  1. package main 
    2. 
   
  
  
  
  2. 
   
  
  
  
  3. import (
    4. 
   
  
  
  
  4.     "log"
    5. 
   
  
  
  
  5.     "internal/server"
    6. 
   
  
  
  
  6. )
    7. 
   
  
  
  
  7. 
   
  
  
  
  8. func main() {
    9. 
   
  
  
  
  9.     srv := server.NewHttpServer(":8080")
    10. 
   
  
  
  
  10.     log.Fatal(srv.ListenAndServe())
    11. 
   
  
  
  
  11. }
    12.

同时为了能够引用internal/server下面的模块，我们需要在cmd/server下先通过go mod init cmd/server进行初始化，然后在go.mod文件中添加如下一行：

  
 
 
 
    
   
  
  
  
  1. replace internal/server => ../../internal/server
    2.

然后执行命令 go mod tidy,这样本地模块就知道根据给定的目录转换去引用模块，最后使用go run main.go启动日志服务，现在我们要做的是测试服务器的可用性，我们同样在目录下创建server_test.go，然后编写测试代码，基本逻辑就是想服务器发送日志写请求，然后再发送读请求，并比较读到的数据是否和我们写入的数据一致，代码如下：

  
 
 
 
    
   
  
  
  
  1. package main
    2. 
   
  
  
  
  2. 
   
  
  
  
  3. import(
    4. 
   
  
  
  
  4.     "encoding/json"
    5. 
   
  
  
  
  5.     "net/http"
    6. 
   
  
  
  
  6.     "internal/server"
    7. 
   
  
  
  
  7.     "bytes"
    8. 
   
  
  
  
  8.     "testing"
    9. 
   
  
  
  
  9.     "io/ioutil"
    10. 
   
  
  
  
  10. )
    11. 
   
  
  
  
  11. 
   
  
  
  
  12. func TestServerLogWrite(t *testing.T) {
    13. 
   
  
  
  
  13.     var tests = []struct {
    14. 
   
  
  
  
  14.         request server.WriteRequest
    15. 
   
  
  
  
  15.         want_response server.WriteResponse 
    16. 
   
  
  
  
  16.     } {
    17. 
   
  
  
  
  17.         {request: server.WriteRequest{server.Record{[]byte(`this is log request 1`), 0}}, 
    18. 
   
  
  
  
  18.          want_response:  server.WriteResponse{Offset: 0, },},
    19. 
   
  
  
  
  19.          {request: server.WriteRequest{server.Record{[]byte(`this is log request 2`), 0}}, 
    20. 
   
  
  
  
  20.          want_response:  server.WriteResponse{Offset: 1, },},
    21. 
   
  
  
  
  21.          {request: server.WriteRequest{server.Record{[]byte(`this is log request 3`), 0}}, 
    22. 
   
  
  
  
  22.          want_response:  server.WriteResponse{Offset: 2, },},
    23. 
   
  
  
  
  23.     }
    24. 
   
  
  
  
  24. 
   
  
  
  
  25.     for _, test := range tests {
    26. 
   
  
  
  
  26.         //将请求转换成json格式并post给日志服务
    27. 
   
  
  
  
  27.         request := &test.request 
    28. 
   
  
  
  
  28.         request_json, err := json.Marshal(request)
    29. 
   
  
  
  
  29.         if err != nil {
    30. 
   
  
  
  
  30.             t.Errorf("convert request to json fail")
    31. 
   
  
  
  
  31.             return 
    32. 
   
  
  
  
  32.         }
    33. 
   
  
  
  
  33. 
   
  
  
  
  34.         resp, err := http.Post("http://localhost:8080", "application/json",bytes.NewBuffer(request_json))
    35. 
   
  
  
  
  35.         defer resp.Body.Close()
    36. 
   
  
  
  
  36.         if err != nil {
    37. 
   
  
  
  
  37.             t.Errorf("http post request fail: %v", err)
    38. 
   
  
  
  
  38.             return
    39. 
   
  
  
  
  39.         }
    40. 
   
  
  
  
  40. 
   
  
  
  
  41.         //解析日志服务返回结果
    42. 
   
  
  
  
  42.         body, err := ioutil.ReadAll(resp.Body)
    43. 
   
  
  
  
  43.         var response server.WriteResponse 
    44. 
   
  
  
  
  44.         err = json.Unmarshal([]byte(body), &response)
    45. 
   
  
  
  
  45.         if err != nil {
    46. 
   
  
  
  
  46.             t.Errorf("Unmarshal write response fail: %v", err)
    47. 
   
  
  
  
  47.         }
    48. 
   
  
  
  
  48. 
   
  
  
  
  49.         //检测结果是否与预期一致
    50. 
   
  
  
  
  50.         if response.Offset != test.want_response.Offset {
    51. 
   
  
  
  
  51.             t.Errorf("got offset: %d, but want offset: %d", response.Offset, test.want_response.Offset)
    52. 
   
  
  
  
  52.         }
    53. 
   
  
  
  
  53. 
   
  
  
  
  54.     }
    55. 
   
  
  
  
  55. 
   
  
  
  
  56.     var read_tests = []struct {
    57. 
   
  
  
  
  57.         request server.ReadRequest 
    58. 
   
  
  
  
  58.         want server.ReadResponse 
    59. 
   
  
  
  
  59.     } {
    60. 
   
  
  
  
  60.         {request: server.ReadRequest{Offset : 0,}, 
    61. 
   
  
  
  
  61.         want: server.ReadResponse{server.Record{[]byte(`this is log request 1`), 0}} },
    62. 
   
  
  
  
  62.         {request: server.ReadRequest{Offset : 1,}, 
    63. 
   
  
  
  
  63.         want: server.ReadResponse{server.Record{[]byte(`this is log request 2`), 0}} },
    64. 
   
  
  
  
  64.         {request: server.ReadRequest{Offset : 2,}, 
    65. 
   
  
  
  
  65.         want: server.ReadResponse{server.Record{[]byte(`this is log request 3`), 0}} },
    66. 
   
  
  
  
  66.     }
    67. 
   
  
  
  
  67. 
   
  
  
  
  68.     for _, test := range read_tests {
    69. 
   
  
  
  
  69.         request := test.request 
    70. 
   
  
  
  
  70.         request_json , err := json.Marshal(request)
    71. 
   
  
  
  
  71.         if err != nil {
    72. 
   
  
  
  
  72.             t.Errorf("convert read request to json fail")
    73. 
   
  
  
  
  73.             return 
    74. 
   
  
  
  
  74.         }
    75. 
   
  
  
  
  75. 
   
  
  
  
  76.         //将请求转换为json并放入GET请求体
    77. 
   
  
  
  
  77.         client := &http.Client{}
    78. 
   
  
  
  
  78.         req, err := http.NewRequest(http.MethodGet, "http://localhost:8080", bytes.NewBuffer(request_json))
    79. 
   
  
  
  
  79.         req.Header.Set("Content-Type", "application/json")
    80. 
   
  
  
  
  80.         resp, err := client.Do(req)
    81. 
   
  
  
  
  81.         if err != nil {
    82. 
   
  
  
  
  82.             t.Errorf("read request fail: %v", err)
    83. 
   
  
  
  
  83.             return 
    84. 
   
  
  
  
  84.         }
    85. 
   
  
  
  
  85. 
   
  
  
  
  86.         //解析读请求返回的结果
    87. 
   
  
  
  
  87.         defer resp.Body.Close()
    88. 
   
  
  
  
  88.         body, err := ioutil.ReadAll(resp.Body)
    89. 
   
  
  
  
  89.         var response server.ReadResponse
    90. 
   
  
  
  
  90.         err = json.Unmarshal([]byte(body), &response)
    91. 
   
  
  
  
  91.         if err != nil {
    92. 
   
  
  
  
  92.             t.Errorf("Unmarshal read response fail: %v", err)
    93. 
   
  
  
  
  93.             return 
    94. 
   
  
  
  
  94.         }
    95. 
   
  
  
  
  95. 
   
  
  
  
  96.         res := bytes.Compare(response.Record.Value, test.want.Record.Value)
    97. 
   
  
  
  
  97.         if res != 0 {
    98. 
   
  
  
  
  98.             t.Errorf("got value: %q, but want value : %q", response.Record.Value, test.want.Record.Value)
    99. 
   
  
  
  
  99.         }
    100. 
   
  
  
  
  100.     }
    101. 
   
  
  
  
  101. 
   
  
  
  
  102. }
    103.

完成上面代码后，使用go test运行，结果如下图所示：

从结果看到，我们的测试能通过，也就是无论是向日志服务提交写入请求还是读取请求，所得的结果跟我们预想的一致。总结一下，本节我们设计了一个简单的JSON/HTTP 日志服务，它能够接收基于JSON的http写请求和读请求，后面我们还会研究基于gPRC技术的微服务开发技术.

代码获取

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

网站栏目：GO实现高并发高可用分布式系统:Log微服务的实现
文章网址：http://www.36103.cn/qtweb/news17/16817.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

域名注册知识

分类信息网

成都服务器机柜租用　　 　微信小程序开发　　 　边坡柔性防护网　　 　上市企业网站建设方案　　 　网站建设公司　　 　精灵网　　 　成都气球装饰　　 　资中网站建设　　 　域名注册　　 　中国电信西部信息中心机房　　 　网页改版　　 　SSL证书　　 　成都机柜租用　　 　成都网站代运营　　 　企业网站建设　　 　双流能亿建站　　 　四川服务器托管　　 　犀浦门窗定制　　 　南充做网站　　 　西云服务器托管