1

我有一个类似于以下的场景： 

txf := func(tx *redis.Tx) error {
  // Phase 1:  
  // read some stuff by several get/hget request which I want to send with pipeline
  // to avoid unnecessarily rounds to the redis server 

  // Phase 2: Prepare new data based on read data

  // Phase 3: Write the new data with a transaction to use the watch protection if original keys changed
  _, err = tx.Pipelined(func(pipe redis.Pipeliner) error {
             // pipe handles the error case
             pipe.Set(key, value, 0)
             return nil})
  return err
}
err := client.Watch(txf, key)

从阅读库代码看来， tx.TxPipeline 和 tx.Pipeline 在这种情况下都返回了一个 multi/exec 实现，这意味着如果我将它用于读取数据（阶段 1），我将在我用来更新数据的第二个管道。
任何解决方案都将得到应用。

4

1 回答 1

1

您可以使用client.Pipelined(...而不是使用管道tx..Pipelined(...，但它将使用来自 go-redis 池的另一个连接（从 redis 服务器角度来看的另一个客户端）发送到 redis 服务器。我不认为这是一个问题。

go-redis事务使用粘性连接来确保以 开头的整个事务WATCH是从同一个连接发送的。内部tx.baseClient未导出。无法使用相同的连接发送管道。

txf := func(tx *redis.Tx) error {
  // Phase 1:  
  var getPipe *redis.StringCmd
  cmds, err := client.Pipelined(func(pipe redis.Pipeliner) error {
        getPipe = pipe.Get("getPipe")
        pipe.Set("pipe1", "p1", 0)
        return nil
  })
  fmt.Println(getPipe)
  fmt.Println(cmds)
  val, _ := getPipe.Result()
  fmt.Println("Value read for 'getPipe':", val)

  // Phase 2: Prepare new data based on read data

  // Phase 3
  _, err = tx.Pipelined(func(pipe redis.Pipeliner) error {
    // pipe handles the error case
    pipe.Set(key, value, 0)
    return nil})
  return err
}
err := client.Watch(txf, key)
fmt.Println(client.Get(key), err)

上述 go 程序的输出：

get getPipe: preVal
[get getPipe: preVal set pipe1 p1: OK]
Value read for 'getPipe': preVal
get myKey: insideMulti <nil>

这就是我在 redis-cli 上看到的MONITOR命令：

1 ...1:65506] "watch" "myKey"
2 ...1:65507] "get" "getPipe"
3 ...1:65507] "set" "pipe1" "p1"
4 ...1:65506] "MULTI"
5 ...1:65506] "set" "myKey" "insideMulti"
6 ...1:65506] "EXEC"
7 ...1:65506] "unwatch"
8 ...1:65506] "get" "myKey"

注意第 2 行和第 3 行在不同的端口上

于 2020-01-10T11:19:41.810 回答