```sql
INSERT INTO `ratings`
(`book_id`, `user_id`, `score`, `rated_at`)
VALUES
(1000, 1, 5, NOW())
ON DUPLICATE KEY UPDATE `score` = 5, `rated_at` = NOW();
```
```java
// ds is an entity of com.mysql.cj.jdbc.MysqlDataSource
try (Connection connection = ds.getConnection()) {
PreparedStatement p = connection.prepareStatement("INSERT INTO `ratings` (`book_id`, `user_id`, `score`, `rated_at`)
VALUES (?, ?, ?, NOW()) ON DUPLICATE KEY UPDATE `score` = ?, `rated_at` = NOW()");
p.setInt(1, 1000);
p.setInt(2, 1);
p.setInt(3, 5);
p.setInt(4, 5);
p.executeUpdate();
} catch (SQLException e) {
e.printStackTrace();
}
```
## 批量更新
需要更新表中多行的数据，可选择 [使用 `UPDATE`](#使用-update)，并使用 `WHERE` 子句过滤需要更新的数据。
但如果你需要更新大量行(数万或更多)的时候，我们建议使用一个迭代，每次都只更新一部分数据，直到更新全部完成。这是因为 TiDB 单个事务大小限制为 [txn-total-size-limit](https://docs.pingcap.com/zh/tidb/stable/tidb-configuration-file#txn-total-size-limit)（默认为 100MB），且一次性过多的数据更新，将导致持有锁时间过长（[悲观事务](https://docs.pingcap.com/zh/tidb/stable/pessimistic-transaction)），或产生大量冲突（[乐观事务](https://docs.pingcap.com/zh/tidb/stable/optimistic-transaction)）。您可以在程序或脚本中使用循环来完成操作。
本页提供了编写脚本来处理循环更新的示例，该示例演示了应如何进行 `SELECT` 和 `UPDATE` 的组合，完成循环更新。
### 编写批量更新循环
首先，您应在您的应用或脚本的循环中，编写一个 `SELECT` 查询。这个查询的返回值可以作为需要更新的行的主键。需要注意的是，定义这个 `SELECT` 查询时，需要注意使用 `WHERE` 子句过滤需要更新的行。
### 例子
假设我们在过去的一年里，用户在我们的 `bookshop` 网站进行了大量的书籍打分，但是原本设计为 5 分制的评分导致书籍评分的区分度不够，大量书籍评分集中在 3 分附近，因此，我们决定将 5 分制改为 10 分制。用来增大书籍评分的区分度。
这时我们需要对 `ratings` 表内之前 5 分制的数据进行乘 2 操作，同时需向 `ratings` 表内添加一个新列，以指示行是否已经被更新了。使用此列，我们可以在 `SELECT` 中过滤掉已经更新的行，这将防止脚本崩溃时对行进行多次更新，导致不合理的数据出现。
例如，您可以创建一个名为 `ten_point`，数据类型为 [BOOL](https://docs.pingcap.com/zh/tidb/stable/data-type-numeric#boolean-%E7%B1%BB%E5%9E%8B) 的列作为是否为 10 分制的标识：
```sql
ALTER TABLE `bookshop`.`ratings` ADD COLUMN `ten_point` BOOL NOT NULL DEFAULT FALSE;
```
> Note:
>
> 此批量更新程序将使用 `DDL` 语句将进行数据表的模式更改。TiDB 的所有 DDL 变更操作全部都是在线进行的，可查看此处，了解此处使用的 [ADD COLUMN](https://docs.pingcap.com/zh/tidb/stable/sql-statement-add-column) 语句。
在 Golang 中，批量更新程序可能会类似于以下内容：
```go
package main
import (
"database/sql"
"fmt"
_ "github.com/go-sql-driver/mysql"
"strings"
"time"
)
func main() {
db, err := sql.Open("mysql", "root:@tcp(127.0.0.1:4000)/bookshop")
if err != nil {
panic(err)
}
defer db.Close()
bookID, userID := updateBatch(db, true, 0, 0)
fmt.Println("first time batch update success")
for {
time.Sleep(time.Second)
bookID, userID = updateBatch(db, false, bookID, userID)
fmt.Printf("batch update success, [bookID] %d, [userID] %d\n", bookID, userID)
}
}
// updateBatch select at most 1000 lines data to update score
func updateBatch(db *sql.DB, firstTime bool, lastBookID, lastUserID int64) (bookID, userID int64) {
// select at most 1000 primary keys in five-point scale data
var err error
var rows *sql.Rows
if firstTime {
rows, err = db.Query("SELECT `book_id`, `user_id` FROM `bookshop`.`ratings` " +
"WHERE `ten_point` != true ORDER BY `book_id`, `user_id` LIMIT 1000")
} else {
rows, err = db.Query("SELECT `book_id`, `user_id` FROM `bookshop`.`ratings` "+
"WHERE `ten_point` != true AND `book_id` > ? AND `user_id` > ? "+
"ORDER BY `book_id`, `user_id` LIMIT 1000", lastBookID, lastUserID)
}
if err != nil || rows == nil {
panic(fmt.Errorf("error occurred or rows nil: %+v", err))
}
// joint all id with a list
var idList []interface{}
for rows.Next() {
var tempBookID, tempUserID int64
if err := rows.Scan(&tempBookID, &tempUserID); err != nil {
panic(err)
}
idList = append(idList, tempBookID, tempUserID)
bookID, userID = tempBookID, tempUserID
}
bulkUpdateSql := fmt.Sprintf("UPDATE `bookshop`.`ratings` SET `ten_point` = true, "+
"`score` = `score` * 2 WHERE (`book_id`, `user_id`) IN (%s)", placeHolder(len(idList)))
db.Exec(bulkUpdateSql, idList...)
return bookID, userID
}
// placeHolder format SQL place holder
func placeHolder(n int) string {
holderList := make([]string, n/2, n/2)
for i := range holderList {
holderList[i] = "(?,?)"
}
return strings.Join(holderList, ",")
}
```
每次迭代中，`SELECT` 按主键顺序进行查询，最多选择 1000 行未更新到 10 分制（`ten_point` 为 `false`）数据的主键值。每次 `SELECT` 都会选择比上一次 `SELECT` 结果的最大主键还要大的数据，防止重复。然后，使用批量更新的方式，对其 `score` 列乘 2，并且将 `ten_point` 设为 `true`，更新 `ten_point` 的意义是在于防止我们的更新程序崩溃重启后，反复更新同一行数据，导致数据损坏。每次循环中的 `time.Sleep(time.Second)` 将使得更新程序暂停1秒，防止批量更新程序占用过多的硬件资源。
在 Java (JDBC) 中，批量更新程序可能会类似于以下内容：
- Java 代码部分：
```java
package com.pingcap.bulkUpdate;
import com.mysql.cj.jdbc.MysqlDataSource;
import java.sql.*;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;
public class BatchUpdateExample {
static class UpdateID {
private Long bookID;
private Long userID;
public UpdateID(Long bookID, Long userID) {
this.bookID = bookID;
this.userID = userID;
}
public Long getBookID() {
return bookID;
}
public void setBookID(Long bookID) {
this.bookID = bookID;
}
public Long getUserID() {
return userID;
}
public void setUserID(Long userID) {
this.userID = userID;
}
@Override
public String toString() {
return "[bookID] " + bookID + ", [userID] " + userID ;
}
}
public static void main(String[] args) throws InterruptedException {
// Configure the example database connection.
// Create a mysql data source instance.
MysqlDataSource mysqlDataSource = new MysqlDataSource();
// Set server name, port, database name, username and password.
mysqlDataSource.setServerName("localhost");
mysqlDataSource.setPortNumber(4000);
mysqlDataSource.setDatabaseName("bookshop");
mysqlDataSource.setUser("root");
mysqlDataSource.setPassword("");
UpdateID lastID = batchUpdate(mysqlDataSource, null);
System.out.println("first time batch update success");
while (true) {
TimeUnit.SECONDS.sleep(1);
lastID = batchUpdate(mysqlDataSource, lastID);
System.out.println("batch update success, [lastID] " + lastID);
}
}
public static UpdateID batchUpdate (MysqlDataSource ds, UpdateID lastID) {
try (Connection connection = ds.getConnection()) {
UpdateID updateID = null;
PreparedStatement selectPs;
if (lastID == null) {
selectPs = connection.prepareStatement(
"SELECT `book_id`, `user_id` FROM `bookshop`.`ratings` " +
"WHERE `ten_point` != true ORDER BY `book_id`, `user_id` LIMIT 1000");
} else {
selectPs = connection.prepareStatement(
"SELECT `book_id`, `user_id` FROM `bookshop`.`ratings` "+
"WHERE `ten_point` != true AND `book_id` > ? AND `user_id` > ? "+
"ORDER BY `book_id`, `user_id` LIMIT 1000");
selectPs.setLong(1, lastID.getBookID());
selectPs.setLong(2, lastID.getUserID());
}
List
idList = new LinkedList<>();
ResultSet res = selectPs.executeQuery();
while (res.next()) {
updateID = new UpdateID(
res.getLong("book_id"),
res.getLong("user_id")
);
idList.add(updateID.getBookID());
idList.add(updateID.getUserID());
}
if (idList.isEmpty()) {
System.out.println("no data should update");
return null;
}
String updateSQL = "UPDATE `bookshop`.`ratings` SET `ten_point` = true, "+
"`score` = `score` * 2 WHERE (`book_id`, `user_id`) IN (" +
placeHolder(idList.size() / 2) + ")";
PreparedStatement updatePs = connection.prepareStatement(updateSQL);
for (int i = 0; i < idList.size(); i++) {
updatePs.setLong(i + 1, idList.get(i));
}
int count = updatePs.executeUpdate();
System.out.println("update " + count + " data");
return updateID;
} catch (SQLException e) {
e.printStackTrace();
}
return null;
}
public static String placeHolder(int n) {
StringBuilder sb = new StringBuilder();
for (int i = 0; i < n ; i++) {
sb.append(i == 0 ? "(?,?)" : ",(?,?)");
}
return sb.toString();
}
}
```
- `hibernate.cfg.xml` 配置部分：
```xml
com.mysql.cj.jdbc.Driver
org.hibernate.dialect.TiDBDialect
jdbc:mysql://localhost:4000/movie
root
false
20
true
true
```
每次迭代中，`SELECT` 按主键顺序进行查询，最多选择 1000 行未更新到 10 分制（`ten_point` 为 `false`）数据的主键值。每次 `SELECT` 都会选择比上一次 `SELECT` 结果的最大主键还要大的数据，防止重复。然后，使用批量更新的方式，对其 `score` 列乘 2，并且将 `ten_point` 设为 `true`，更新 `ten_point` 的意义是在于防止我们的更新程序崩溃重启后，反复更新同一行数据，导致数据损坏。每次循环中的 `TimeUnit.SECONDS.sleep(1);` 将使得更新程序暂停1秒，防止批量更新程序占用过多的硬件资源。