今天要说的技术方案也是有一定项目背景的。在上一个项目中，我们需要对一个redis集群中过期的key进行处理，这是一个分布式
系统，考虑到高可用性，需要具备过期处理功能的服务有多个副本，这样我们就要求在同一时间内仅有一个副本可以对过期的key>进行处理，如果该副本挂掉，系统会在其他副本中再挑选出一个来处理过期的key。

很显然，这里涉及到一个选主(leader election)的过程。每当涉及选主，很多人就会想到一些高大上的分布式一致性/共识算法，
比如：raft、paxos等。当然使用这
些算法自然没有问题，但是也给系统徒增了很多复杂性。能否有一些更简单直接的方案呢？我们已经有了一个redis集群，是否可>以利用redis集群的能力来完成这一点呢？

Redis原生并没有提供leader election算法，但Redis作者提供了分布式锁的算法，也就>是说我们可以用分布式锁来实现一个简单的选主功能，见下图：

在上图中我们看到，只有持有锁的服务才具备操作数据的资格，也就是说持有锁的服务的角色是leader，而其他服务则继续尝试去持有锁，它们是follower的角色。

1. 基于单节点redis的分布式锁

在redis官方有关分布式锁算法的介绍页面中，作者给出了各种编程语言的推荐实现，而Go语言的推荐实现仅redsync这一种。在这篇短文中，我们就来使用redsync实现基于Redis分布式锁的选主方案。

在Go生态中，连接和操作redis的主流go客户端库有go-redis和redigo。最新的redsync版本底层redis driver既支持go-redis，也支持redigo，我个人日常使用最多的是go-redis这个客户端，这里我们就用go-redis。

redsync github主页中给出的例子是基于单redis node的分布式锁示例。下面我们也先以单redis节点来看看如何通过Redis的分布式锁实现我们的业务逻辑：

// github.com/bigwhite/experiments/blob/master/redis-cluster-distributed-lock/standalone/main.go

     1  package main
     2
     3  import (
     4      "context"
     5      "log"
     6      "os"
     7      "os/signal"
     8      "sync"
     9      "sync/atomic"
    10      "syscall"
    11      "time"
    12
    13      goredislib "github.com/go-redis/redis/v8"
    14      "github.com/go-redsync/redsync/v4"
    15      "github.com/go-redsync/redsync/v4/redis/goredis/v8"
    16  )
    17
    18  const (
    19      redisKeyExpiredEventSubj = `__keyevent@0__:expired`
    20  )
    21
    22  var (
    23      isLeader  int64
    24      m         atomic.Value
    25      id        string
    26      mutexName = "the-year-of-the-ox-2021"
    27  )
    28
    29  func init() {
    30      if len(os.Args) < 2 {
    31          panic("args number is not correct")
    32      }
    33      id = os.Args[1]
    34  }
    35
    36  func tryToBecomeLeader() (bool, func() (bool, error), error) {
    37      client := goredislib.NewClient(&goredislib.Options{
    38          Addr: "localhost:6379",
    39      })
    40      pool := goredis.NewPool(client)
    41      rs := redsync.New(pool)
    42
    43      mutex := rs.NewMutex(mutexName)
    44
    45      if err := mutex.Lock(); err != nil {
    46          client.Close()
    47          return false, nil, err
    48      }
    49
    50      return true, func() (bool, error) {
    51          return mutex.Unlock()
    52      }, nil
    53  }
    54
    55  func doElectionAndMaintainTheStatus(quit <-chan struct{}) {
    56      ticker := time.NewTicker(time.Second * 5)
    57      var err error
    58      var ok bool
    59      var cf func() (bool, error)
    60
    61      c := goredislib.NewClient(&goredislib.Options{
    62          Addr: "localhost:6379",
    63      })
    64      defer c.Close()
    65      for {
    66          select {
    67          case <-ticker.C:
    68              if atomic.LoadInt64(&isLeader) == 0 {
    69                  ok, cf, err = tryToBecomeLeader()
    70                  if ok {
    71                      log.Printf("prog-%s become leader successfully\n", id)
    72                      atomic.StoreInt64(&isLeader, 1)
    73                      defer cf()
    74                  }
    75                  if !ok || err != nil {
    76                      log.Printf("prog-%s try to become leader failed: %s\n", id, err)
    77                  }
    78              } else {
    79                  log.Printf("prog-%s is the leader\n", id)
    80                  // update the lock live time and maintain the leader status
    81                  c.Expire(context.Background(), mutexName, 8*time.Second)
    82              }
    83          case <-quit:
    84              return
    85          }
    86      }
    87  }
    88
    89  func doExpire(quit <-chan struct{}) {
    90      // subscribe the expire event of redis
    91      c := goredislib.NewClient(&goredislib.Options{
    92          Addr: "localhost:6379"})
    93      defer c.Close()
    94
    95      ctx := context.Background()
    96      pubsub := c.Subscribe(ctx, redisKeyExpiredEventSubj)
    97      _, err := pubsub.Receive(ctx)
    98      if err != nil {
    99          log.Printf("prog-%s subscribe expire event failed: %s\n", id, err)
   100          return
   101      }
   102      log.Printf("prog-%s subscribe expire event ok\n", id)
   103
   104      // Go channel which receives messages from redis db
   105      ch := pubsub.Channel()
   106      for {
   107          select {
   108          case event := <-ch:
   109              key := event.Payload
   110              if atomic.LoadInt64(&isLeader) == 0 {
   111                  break
   112              }
   113              log.Printf("prog-%s 收到并处理一条过期消息[key:%s]", id, key)
   114          case <-quit:
   115              return
   116          }
   117      }
   118  }
   119
   120  func main() {
   121      var wg sync.WaitGroup
   122      wg.Add(2)
   123      var quit = make(chan struct{})
   124
   125      go func() {
   126          doElectionAndMaintainTheStatus(quit)
   127          wg.Done()
   128      }()
   129      go func() {
   130          doExpire(quit)
   131          wg.Done()
   132      }()
   133
   134      c := make(chan os.Signal, 1)
   135      signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
   136      _ = <-c
   137      close(quit)
   138      log.Printf("recv exit signal...")
   139      wg.Wait()
   140      log.Printf("program exit ok")
   141  }

上面示例代码比较长，但它很完整。我们一点点来看。

首先，我们看120~141行的main函数结构。在这个函数中，我们创建了两个新goroutine，main goroutine通过sync.WaitGroup等待这两个子goroutine的退出并使用quit channel模式(关于goroutine的并发模式的详解，可以参考我的专栏文章《Go并发模型和常见并发模式》)在收到系统信号(关于signal包的使用，请参见我的专栏文章《小心被kill！不要忽略对系统信号的处理》)后通知两个子goroutine退出。

接下来，我们逐个看两个子goroutine的执行逻辑。第一个goroutine执行的是doElectionAndMaintainTheStatus函数。该函数会持续尝试去持有分布式锁(tryToBecomeLeader)，一旦持有，它就变成了分布式系统中的leader角色；成为leader角色的副本会保持其角色状态(见81行)。

尝试持有分布式锁并成为leader是tryToBecomeLeader函数的主要职责，该函数直接使用了redsync包的算法，并利用与redis node建立的连接(NewClient)，尝试建立并持有分布式锁“the-year-of-the-ox-2021”。我们使用的是默认的锁属性，从redsync包的NewMutex方法源码，我们能看到锁默认属性如下：

// github.com/go-redsync/redsync/redsync.go

// NewMutex returns a new distributed mutex with given name.
func (r *Redsync) NewMutex(name string, options ...Option) *Mutex {
        m := &Mutex{
                name:         name,
                expiry:       8 * time.Second,
                tries:        32,
                delayFunc:    func(tries int) time.Duration { return 500 * time.Millisecond },
                genValueFunc: genValue,
                factor:       0.01,
                quorum:       len(r.pools)/2 + 1,
                pools:        r.pools,
        }
        for _, o := range options {
                o.Apply(m)
        }
        return m
}

我们看到锁有一个过期时间属性(expiry)，过期时间默认仅有8秒。问题来了：一旦锁过期了，那么情况会怎样？事实是一旦锁过期掉，在leader尚未解锁时，其follower也会加锁成功，因为原锁的key已经因过期而被删除掉了。长此以往，整个分布式系统就会存在多个自视为leader的进程，整个处理逻辑就乱了！

解决这个问题至少可以有三种方案：

• 方案1：将锁的expiry设置的很长，长到一旦某个服务持有了锁，不需担心锁过期的问题；
• 方案2：在所的默认expiry到期之前解锁，所有服务重新竞争锁；
• 方案3：一旦某个服务持有了锁，则需要定期重设锁的expiry时间，保证锁不会过期，直到该服务主动执行unlock。

方案1的问题在于，一旦持有锁的leader因意外异常退出并且尚未unlock，那么由于锁的过期时间超级长，其他follower依然无法持有锁而变成下一任leader，导致整个分布式系统的leader缺失，业务逻辑无法继续进行；

方案2其实是基于Redis分布式锁的常规使用方式，但对于像我这里的业务场景，频繁lock和unlock没必要，我只需要保证系统中有一个leader一直在处理过期event即可，在服务间轮流处理并非我的需求。但这个方案是一个可行的方案，代码逻辑清晰也简单。

方案3则是非常适合我的业务场景的方案，持有锁的leader通过定期(<8s)的更新锁的过期时间来保证锁的有效性，这样避免了leader频繁切换。这里我们就使用了这一方案，见78~82行，我们在定时器的帮助下，定期重新设置了锁的过期时间(8s)。

在上述示例代码中，我们用一个变量isLeader来标识该服务是否持有了锁，由于该变量被多个goroutine访问和修改，因此我们通过atomic包实现对其的原子访问以避免出现race问题。

最后，我们说说这段示例承载的业务逻辑(doExpire函数)。真正的业务逻辑由doExpire函数实现。它通过监听redis 0号库的key空间的过期事件实现对目标key的过期处理(这里并未体现这一点)。

subscribe的subject字符串为keyevent@0:expired，这个字符串的组成含义可以参考redis官方对notifications的说明，这里的字串表明我们要监听key事件，在0号数据库，事件类型是key过期。

当在0号数据库有key过期后，我们的订阅channel(105行)就会收到一个事件，通过event的Payload我们可以得到key的名称，后续我们可以根据key的名字来过滤掉我们不关心的key，而仅对期望的key做相应处理。

在默认配置下， redis的通知功能处于关闭状态。我们需要通过命令或在redis.conf中开启这一功能。

$redis-cli
127.0.0.1:6379> config set notify-keyspace-events KEx
OK

到这里，我们已经搞清楚了上面示例代码的原理，下面我们就来真实运行一次上面的代码，我们编译上面代码并启动三个实例：

$go build main.go
$./main 1
$./main 2
$./main 3

由于./main 1先启动，因此第一个启动的服务一般会先成为leader：

$main 1
2021/02/11 05:43:15 prog-1 subscribe expire event ok
2021/02/11 05:43:20 prog-1 become leader successfully
2021/02/11 05:43:25 prog-1 is the leader
2021/02/11 05:43:30 prog-1 is the leader

而其他两个服务会定期尝试去持有锁：

$main 2
2021/02/11 05:43:17 prog-2 subscribe expire event ok
2021/02/11 05:43:37 prog-2 try to become leader failed: redsync: failed to acquire lock
2021/02/11 05:43:53 prog-2 try to become leader failed: redsync: failed to acquire lock

$main 3
2021/02/11 05:43:18 prog-3 subscribe expire event ok
2021/02/11 05:43:38 prog-3 try to become leader failed: redsync: failed to acquire lock
2021/02/11 05:43:54 prog-3 try to become leader failed: redsync: failed to acquire lock

这时我们通过redis-cli在0号数据库中创建一个key1，过期时间5s：

$redis-cli
127.0.0.1:6379> setex key1 5 value1
OK

5s后，我们会在prog-1这个服务实例的输出日志中看到如下内容：

2021/02/11 05:43:50 prog-1 is the leader
2021/02/11 05:43:53 prog-1 收到并处理一条过期消息[key:key1]
2021/02/11 05:43:55 prog-1 is the leader

接下来，我们停掉prog-1：

2021/02/11 05:44:00 prog-1 is the leader
^C2021/02/11 05:44:01 recv exit signal...
redis: 2021/02/11 05:44:01 pubsub.go:168: redis: discarding bad PubSub connection: read tcp [::1]:56594->[::1]:6379: use of closed network connection
2021/02/11 05:44:01 program exit ok

在停掉prog-1后的瞬间，prog-2成功持有了锁，并成为leader：

2021/02/11 05:44:01 prog-2 become leader successfully
2021/02/11 05:44:01 prog-2 is the leader

我们再通过redis-cli在0号数据库中创建一个key2，过期时间5s：

$redis-cli
127.0.0.1:6379> setex key2 5 value2
OK

5s后，我们会在prog-2这个服务实例的输出日志中看到如下内容：

2021/02/11 05:44:17 prog-2 is the leader
2021/02/11 05:44:19 prog-2 收到并处理一条过期消息[key:key2]
2021/02/11 05:44:22 prog-2 is the leader

从运行的结果来看，该分布式系统的运行逻辑是符合我们的设计预期的。

2. 基于redis集群的分布式锁

上面，我们实现了基于单个redis节点的分布式锁的选主功能。在生产环境，我们很少会使用单节点的Redis，通常会使用Redis集群以保证高可用性。

最新的redsync已经支持了redis cluster(基于go-redis)。和单节点唯一不同的是，我们传递给redsync的pool所使用的与redis的连接由Client类型变为了ClusterClient类型：

// github.com/bigwhite/experiments/blob/master/redis-cluster-distributed-lock/cluster/v1/main.go
const (
        redisClusterMasters      = "localhost:30001,localhost:30002,localhost:30003"
)

func main() {
    ... ...
        client := goredislib.NewClusterClient(&goredislib.ClusterOptions{
                Addrs: strings.Split(redisClusterMasters, ",")})
        defer client.Close()
    ... ...
}

我们在本地启动的redis cluster，三个master的地址分别为：localhost:30001、localhost:30002和localhost:30003。我们将master的地址组成一个逗号分隔的常量redisClusterMasters。

我们对上面单节点的代码做了改进，将Redis连接的创建放在了main中，并将client连接作为参数传递给各个goroutine的运行函数。下面是cluster版示例代码完整版(v1)：

// github.com/bigwhite/experiments/blob/master/redis-cluster-distributed-lock/cluster/v1/main.go

     1  package main
     2
     3  import (
     4      "context"
     5      "log"
     6      "os"
     7      "os/signal"
     8      "strings"
     9      "sync"
    10      "sync/atomic"
    11      "syscall"
    12      "time"
    13
    14      goredislib "github.com/go-redis/redis/v8"
    15      "github.com/go-redsync/redsync/v4"
    16      "github.com/go-redsync/redsync/v4/redis/goredis/v8"
    17  )
    18
    19  const (
    20      redisKeyExpiredEventSubj = `__keyevent@0__:expired`
    21      redisClusterMasters      = "localhost:30001,localhost:30002,localhost:30003"
    22  )
    23
    24  var (
    25      isLeader  int64
    26      m         atomic.Value
    27      id        string
    28      mutexName = "the-year-of-the-ox-2021"
    29  )
    30
    31  func init() {
    32      if len(os.Args) < 2 {
    33          panic("args number is not correct")
    34      }
    35      id = os.Args[1]
    36  }
    37
    38  func tryToBecomeLeader(client *goredislib.ClusterClient) (bool, func() (bool, error), error) {
    39      pool := goredis.NewPool(client)
    40      rs := redsync.New(pool)
    41
    42      mutex := rs.NewMutex(mutexName)
    43
    44      if err := mutex.Lock(); err != nil {
    45          return false, nil, err
    46      }
    47
    48      return true, func() (bool, error) {
    49          return mutex.Unlock()
    50      }, nil
    51  }
    52
    53  func doElectionAndMaintainTheStatus(c *goredislib.ClusterClient, quit <-chan struct{}) {
    54      ticker := time.NewTicker(time.Second * 5)
    55      var err error
    56      var ok bool
    57      var cf func() (bool, error)
    58
    59      for {
    60          select {
    61          case <-ticker.C:
    62              if atomic.LoadInt64(&isLeader) == 0 {
    63                  ok, cf, err = tryToBecomeLeader(c)
    64                  if ok {
    65                      log.Printf("prog-%s become leader successfully\n", id)
    66                      atomic.StoreInt64(&isLeader, 1)
    67                      defer cf()
    68                  }
    69                  if !ok || err != nil {
    70                      log.Printf("prog-%s try to become leader failed: %s\n", id, err)
    71                  }
    72              } else {
    73                  log.Printf("prog-%s is the leader\n", id)
    74                  // update the lock live time and maintain the leader status
    75                  c.Expire(context.Background(), mutexName, 8*time.Second)
    76              }
    77          case <-quit:
    78              return
    79          }
    80      }
    81  }
    82
    83  func doExpire(c *goredislib.ClusterClient, quit <-chan struct{}) {
    84      // subscribe the expire event of redis
    85      ctx := context.Background()
    86      pubsub := c.Subscribe(ctx, redisKeyExpiredEventSubj)
    87      _, err := pubsub.Receive(ctx)
    88      if err != nil {
    89          log.Printf("prog-%s subscribe expire event failed: %s\n", id, err)
    90          return
    91      }
    92      log.Printf("prog-%s subscribe expire event ok\n", id)
    93
    94      // Go channel which receives messages from redis db
    95      ch := pubsub.Channel()
    96      for {
    97          select {
    98          case event := <-ch:
    99              key := event.Payload
   100              if atomic.LoadInt64(&isLeader) == 0 {
   101                  break
   102              }
   103              log.Printf("prog-%s 收到并处理一条过期消息[key:%s]", id, key)
   104          case <-quit:
   105              return
   106          }
   107      }
   108  }
   109
   110  func main() {
   111      var wg sync.WaitGroup
   112      wg.Add(2)
   113      var quit = make(chan struct{})
   114      client := goredislib.NewClusterClient(&goredislib.ClusterOptions{
   115          Addrs: strings.Split(redisClusterMasters, ",")})
   116      defer client.Close()
   117
   118      go func() {
   119          doElectionAndMaintainTheStatus(client, quit)
   120          wg.Done()
   121      }()
   122      go func() {
   123          doExpire(client, quit)
   124          wg.Done()
   125      }()
   126
   127      c := make(chan os.Signal, 1)
   128      signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
   129      _ = <-c
   130      close(quit)
   131      log.Printf("recv exit signal...")
   132      wg.Wait()
   133      log.Printf("program exit ok")
   134  }

和单一节点一样，我们运行三个服务实例：

$go build main.go
$main 1
2021/02/11 09:49:16 prog-1 subscribe expire event ok
2021/02/11 09:49:22 prog-1 become leader successfully
2021/02/11 09:49:26 prog-1 is the leader
2021/02/11 09:49:31 prog-1 is the leader
2021/02/11 09:49:36 prog-1 is the leader
... ...

$main 2
2021/02/11 09:49:19 prog-2 subscribe expire event ok
2021/02/11 09:49:40 prog-2 try to become leader failed: redsync: failed to acquire lock
2021/02/11 09:49:55 prog-2 try to become leader failed: redsync: failed to acquire lock
... ...

$main 3
2021/02/11 09:49:31 prog-3 subscribe expire event ok
2021/02/11 09:49:52 prog-3 try to become leader failed: redsync: failed to acquire lock
2021/02/11 09:50:07 prog-3 try to become leader failed: redsync: failed to acquire lock
... ...

我们看到基于Redis集群版的分布式锁也生效了！prog-1成功持有锁并成为leader! 接下来我们再来看看对过期key事件的处理！

我们通过下面命令让redis-cli连接到集群中的所有节点并设置每个节点开启key空间的事件通知：

三主：

$redis-cli -c -h localhost -p 30001
localhost:30001> config set notify-keyspace-events KEx
OK

$redis-cli -c -h localhost -p 30002
localhost:30002> config set notify-keyspace-events KEx
OK

$redis-cli -c -h localhost -p 30003
localhost:30003> config set notify-keyspace-events KEx
OK

三从：

$redis-cli -c -h localhost -p 30004
localhost:30004> config set notify-keyspace-events KEx
OK

$redis-cli -c -h localhost -p 30005
localhost:30005> config set notify-keyspace-events KEx
OK

$redis-cli -c -h localhost -p 30006
localhost:30006> config set notify-keyspace-events KEx
OK

在node1节点上，我们set一个有效期为5s的key：key1：

localhost:30001> setex key1 5 value1
-> Redirected to slot [9189] located at 127.0.0.1:30002
OK

等待5s后，我们的leader：prog-1并没有如预期那样受到expire通知！ 这是怎么回事呢？追本溯源，我们查看一下redis官方文档关于notifications的说明，我们在文档最后一段找到如下描述：

Events in a cluster

Every node of a Redis cluster generates events about its own subset of the keyspace as described above. However, unlike regular Pub/Sub communication in a cluster, events' notifications are not broadcasted to all nodes. Put differently, keyspace events are node-specific. This means that to receive all keyspace events of a cluster, clients need to subscribe to each of the nodes.

这段话大致意思是Redis集群中的每个redis node都有自己的keyspace，事件通知不会被广播到集群内的所有节点，即keyspace的事件是node相关的。如果要接收一个集群中的所有keyspace的event，那客户端就需要Subcribe集群内的所有节点。我们来改一下代码，形成v2版(考虑到篇幅就不列出所有代码了，仅列出相对于v1版变化的代码)：

// github.com/bigwhite/experiments/blob/master/redis-cluster-distributed-lock/cluster/v2/main.go

... ...
    19  const (
    20      redisKeyExpiredEventSubj = `__keyevent@0__:expired`
    21      redisClusterMasters      = "localhost:30001,localhost:30002,localhost:30003,localhost:30004,localhost:30005,localhost:30006"
    22  )
... ...
    83  func doExpire(quit <-chan struct{}) {
    84      var ch = make(chan *goredislib.Message)
    85      nodes := strings.Split(redisClusterMasters, ",")
    86
    87      for _, node := range nodes {
    88          node := node
    89          go func(quit <-chan struct{}) {
    90              c := goredislib.NewClient(&goredislib.Options{
    91                  Addr: node})
    92              defer c.Close()
    93
    94              // subscribe the expire event of redis
    95              ctx := context.Background()
    96              pubsub := c.Subscribe(ctx, redisKeyExpiredEventSubj)
    97              _, err := pubsub.Receive(ctx)
    98              if err != nil {
    99                  log.Printf("prog-%s subscribe expire event of node[%s] failed: %s\n",
   100                      id, node, err)
   101                  return
   102              }
   103              log.Printf("prog-%s subscribe expire event of node[%s] ok\n", id, node)
   104
   105              // Go channel which receives messages from redis db
   106              pch := pubsub.Channel()
   107
   108              for {
   109                  select {
   110                  case event := <-pch:
   111                      ch <- event
   112                  case <-quit:
   113                      return
   114                  }
   115              }
   116          }(quit)
   117      }
   118      for {
   119          select {
   120          case event := <-ch:
   121              key := event.Payload
   122              if atomic.LoadInt64(&isLeader) == 0 {
   123                  break
   124              }
   125              log.Printf("prog-%s 收到并处理一条过期消息[key:%s]", id, key)
   126          case <-quit:
   127              return
   128          }
   129      }
   130  }
   131
   132  func main() {
   133      var wg sync.WaitGroup
   134      wg.Add(2)
   135      var quit = make(chan struct{})
   136      client := goredislib.NewClusterClient(&goredislib.ClusterOptions{
   137          Addrs: strings.Split(redisClusterMasters, ",")})
   138      defer client.Close()
   139
   140      go func() {
   141          doElectionAndMaintainTheStatus(client, quit)
   142          wg.Done()
   143      }()
   144      go func() {
   145          doExpire(quit)
   146          wg.Done()
   147      }()
   148
   149      c := make(chan os.Signal, 1)
   150      signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
   151      _ = <-c
   152      close(quit)
   153      log.Printf("recv exit signal...")
   154      wg.Wait()
   155      log.Printf("program exit ok")
   156  }

在这个新版代码中，我们在每个新goroutine中实现对redis一个节点的Subscribe，并将收到的Event notifications通过“扇入”模式(更多关于并发扇入模式的内容，可以参考我的Go技术专栏文章《Go并发模型和常见并发模式》)统一写入到运行doExpire的goroutine中做统一处理。

我们再来运行一下这个示例，并在不同时机创建多个key来验证通知接收和处理的效果：

$main 1
2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30004] ok
2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30001] ok
2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30006] ok
2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30002] ok
2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30003] ok
2021/02/11 10:29:21 prog-1 subscribe expire event of node[localhost:30005] ok
2021/02/11 10:29:26 prog-1 become leader successfully
2021/02/11 10:29:31 prog-1 is the leader
2021/02/11 10:29:36 prog-1 is the leader
2021/02/11 10:29:41 prog-1 is the leader
2021/02/11 10:29:46 prog-1 is the leader
2021/02/11 10:29:47 prog-1 收到并处理一条过期消息[key:key1]
2021/02/11 10:29:51 prog-1 is the leader
2021/02/11 10:29:51 prog-1 收到并处理一条过期消息[key:key2]
2021/02/11 10:29:56 prog-1 收到并处理一条过期消息[key:key3]
2021/02/11 10:29:56 prog-1 is the leader
2021/02/11 10:30:01 prog-1 is the leader
2021/02/11 10:30:06 prog-1 is the leader
^C2021/02/11 10:30:08 recv exit signal...

$main 3
2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30004] ok
2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30006] ok
2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30002] ok
2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30001] ok
2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30005] ok
2021/02/11 10:29:27 prog-3 subscribe expire event of node[localhost:30003] ok
2021/02/11 10:29:48 prog-3 try to become leader failed: redsync: failed to acquire lock
2021/02/11 10:30:03 prog-3 try to become leader failed: redsync: failed to acquire lock
2021/02/11 10:30:08 prog-3 become leader successfully
2021/02/11 10:30:08 prog-3 is the leader
2021/02/11 10:30:12 prog-3 is the leader
2021/02/11 10:30:17 prog-3 is the leader
2021/02/11 10:30:22 prog-3 is the leader
2021/02/11 10:30:23 prog-3 收到并处理一条过期消息[key:key4]
2021/02/11 10:30:27 prog-3 is the leader
^C2021/02/11 10:30:28 recv exit signal...

$main 2
2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30005] ok
2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30006] ok
2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30003] ok
2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30004] ok
2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30002] ok
2021/02/11 10:29:24 prog-2 subscribe expire event of node[localhost:30001] ok
2021/02/11 10:29:45 prog-2 try to become leader failed: redsync: failed to acquire lock
2021/02/11 10:30:01 prog-2 try to become leader failed: redsync: failed to acquire lock
2021/02/11 10:30:16 prog-2 try to become leader failed: redsync: failed to acquire lock
2021/02/11 10:30:28 prog-2 become leader successfully
2021/02/11 10:30:28 prog-2 is the leader
2021/02/11 10:30:29 prog-2 is the leader
2021/02/11 10:30:34 prog-2 is the leader
2021/02/11 10:30:39 prog-2 收到并处理一条过期消息[key:key5]
2021/02/11 10:30:39 prog-2 is the leader
^C2021/02/11 10:30:41 recv exit signal...

这个运行结果如预期！

不过这个方案显然也不是那么理想，毕竟我们要单独Subscribe每个集群内的redis节点，目前没有理想方案，除非redis cluster支持带广播的Event notification。

以上示例代码可以在这里 https://github.com/bigwhite/experiments/tree/master/redis-cluster-distributed-lock 下载 。

“Gopher部落”知识星球开球了！高品质首发Go技术文章，“三天”首发阅读权，每年两期Go语言发展现状分析，每天提前1小时阅读到新鲜的Gopher日报，网课、技术专栏、图书内容前瞻，六小时内必答保证等满足你关于Go语言生态的所有需求！星球首开，福利自然是少不了的！2020年年底之前，8.8折(很吉利吧^_^)加入星球，下方图片扫起来吧！

Go技术专栏“改善Go语⾔编程质量的50个有效实践”正在慕课网火热热销中！本专栏主要满足广大gopher关于Go语言进阶的需求，围绕如何写出地道且高质量Go代码给出50条有效实践建议，上线后收到一致好评！欢迎大家订阅！

我的网课“Kubernetes实战：高可用集群搭建、配置、运维与应用”在慕课网热卖中，欢迎小伙伴们订阅学习！

我爱发短信：企业级短信平台定制开发专家 https://tonybai.com/。smspush : 可部署在企业内部的定制化短信平台，三网覆盖，不惧大并发接入，可定制扩展； 短信内容你来定，不再受约束, 接口丰富，支持长短信，签名可选。2020年4月8日，中国三大电信运营商联合发布《5G消息白皮书》，51短信平台也会全新升级到“51商用消息平台”，全面支持5G RCS消息。

著名云主机服务厂商DigitalOcean发布最新的主机计划，入门级Droplet配置升级为：1 core CPU、1G内存、25G高速SSD，价格5$/月。有使用DigitalOcean需求的朋友，可以打开这个链接地址：https://m.do.co/c/bff6eed92687 开启你的DO主机之路。

Gopher Daily(Gopher每日新闻)归档仓库 – https://github.com/bigwhite/gopherdaily

我的联系方式：

• 微博：https://weibo.com/bigwhite20xx
• 微信公众号：iamtonybai
• 博客：tonybai.com
• github: https://github.com/bigwhite
• “Gopher部落”知识星球：https://public.zsxq.com/groups/51284458844544

微信赞赏：

商务合作方式：撰稿、出书、培训、在线课程、合伙创业、咨询、广告合作。

1. Form简介

Form(中文译为表单)，是HTML标记语言中的重要语法元素。一个Form不仅包含正常的文本内容、标记等，还包含被称为控件的特殊元素。用户通常通过修改控件（比如：输入文本、选择菜单项等）来“完成”表单，然后将表单数据以HTTP Get或Post请求的形式提交（submit）给Web服务器。

很多初学者总是混淆HTML和HTTP。其实，http通常作为html传输的承载体，打个比方，html就像乘客，http就像出租车，将乘客从一个地方运输到另外一个地方。但显然http这辆出租车可不仅仅只拉html这一个乘客，很多格式均可作为http这辆出租车的乘客，比如json(over http)、xml(over http)。

在一个HTML文档中，一个表单的标准格式如下：

<form action="http://localhost:8080/repositories" method="get">
   <input type="text" name="language" value="go" />
   <input type="text" name="since" value="monthly" />
   <input type="submit" />
</form>

这样的一个Form被加载到浏览器中后会呈现为一个表单的样式，当在两个文本框中分别输入文本(或以默认的文本作为输入)后，点击“提交(submit)”，浏览器会向http://localhost:8080发出一个HTTP请求，由于Form的method属性为get，因此该HTTP请求会将表单的输入文本作为查询字符串参数(Query String Parameter，在这里即是?language=go&since=monthly)。服务器端处理完该请求后，会返回一个HTTP承载的应答，该应答被浏览器接收后会按特定样式呈现在浏览器窗口中。上述这个过程可以用总结为下面这幅示意图：

Form中的method也可以使用post，就像下面这样：

<form action="http://localhost:8080/repositories" method="post">
   <input type="text" name="language" value="go" />
   <input type="text" name="since" value="monthly" />
   <input type="submit" />
</form>

改为post的Form表单在点击提交后发出的http请求与method=get时的请求有何不同呢？不同之处就在于在method=post的情况下，表单的参数不会再以查询字符串参数的形式放在请求的URL中，而是会被写入HTTP的BODY中。我们也将这一过程用一幅示意图的形式总结一下：

由于表单参数被放置在HTTP Body中传输(body中的数据为：language=go&since=monthly)，因此在该HTTP请求的headers中我们会发现新增一个header字段：Content-Type，在这里例子中，它的值为application/x-www-form-urlencoded。我们可以在Form中使用enctype属性改变Form传输数据的内容编码类型，该属性的默认值就是application/x-www-form-urlencoded(即key1=value1&key2=value2&…的形式)。enctype的其它可选值还包括：

• text/plain
• multipart/form-data

采用method=get的Form的表单参数以查询字符串参数的形式放入http请求，这使得其应用场景相对局限，比如：

• 当参数值很多，参数值很长时，可能会超出URL最大长度限制；
• 传递敏感数据时，参数值以明文放在HTTP请求头是不安全的；
• 无法胜任传递二进制数据(比如一个文件内容)的情形。

因此，在面对上述这些情形时，method=post的表单更有优势。当enctype为不同值时，method=post的表单在http Body中传输的数据形式如下图：

我们看到：enctype=application/x-www-urlencoded时，Body中的数据呈现为key1=value1&key2=value2&…的形式，好似URL的查询字符串参数的组合呈现形式；当enctype=text/plain时，这种编码格式也称为raw，即将数据内容原封不动的放入Body中传输，保持数据的原先的编码方式(通常为utf-8)；而当enctype=multipart/form-data时，HTTP Body中的数据以多段(part)的形式呈现，段与段之间使用指定的随机字符串分隔，该随机字符串也会随着HTTP Post请求一并传给服务端(放在Header中的Content-Type的值中，与multipart/form-data使用分号相隔)，如：

Content-Type: multipart/form-data; boundary=--------------------------399501358433894470769897

我们来看一个稍微复杂些的enctype=multipart/form-data的例子的示意图：

我们用Postman模拟了一个包含5个分段(part)的Post请求，其中包含两个文本分段(text)和三个文件分段，并且这三个文件是不同格式的文件，分别是txt，png和json。针对文件分段，Postman使用每个分段中的Content-Type来指明这个分段的数据内容类型。当服务端接收到这些数据时，根据分段Content-Type的指示，便可以有针对性的对分段数据进行解析了。文件分段的默认Content-Type为text/plain；对于无法识别的文件类型（比如：没有扩展名），文件分段的Content-Type通常会设置为application/octet-stream。

通过Form上传文件是RFC1867规范赋予html的一种能力，并且该能力已被证明非常有用，并被广泛使用，甚至我们可以直接将multipart/form-data作为HTTP Post body的一种数据承载协议在两个端之间传输文件数据。

2. 支持以multipart/form-data格式上传文件的Go服务器

http.Request提供了ParseMultipartForm的方法对以multipart/form-data格式传输的数据进行解析，解析即是将数据映射为Request结构的MultipartForm字段的过程：

// $GOROOT/src/net/http/request.go

type Request struct {
    ... ...
    // MultipartForm is the parsed multipart form, including file uploads.
    // This field is only available after ParseMultipartForm is called.
    // The HTTP client ignores MultipartForm and uses Body instead.
    MultipartForm *multipart.Form
    ... ...
}

multipart.Form代表了一个解析后的multipart/form-data的Body，其结构如下:

// $GOROOT/src/mime/multipart/formdata.go

// Form is a parsed multipart form.
// Its File parts are stored either in memory or on disk,
// and are accessible via the *FileHeader's Open method.
// Its Value parts are stored as strings.
// Both are keyed by field name.
type Form struct {
        Value map[string][]string
        File  map[string][]*FileHeader
}

我们看到这个Form结构由两个map组成，一个map中存放了所有的value part(就像前面的name、age)，另外一个map存放了所有的file part(就像前面的part1.txt、part2.png和part3.json)。value part集合没什么可说的，map的key就是每个值分段中的”name”； 我们的重点在file part上。每个file part对应一组FileHeader，FileHeader的结构如下：

// $GOROOT/src/mime/multipart/formdata.go
type FileHeader struct {
        Filename string
        Header   textproto.MIMEHeader
        Size     int64

        content []byte
        tmpfile string
}

每个file part的FileHeader包含五个字段：

• Filename – 上传文件的原始文件名
• Size – 上传文件的大小（单位：字节）
• content – 内存中存储的上传文件的（部分或全部）数据内容
• tmpfile – 在服务器本地的临时文件中存储的部分上传文件的数据内容(如果上传的文件大小大于传给ParseMultipartForm的参数maxMemory，剩余部分存储在临时文件中)
• Header – file part的header内容，它亦是一个map，其结构如下：

// $GOROOT/src/net/textproto/header.go

// A MIMEHeader represents a MIME-style header mapping
// keys to sets of values.
type MIMEHeader map[string][]string

我们可以将ParseMultipartForm方法实现的数据映射过程表述为下面这张示意图，这样看起来更为直观：

有了上述对通过multipart/form-data格式上传文件的原理的拆解，我们就可以很容易地利用Go http包实现一个简单的支持以multipart/form-data格式上传文件的Go服务器：

// github.com/bigwhite/experiments/multipart-formdata/server/file_server1.go
package main

import (
    "fmt"
    "io"
    "net/http"
    "os"
)

const uploadPath = "./upload"

func handleUploadFile(w http.ResponseWriter, r *http.Request) {
    r.ParseMultipartForm(100)
    mForm := r.MultipartForm

    for k, _ := range mForm.File {
        // k is the key of file part
        file, fileHeader, err := r.FormFile(k)
        if err != nil {
            fmt.Println("inovke FormFile error:", err)
            return
        }
        defer file.Close()
        fmt.Printf("the uploaded file: name[%s], size[%d], header[%#v]\n",
            fileHeader.Filename, fileHeader.Size, fileHeader.Header)

        // store uploaded file into local path
        localFileName := uploadPath + "/" + fileHeader.Filename
        out, err := os.Create(localFileName)
        if err != nil {
            fmt.Printf("failed to open the file %s for writing", localFileName)
            return
        }
        defer out.Close()
        _, err = io.Copy(out, file)
        if err != nil {
            fmt.Printf("copy file err:%s\n", err)
            return
        }
        fmt.Printf("file %s uploaded ok\n", fileHeader.Filename)
    }
}

func main() {
    http.HandleFunc("/upload", handleUploadFile)
    http.ListenAndServe(":8080", nil)
}

我们可以用Postman或下面curl命令向上述文件服务器同时上传两个文件part1.txt和part3.json：

curl --location --request POST ':8080/upload' \
--form 'name="tony bai"' \
--form 'age="23"' \
--form 'file1=@"/your_local_path/part1.txt"' \
--form 'file3=@"/your_local_path/part3.json"'

文件上传服务器的运行输出日志如下：

$go run file_server1.go
the uploaded file: name[part3.json], size[130], header[textproto.MIMEHeader{"Content-Disposition":[]string{"form-data; name=\"file3\"; filename=\"part3.json\""}, "Content-Type":[]string{"application/json"}}]
file part3.json uploaded ok
the uploaded file: name[part1.txt], size[15], header[textproto.MIMEHeader{"Content-Disposition":[]string{"form-data; name=\"file1\"; filename=\"part1.txt\""}, "Content-Type":[]string{"text/plain"}}]
file part1.txt uploaded ok

之后我们可以看到：文件上传服务器成功地将接收到的part1.txt和part3.json存储到了当前路径下的upload目录中了！

3. 支持以multipart/form-data格式上传文件的Go客户端

前面进行文件上传的客户端要么是浏览器，要么是Postman，要么是curl，如果我们自己构要造一个支持以multipart/form-data格式上传文件的客户端，应该如何做呢？我们需要按照multipart/form-data的格式构造HTTP请求的包体(Body)，还好通过Go标准库提供的mime/multipart包，我们可以很容易地构建出满足要求的包体：

// github.com/bigwhite/experiments/multipart-formdata/client/client1.go

... ...
var (
    filePath string
    addr     string
)

func init() {
    flag.StringVar(&filePath, "file", "", "the file to upload")
    flag.StringVar(&addr, "addr", "localhost:8080", "the addr of file server")
    flag.Parse()
}

func main() {
    if filePath == "" {
        fmt.Println("file must not be empty")
        return
    }

    err := doUpload(addr, filePath)
    if err != nil {
        fmt.Printf("upload file [%s] error: %s", filePath, err)
        return
    }
    fmt.Printf("upload file [%s] ok\n", filePath)
}

func createReqBody(filePath string) (string, io.Reader, error) {
    var err error

    buf := new(bytes.Buffer)
    bw := multipart.NewWriter(buf) // body writer

    f, err := os.Open(filePath)
    if err != nil {
        return "", nil, err
    }
    defer f.Close()

    // text part1
    p1w, _ := bw.CreateFormField("name")
    p1w.Write([]byte("Tony Bai"))

    // text part2
    p2w, _ := bw.CreateFormField("age")
    p2w.Write([]byte("15"))

    // file part1
    _, fileName := filepath.Split(filePath)
    fw1, _ := bw.CreateFormFile("file1", fileName)
    io.Copy(fw1, f)

    bw.Close() //write the tail boundry
    return bw.FormDataContentType(), buf, nil
}

func doUpload(addr, filePath string) error {
    // create body
    contType, reader, err := createReqBody(filePath)
    if err != nil {
        return err
    }

    url := fmt.Sprintf("http://%s/upload", addr)
    req, err := http.NewRequest("POST", url, reader)

    // add headers
    req.Header.Add("Content-Type", contType)

    client := &http.Client{}
    resp, err := client.Do(req)
    if err != nil {
        fmt.Println("request send error:", err)
        return err
    }
    resp.Body.Close()
    return nil
}

显然上面这个client端的代码的核心是createReqBody函数：

• 该client在body中创建了三个分段，前两个分段仅仅是我为了演示如何创建text part而故意加入的，真正的上传文件客户端是不需要创建这两个分段(part)的；
• createReqBody使用bytes.Buffer作为http body的临时存储；
• 构建完body内容后，不要忘记调用multipart.Writer的Close方法以写入结尾的boundary标记。

我们使用这个客户端向前面的支持以multipart/form-data格式上传文件的服务器上传一个文件：

// 客户端
$go run client1.go -file hello.txt
upload file [hello.txt] ok

// 服务端
$go run file_server1.go

http request: http.Request{Method:"POST", URL:(*url.URL)(0xc00016e100), Proto:"HTTP/1.1", ProtoMajor:1, ProtoMinor:1, Header:http.Header{"Accept-Encoding":[]string{"gzip"}, "Content-Length":[]string{"492"}, "Content-Type":[]string{"multipart/form-data; boundary=b55090594eaa1aaac1abad1d89a77ae689130d79d6f66af82590036bd8ba"}, "User-Agent":[]string{"Go-http-client/1.1"}}, Body:(*http.body)(0xc000146380), GetBody:(func() (io.ReadCloser, error))(nil), ContentLength:492, TransferEncoding:[]string(nil), Close:false, Host:"localhost:8080", Form:url.Values{"age":[]string{"15"}, "name":[]string{"Tony Bai"}}, PostForm:url.Values{"age":[]string{"15"}, "name":[]string{"Tony Bai"}}, MultipartForm:(*multipart.Form)(0xc000110d50), Trailer:http.Header(nil), RemoteAddr:"[::1]:58569", RequestURI:"/upload", TLS:(*tls.ConnectionState)(nil), Cancel:(<-chan struct {})(nil), Response:(*http.Response)(nil), ctx:(*context.cancelCtx)(0xc0001463c0)}
the uploaded file: name[hello.txt], size[15], header[textproto.MIMEHeader{"Content-Disposition":[]string{"form-data; name=\"file1\"; filename=\"hello.txt\""}, "Content-Type":[]string{"application/octet-stream"}}]
file hello.txt uploaded ok

我们看到hello.txt这个文本文件被成功上传！

4. 自定义file分段中的header

从上面file_server1的输出来看，client1这个客户端上传文件时在file分段(part)中设置的Content-Type为默认的application/octet-stream。有时候，服务端可能会需要根据这个Content-Type做分类处理，需要客户端给出准确的值。上面的client1实现中，我们使用了multipart.Writer.CreateFormFile这个方法来创建file part：

// file part1
_, fileName := filepath.Split(filePath)
fw1, _ := bw.CreateFormFile("file1", fileName)
io.Copy(fw1, f)

下面是标准库中CreateFormFile方法的实现代码：

// $GOROOT/mime/multipart/writer.go
func (w *Writer) CreateFormFile(fieldname, filename string) (io.Writer, error) {
        h := make(textproto.MIMEHeader)
        h.Set("Content-Disposition",
                fmt.Sprintf(`form-data; name="%s"; filename="%s"`,
                        escapeQuotes(fieldname), escapeQuotes(filename)))
        h.Set("Content-Type", "application/octet-stream")
        return w.CreatePart(h)
}

我们看到无论待上传的文件是什么类型，CreateFormFile均将Content-Type置为application/octet-stream这一默认值。如果我们要自定义file part中Header字段Content-Type的值，我们就不能直接使用CreateFormFile，不过我们可以参考其实现：

// github.com/bigwhite/experiments/multipart-formdata/client/client2.go

var quoteEscaper = strings.NewReplacer("\\", "\\\\", `"`, "\\\"")

func escapeQuotes(s string) string {
    return quoteEscaper.Replace(s)
}

func createReqBody(filePath string) (string, io.Reader, error) {
    var err error

    buf := new(bytes.Buffer)
    bw := multipart.NewWriter(buf) // body writer

    f, err := os.Open(filePath)
    if err != nil {
        return "", nil, err
    }
    defer f.Close()

    // text part1
    p1w, _ := bw.CreateFormField("name")
    p1w.Write([]byte("Tony Bai"))

    // text part2
    p2w, _ := bw.CreateFormField("age")
    p2w.Write([]byte("15"))

    // file part1
    _, fileName := filepath.Split(filePath)
    h := make(textproto.MIMEHeader)
    h.Set("Content-Disposition",
        fmt.Sprintf(`form-data; name="%s"; filename="%s"`,
            escapeQuotes("file1"), escapeQuotes(fileName)))
    h.Set("Content-Type", "text/plain")
    fw1, _ := bw.CreatePart(h)
    io.Copy(fw1, f)

    bw.Close() //write the tail boundry
    return bw.FormDataContentType(), buf, nil
}

我们通过textproto.MIMEHeader实例来自定义file part的header部分，然后基于该实例调用CreatePart创建file part，之后将hello.txt的文件内容写到该part的header后面。

我们运行client2来上传hello.txt文件，在file_server侧，我们就能看到如下日志：

the uploaded file: name[hello.txt], size[15], header[textproto.MIMEHeader{"Content-Disposition":[]string{"form-data; name=\"file1\"; filename=\"hello.txt\""}, "Content-Type":[]string{"text/plain"}}]
file hello.txt uploaded ok

我们看到file part的Content-Type的值已经变为我们设定的text/plain了。

5. 解决上传大文件的问题

在上面的客户端中存在一个问题，那就是我们在构建http body的时候，使用了一个bytes.Buffer加载了待上传文件的所有内容，这样一来，如果待上传的文件很大的话，内存空间消耗势必过大。那么如何将每次上传内存文件时对内存的使用限制在一个适当的范围，或者说上传文件所消耗的内存空间不因待传文件的变大而变大呢？我们来看下面的这个解决方案：

// github.com/bigwhite/experiments/multipart-formdata/client/client3.go
... ...
func createReqBody(filePath string) (string, io.Reader, error) {
    var err error
    pr, pw := io.Pipe()
    bw := multipart.NewWriter(pw) // body writer
    f, err := os.Open(filePath)
    if err != nil {
        return "", nil, err
    }

    go func() {
        defer f.Close()
        // text part1
        p1w, _ := bw.CreateFormField("name")
        p1w.Write([]byte("Tony Bai"))

        // text part2
        p2w, _ := bw.CreateFormField("age")
        p2w.Write([]byte("15"))

        // file part1
        _, fileName := filepath.Split(filePath)
        h := make(textproto.MIMEHeader)
        h.Set("Content-Disposition",
            fmt.Sprintf(`form-data; name="%s"; filename="%s"`,
                escapeQuotes("file1"), escapeQuotes(fileName)))
        h.Set("Content-Type", "application/pdf")
        fw1, _ := bw.CreatePart(h)
        cnt, _ := io.Copy(fw1, f)
        log.Printf("copy %d bytes from file %s in total\n", cnt, fileName)
        bw.Close() //write the tail boundry
        pw.Close()
    }()
    return bw.FormDataContentType(), pr, nil
}

func doUpload(addr, filePath string) error {
    // create body
    contType, reader, err := createReqBody(filePath)
    if err != nil {
        return err
    }

    log.Printf("createReqBody ok\n")
    url := fmt.Sprintf("http://%s/upload", addr)
    req, err := http.NewRequest("POST", url, reader)

    //add headers
    req.Header.Add("Content-Type", contType)

    client := &http.Client{}
    log.Printf("upload %s...\n", filePath)
    resp, err := client.Do(req)
    if err != nil {
        fmt.Println("request send error:", err)
        return err
    }
    resp.Body.Close()
    log.Printf("upload %s ok\n", filePath)
    return nil
}

在这个方案中，我们通过io.Pipe函数创建了一个读写管道，其写端作为io.Writer实例传给multipart.NewWriter，读端返回给调用者，用于构建http request时使用。io.Pipe基于channel实现，其内部不维护任何内存缓存：

// $GOROOT/src/io/pipe.go
func Pipe() (*PipeReader, *PipeWriter) {
        p := &pipe{
                wrCh: make(chan []byte),
                rdCh: make(chan int),
                done: make(chan struct{}),
        }
        return &PipeReader{p}, &PipeWriter{p}
}

通过Pipe返回的读端读取管道中数据时，如果尚未有数据写入管道，那么读端会像读取channel那样阻塞在那里。由于http request在被发送时(client.Do(req))才会真正基于构建req时传入的reader对Body数据进行读取，因此client会阻塞在对管道的read上。显然我们不能将读写两端的操作放在一个goroutine中，那样会因所有goroutine都挂起而导致panic。在上面的client3.go代码中，函数createReqBody内部创建了一个新goroutine，将真正构建multipart/form-data body的工作放在了新goroutine中。新goroutine最终会将待上传文件的数据通过管道写端写入管道：

cnt, _ := io.Copy(fw1, f)

而这些数据也会被client读取并通过网络连接传输出去。io.Copy的实现如下：

// $GOROOT/src/io/io.go
func Copy(dst Writer, src Reader) (written int64, err error) {
        return copyBuffer(dst, src, nil)
}

io.copyBuffer内部维护了一个默认32k的小buffer，它每次从src尝试最大读取32k的数据，并写入到dst中，直到读完为止。这样无论待上传的文件有多大，我们实际上每次上传所分配的内存仅有32k。

下面就是我们用client3.go上传一个大小为252M的文件的日志：

$go run client3.go -file /Users/tonybai/Downloads/ICME-2019-Tutorial-final.pdf
2021/01/10 12:56:45 createReqBody ok
2021/01/10 12:56:45 upload /Users/tonybai/Downloads/ICME-2019-Tutorial-final.pdf...
2021/01/10 12:56:46 copy 264517032 bytes from file ICME-2019-Tutorial-final.pdf in total
2021/01/10 12:56:46 upload /Users/tonybai/Downloads/ICME-2019-Tutorial-final.pdf ok
upload file [/Users/tonybai/Downloads/ICME-2019-Tutorial-final.pdf] ok

$go run file_server1.go
http request: http.Request{Method:"POST", URL:(*url.URL)(0xc000078200), Proto:"HTTP/1.1", ProtoMajor:1, ProtoMinor:1, Header:http.Header{"Accept-Encoding":[]string{"gzip"}, "Content-Type":[]string{"multipart/form-data; boundary=4470ba3867218f1130878713da88b5bd79f33dfbed65566e4fd76a1ae58d"}, "User-Agent":[]string{"Go-http-client/1.1"}}, Body:(*http.body)(0xc000026240), GetBody:(func() (io.ReadCloser, error))(nil), ContentLength:-1, TransferEncoding:[]string{"chunked"}, Close:false, Host:"localhost:8080", Form:url.Values{"age":[]string{"15"}, "name":[]string{"Tony Bai"}}, PostForm:url.Values{"age":[]string{"15"}, "name":[]string{"Tony Bai"}}, MultipartForm:(*multipart.Form)(0xc0000122a0), Trailer:http.Header(nil), RemoteAddr:"[::1]:54899", RequestURI:"/upload", TLS:(*tls.ConnectionState)(nil), Cancel:(<-chan struct {})(nil), Response:(*http.Response)(nil), ctx:(*context.cancelCtx)(0xc000026280)}
the uploaded file: name[ICME-2019-Tutorial-final.pdf], size[264517032], header[textproto.MIMEHeader{"Content-Disposition":[]string{"form-data; name=\"file1\"; filename=\"ICME-2019-Tutorial-final.pdf\""}, "Content-Type":[]string{"application/pdf"}}]
file ICME-2019-Tutorial-final.pdf uploaded ok

$ls -l upload
-rw-r--r--  1 tonybai  staff  264517032  1 14 12:56 ICME-2019-Tutorial-final.pdf

如果你觉得32k仍然很大，每次上传要使用更小的buffer，你可以用io.CopyBuffer替代io.Copy：

// github.com/bigwhite/experiments/multipart-formdata/client/client4.go

func createReqBody(filePath string) (string, io.Reader, error) {
    var err error
    pr, pw := io.Pipe()
    bw := multipart.NewWriter(pw) // body writer
    f, err := os.Open(filePath)
    if err != nil {
        return "", nil, err
    }

    go func() {
        defer f.Close()
        // text part1
        p1w, _ := bw.CreateFormField("name")
        p1w.Write([]byte("Tony Bai"))

        // text part2
        p2w, _ := bw.CreateFormField("age")
        p2w.Write([]byte("15"))

        // file part1
        _, fileName := filepath.Split(filePath)
        h := make(textproto.MIMEHeader)
        h.Set("Content-Disposition",
            fmt.Sprintf(`form-data; name="%s"; filename="%s"`,
                escapeQuotes("file1"), escapeQuotes(fileName)))
        h.Set("Content-Type", "application/pdf")
        fw1, _ := bw.CreatePart(h)
        var buf = make([]byte, 1024)
        cnt, _ := io.CopyBuffer(fw1, f, buf)
        log.Printf("copy %d bytes from file %s in total\n", cnt, fileName)
        bw.Close() //write the tail boundry
        pw.Close()
    }()
    return bw.FormDataContentType(), pr, nil
}

运行这个client4：

$go run client4.go -file /Users/tonybai/Downloads/ICME-2019-Tutorial-final.pdf
2021/01/10 13:39:06 createReqBody ok
2021/01/10 13:39:06 upload /Users/tonybai/Downloads/ICME-2019-Tutorial-final.pdf...
2021/01/10 13:39:09 copy 264517032 bytes from file ICME-2019-Tutorial-final.pdf in total
2021/01/10 13:39:09 upload /Users/tonybai/Downloads/ICME-2019-Tutorial-final.pdf ok
upload file [/Users/tonybai/Downloads/ICME-2019-Tutorial-final.pdf] ok

你会看到虽然上传成功了，但由于每次read仅能读1k数据，对于大文件来说，其上传的时间消耗增加了不少。

6. 下载文件

客户端基于multipart/form-data下载文件的过程的原理与上面的file_server1接收客户端上传文件的原理是一样的，这里就将这个功能的Go实现作为“作业”留给各位读者了:)。

7. 参考资料

• Form-based File Upload in HTML
• Returning Values from Forms: multipart/form-data
• 《Go Web Programming》
• Hypertext Transfer Protocol (HTTP/1.1): Range Requests

本文中涉及的源码可以在这里(https://github.com/bigwhite/experiments/tree/master/multipart-formdata)下载。

“Gopher部落”知识星球开球了！高品质首发Go技术文章，“三天”首发阅读权，每年两期Go语言发展现状分析，每天提前1小时阅读到新鲜的Gopher日报，网课、技术专栏、图书内容前瞻，六小时内必答保证等满足你关于Go语言生态的所有需求！星球首开，福利自然是少不了的！2020年年底之前，8.8折(很吉利吧^_^)加入星球，下方图片扫起来吧！

Go技术专栏“改善Go语⾔编程质量的50个有效实践”正在慕课网火热热销中！本专栏主要满足广大gopher关于Go语言进阶的需求，围绕如何写出地道且高质量Go代码给出50条有效实践建议，上线后收到一致好评！欢迎大家订阅！

我的网课“Kubernetes实战：高可用集群搭建、配置、运维与应用”在慕课网热卖中，欢迎小伙伴们订阅学习！

我爱发短信：企业级短信平台定制开发专家 https://tonybai.com/
smspush : 可部署在企业内部的定制化短信平台，三网覆盖，不惧大并发接入，可定制扩展； 短信内容你来定，不再受约束, 接口丰富，支持长短信，签名可选。

2020年4月8日，中国三大电信运营商联合发布《5G消息白皮书》，51短信平台也会全新升级到“51商用消息平台”，全面支持5G RCS消息。

著名云主机服务厂商DigitalOcean发布最新的主机计划，入门级Droplet配置升级为：1 core CPU、1G内存、25G高速SSD，价格5$/月。有使用DigitalOcean需求的朋友，可以打开这个链接地址：https://m.do.co/c/bff6eed92687 开启你的DO主机之路。

Gopher Daily(Gopher每日新闻)归档仓库 – https://github.com/bigwhite/gopherdaily

我的联系方式：

• 微博：https://weibo.com/bigwhite20xx
• 微信公众号：iamtonybai
• 博客：tonybai.com
• github: https://github.com/bigwhite
• “Gopher部落”知识星球：https://public.zsxq.com/groups/51284458844544

微信赞赏：

商务合作方式：撰稿、出书、培训、在线课程、合伙创业、咨询、广告合作。