PostgreSQL上线优化案例QPS从500到上万级别压测之路

// WritePeer 在事务内，写一端的数据
func (pg *PostgreSQLProcessor) WritePeer
(ctx context.Context, req *graph.WriteReq, rsp *graph.WriteRsp) 
(int32, error) {
  pg.log.Debug("写tdsql一端数据:%+v", req)
  vertexConfig := pg.GetVertexConfig(&req.GraphBasic)
  if vertexConfig.IsStoreOfflineEdges == false {
     本端不用写
    return 0, nil
  }
  targetsIDs := make([]string, 0)
  for _, v := range req.TargetsIDs {
    targetsIDs = append(targetsIDs, v.ID)
  }
  tablename := pg.GetTableName(req.GraphBasic.NodeType)
  pg.log.Debug("tablename is %v", tablename)
  degreeChange := 0
   首先检索数据存在
   args := make([]interface{}, 0)
   args = append(args, req.SourceID)
   for _, id := range targetsIDs {
     args = append(args, id)
   }

  rows, sqlerr := pg.tx.QueryContext(ctx, fmt.Sprintf
  ("select id,target,unixnano,status from %s where id = $1 and target = ANY($2) for update", tablename), 
  req.SourceID, pq.Array(targetsIDs))


  if sqlerr != nil {
    pg.log.Error("检索已有数据失败:%s", sqlerr)
    return -7700, sqlerr
  }
  defer rows.Close()


  edgeRows := make([]EdgeRow, 0)


  for rows.Next() {
    e := EdgeRow{}
    err := rows.Scan(&e.ID, &e.Target, &e.Unixnano, &e.Status)
    if err != nil {
      return -7800, err
    }
    edgeRows = append(edgeRows, e)
  }
  pg.log.Debug("检索到数据 %+v", edgeRows)


  // 记下数据库时间戳
  timeMap := make(map[string]*EdgeRow)
  for _, r := range edgeRows {
    timeMap[r.Target] = &r
  }


  pg.log.Debug("检索到数据时间戳 %+v", timeMap)


  // 根据情况写入数据
  for _, target := range req.TargetsIDs {
    dbrow, ok := timeMap[target.ID]
    if ok {
      if dbrow.Unixnano >= target.UnixNano {
        // 数据库中的时间大 不处理
        pg.log.Debug("数据库中的时间大 不处理")
      }
      if dbrow.Unixnano < target.UnixNano && dbrow.Status == req.Operation {
        // 数据库中时间小 但是状态与数据库中的时间一致  则只刷新时间
        pg.log.Debug("数据库中时间小 但是写入状态与数据库中的一致  则只刷新时间")
        rsp.RefreshIDs = append(rsp.RefreshIDs, target)
        _, sqlerr = pg.tx.ExecContext(ctx, fmt.Sprintf
        ("update %s set unixnano = $1 where id = $2 and target = $3", tablename), 
        target.UnixNano, req.SourceID, target.ID)
        if sqlerr != nil {
          pg.log.Error("更新数据失败:%s", sqlerr)
          return -7901, sqlerr
        }
      }
      if dbrow.Unixnano < target.UnixNano && dbrow.Status != req.Operation {
        // 数据库时间小 但是状态与数据库中不一致 覆盖时间和操作
        pg.log.Debug("数据库时间小 但是状态与数据库中不一致 覆盖时间和操作")
        rsp.ExecutedIDs = append(rsp.ExecutedIDs, target)
        var result sql.Result
        result, sqlerr = pg.tx.ExecContext(ctx, fmt.Sprintf
        ("update %s set unixnano = $1 , status = $2 where id = $3 and target = $4", tablename), 
        target.UnixNano, req.Operation, req.SourceID, target.ID)
        if sqlerr != nil {
          pg.log.Error("更新数据失败:%s", sqlerr)
          return -7902, sqlerr
        }
        rowsAff, sqlerr := result.RowsAffected()
        if sqlerr != nil {
          pg.log.Error("获取不到更新行：%s", sqlerr)
        }
        if rowsAff > 0 {
          // 判断覆盖方式对度的影响
          if req.Operation == graph.ESet {
            degreeChange++
          } else if req.Operation == graph.EUnset {
            degreeChange--
          }
        }
        pg.log.Debug("更新了时间 %v", target.UnixNano)
      }
    } else {
      // 数据库中不存在数据 插入数据
      pg.log.Debug("数据库中不存在数据 %v 插入数据", target.ID)
      rsp.ExecutedIDs = append(rsp.ExecutedIDs, target)
      // 判断覆盖方式对度的影响
      if req.Operation == graph.ESet {
        degreeChange++
      }
      // 如果不存在数据时，插入了取消边操作，则度不变化
      _, sqlerr = pg.tx.ExecContext(ctx, fmt.Sprintf
      ("INSERT INTO %s (id, target, unixnano, status) 
      VALUES ($1, $2, $3, $4)", tablename),
        req.SourceID, target.ID, target.UnixNano, req.Operation)
      if sqlerr != nil {
        pg.log.Error("更新数据失败:%s", sqlerr)
        return -7903, sqlerr
      }
    }
  }
  if degreeChange != 0 {
    pg.log.Debug("度变化：%v", degreeChange)
    _, sqlerr = pg.tx.ExecContext(ctx, fmt.Sprintf
    ("INSERT INTO %s (id, degree) VALUES ($1, $2) on conflict(id) 
    do UPDATE set degree = %s.degree + $2", 
    tablename+degreeSuffix, tablename+degreeSuffix), req.SourceID, degreeChange)
    if sqlerr != nil {
      pg.log.Error("更新数据失败:%s", sqlerr)
      return -7904, sqlerr
    }
  }
  return 0, nil
}

read:
select target, unixnano from %s where id = $1 and status = 0 order by unixnano desc limit $2
select target, unixnano from %s where id = $1 and unixnano < $2 and status = 0 order by unixnano desc limit $3
select target, unixnano from %s where id = $1 and target = ANY($2) and status = 0
select degree from %s where id = $1
write:
select id,target,unixnano,status from %s where id = $1 and target = ANY($2) for update
update %s set unixnano = $1 where id = $2 and target = $3
update %s set unixnano = $1 , status = $2 where id = $3 and target = $4
INSERT INTO %s (id, target, unixnano, status) VALUES ($1, $2, $3, $4)
INSERT INTO %s (id, degree) VALUES ($1, $2) on conflict(id) do UPDATE set degree = %s.degree + $2
select target, unixnano from %s where id = $1 and status = 0 order by unixnano desc limit $2

INSERT INTO %s (id, target, unixnano, status) VALUES 
(123, 456, 789, 0) on conflict(id, target) 
do UPDATE set unixnano = $1, status = $2;

moment_fav_test=# explain analyze INSERT INTO vertexa 
(id, target, unixnano, status) VALUES (123, 456, 789, 0) on conflict
(id, target) do UPDATE set unixnano = 789, status = 0;
                                               QUERY PLAN                                                
----------------------------------------------------------
-----------------------------------------------
 Remote Fast Query Execution  
 (cost=0.00..0.00 rows=0 width=0) 
 (actual time=8.105..8.105 rows=0 loops=1)
   Node expr: 123
 Planning time: 0.087 ms
 Execution time: 8.134 ms
(4 rows)