Repository: tikv/client-go
Pull request #1646 (2025-05-28 15:22:28)

*: support async batch get
ref #1586

ti-chi-bot[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/tikv/client-go/pull/1646#pullrequestreview-2913085061" title="Approved">cfzjywxk</a>*

The full list of commands accepted by this bot can be found [here](https://prow.tidb.net/command-help?repo=tikv%2Fclient-go).

The pull request process is described [here](https://book.prow.tidb.net/#/workflows/pr)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/tikv/client-go/blob/master/OWNERS)~~ [cfzjywxk]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->

Patch:
diff --git a/config/config.go b/config/config.go
index 8b06264..a23c1ff 100644
--- a/config/config.go
+++ b/config/config.go
@@ -82,6 +82,8 @@ type Config struct {
 	RegionsRefreshInterval uint64
 	// EnablePreload indicates whether to preload region info when initializing the client.
 	EnablePreload bool
+	// EnableAsyncBatchGet indicates whether to use async api for batch get.
+	EnableAsyncBatchGet bool
 }
 
 // DefaultConfig returns the default configuration.
@@ -99,6 +101,7 @@ func DefaultConfig() Config {
 		TxnScope:              "",
 		EnableAsyncCommit:     false,
 		Enable1PC:             false,
+		EnableAsyncBatchGet:   false,
 	}
 }
 
diff --git a/internal/client/client.go b/internal/client/client.go
index 6af800b..46ee314 100644
--- a/internal/client/client.go
+++ b/internal/client/client.go
@@ -638,14 +638,6 @@ func (c *RPCClient) sendRequest(ctx context.Context, addr string, req *tikvrpc.R
 		elapsed := time.Since(start)
 		connArray.updateRPCMetrics(req, resp, elapsed)
 
-		if stmtExec := ctx.Value(util.ExecDetailsKey); stmtExec != nil {
-			execDetails := stmtExec.(*util.ExecDetails)
-			atomic.AddInt64(&execDetails.WaitKVRespDuration, int64(elapsed))
-			execNetworkCollector := networkCollector{}
-			execNetworkCollector.onReq(req, execDetails)
-			execNetworkCollector.onResp(req, resp, execDetails)
-		}
-
 		if spanRPC != nil && util.TraceExecDetailsEnabled(ctx) {
 			if si := buildSpanInfoFromResp(resp); si != nil {
 				si.addTo(spanRPC, start)
diff --git a/internal/client/client_async.go b/internal/client/client_async.go
index c33b8c0..fc2d253 100644
--- a/internal/client/client_async.go
+++ b/internal/client/client_async.go
@@ -121,15 +121,6 @@ func (c *RPCClient) SendRequestAsync(ctx context.Context, addr string, req *tikv
 		// rpc metrics
 		connArray.updateRPCMetrics(req, resp, elapsed)
 
-		// resource control
-		if stmtExec := ctx.Value(util.ExecDetailsKey); stmtExec != nil {
-			execDetails := stmtExec.(*util.ExecDetails)
-			atomic.AddInt64(&execDetails.WaitKVRespDuration, int64(elapsed))
-			execNetworkCollector := networkCollector{}
-			execNetworkCollector.onReq(req, execDetails)
-			execNetworkCollector.onResp(req, resp, execDetails)
-		}
-
 		// tracing
 		if spanRPC != nil {
 			if util.TraceExecDetailsEnabled(ctx) {
diff --git a/internal/client/client_collapse.go b/internal/client/client_collapse.go
index 3ea77b9..5cf5043 100644
--- a/internal/client/client_collapse.go
+++ b/internal/client/client_collapse.go
@@ -40,6 +40,7 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/gogo/protobuf/proto"
 	"github.com/pkg/errors"
 	"github.com/tikv/client-go/v2/tikvrpc"
 	"github.com/tikv/client-go/v2/util/async"
@@ -136,6 +137,9 @@ func (r reqCollapse) collapse(ctx context.Context, key string, sf *singleflight.
 	addr string, req *tikvrpc.Request, timeout time.Duration) (resp *tikvrpc.Response, err error) {
 	// because the request may be used by other goroutines, copy the request to avoid data race.
 	copyReq := *req
+	if req.Type == tikvrpc.CmdResolveLock && req.Req != nil {
+		copyReq.Req = proto.Clone(req.ResolveLock())
+	}
 	rsC := sf.DoChan(key, func() (interface{}, error) {
 		return r.Client.SendRequest(context.Background(), addr, &copyReq, ReadTimeoutShort) // use resolveLock timeout.
 	})
diff --git a/internal/client/metrics_collector.go b/internal/locate/metrics_collector.go
similarity index 73%
rename from internal/client/metrics_collector.go
rename to internal/locate/metrics_collector.go
index 977ad7e..259b4c3 100644
--- a/internal/client/metrics_collector.go
+++ b/internal/locate/metrics_collector.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package client
+package locate
 
 import (
 	"sync/atomic"
@@ -20,36 +20,15 @@ import (
 	"github.com/pingcap/kvproto/pkg/coprocessor"
 	"github.com/pingcap/kvproto/pkg/kvrpcpb"
 	"github.com/pingcap/kvproto/pkg/mpp"
+	"github.com/pingcap/kvproto/pkg/tikvpb"
 	"github.com/tikv/client-go/v2/kv"
 	"github.com/tikv/client-go/v2/metrics"
 	"github.com/tikv/client-go/v2/tikvrpc"
 	"github.com/tikv/client-go/v2/util"
 )
 
-type staleReadMetricsCollector struct {
-}
-
-func (s *staleReadMetricsCollector) onReq(size float64, isCrossZoneTraffic bool) {
-	if isCrossZoneTraffic {
-		metrics.StaleReadRemoteOutBytes.Add(float64(size))
-		metrics.StaleReadReqCrossZoneCounter.Add(1)
-
-	} else {
-		metrics.StaleReadLocalOutBytes.Add(float64(size))
-		metrics.StaleReadReqLocalCounter.Add(1)
-	}
-}
-
-func (s *staleReadMetricsCollector) onResp(size float64, isCrossZoneTraffic bool) {
-	if isCrossZoneTraffic {
-		metrics.StaleReadRemoteInBytes.Add(float64(size))
-	} else {
-		metrics.StaleReadLocalInBytes.Add(float64(size))
-	}
-}
-
 type networkCollector struct {
-	staleReadMetricsCollector
+	staleRead bool
 }
 
 func (s *networkCollector) onReq(req *tikvrpc.Request, details *util.ExecDetails) {
@@ -92,24 +71,25 @@ func (s *networkCollector) onReq(req *tikvrpc.Request, details *util.ExecDetails
 		// ignore others
 		return
 	}
-	isTiflashTarget := req.StoreTp == tikvrpc.TiFlash
-	var total, crossZone *int64
-	if isTiflashTarget {
-		total = &details.UnpackedBytesSentMPPTotal
-		crossZone = &details.UnpackedBytesSentMPPCrossZone
-	} else {
-		total = &details.UnpackedBytesSentKVTotal
-		crossZone = &details.UnpackedBytesSentKVCrossZone
-	}
-
-	atomic.AddInt64(total, int64(size))
 	isCrossZoneTraffic := req.AccessLocation == kv.AccessCrossZone
-	if isCrossZoneTraffic {
-		atomic.AddInt64(crossZone, int64(size))
+	if details != nil {
+		isTiflashTarget := req.StoreTp == tikvrpc.TiFlash
+		var total, crossZone *int64
+		if isTiflashTarget {
+			total = &details.UnpackedBytesSentMPPTotal
+			crossZone = &details.UnpackedBytesSentMPPCrossZone
+		} else {
+			total = &details.UnpackedBytesSentKVTotal
+			crossZone = &details.UnpackedBytesSentKVCrossZone
+		}
+		atomic.AddInt64(total, int64(size))
+		if isCrossZoneTraffic {
+			atomic.AddInt64(crossZone, int64(size))
+		}
 	}
 	// stale read metrics
-	if req.StaleRead {
-		s.staleReadMetricsCollector.onReq(float64(size), isCrossZoneTraffic)
+	if s.staleRead {
+		s.onReqStaleRead(float64(size), isCrossZoneTraffic)
 	}
 }
 
@@ -117,6 +97,9 @@ func (s *networkCollector) onResp(req *tikvrpc.Request, resp *tikvrpc.Response,
 	if resp == nil {
 		return
 	}
+	if _, ok := resp.Resp.(*tikvpb.BatchCommandsEmptyResponse); ok {
+		return
+	}
 	size := 0
 	switch req.Type {
 	case tikvrpc.CmdGet:
@@ -160,23 +143,46 @@ func (s *networkCollector) onResp(req *tikvrpc.Request, resp *tikvrpc.Response,
 		// ignore others
 		return
 	}
-	var total, crossZone *int64
-	isTiflashTarget := req.StoreTp == tikvrpc.TiFlash
-	if isTiflashTarget {
-		total = &details.UnpackedBytesReceivedMPPTotal
-		crossZone = &details.UnpackedBytesReceivedMPPCrossZone
-	} else {
-		total = &details.UnpackedBytesReceivedKVTotal
-		crossZone = &details.UnpackedBytesReceivedKVCrossZone
-	}
 
-	atomic.AddInt64(total, int64(size))
 	isCrossZoneTraffic := req.AccessLocation == kv.AccessCrossZone
-	if isCrossZoneTraffic {
-		atomic.AddInt64(crossZone, int64(size))
+
+	// exec details
+	if details != nil {
+		var total, crossZone *int64
+		isTiflashTarget := req.StoreTp == tikvrpc.TiFlash
+		if isTiflashTarget {
+			total = &details.UnpackedBytesReceivedMPPTotal
+			crossZone = &details.UnpackedBytesReceivedMPPCrossZone
+		} else {
+			total = &details.UnpackedBytesReceivedKVTotal
+			crossZone = &details.UnpackedBytesReceivedKVCrossZone
+		}
+		atomic.AddInt64(total, int64(size))
+		if isCrossZoneTraffic {
+			atomic.AddInt64(crossZone, int64(size))
+		}
 	}
+
 	// stale read metrics
-	if req.StaleRead {
-		s.staleReadMetricsCollector.onResp(float64(size), isCrossZoneTraffic)
+	if s.staleRead {
+		s.onRespStaleRead(float64(size), isCrossZoneTraffic)
+	}
+}
+
+func (s *networkCollector) onReqStaleRead(size float64, isCrossZoneTraffic bool) {
+	if isCrossZoneTraffic {
+		metrics.StaleReadRemoteOutBytes.Add(size)
+		metrics.StaleReadReqCrossZoneCounter.Add(1)
+	} else {
+		metrics.StaleReadLocalOutBytes.Add(size)
+		metrics.StaleReadReqLocalCounter.Add(1)
+	}
+}
+
+func (s *networkCollector) onRespStaleRead(size float64, isCrossZoneTraffic bool) {
+	if isCrossZoneTraffic {
+		metrics.StaleReadRemoteInBytes.Add(size)
+	} else {
+		metrics.StaleReadLocalInBytes.Add(size)
 	}
 }
diff --git a/internal/locate/region_request.go b/internal/locate/region_request.go
index fe36b2f..0bfdc71 100644
--- a/internal/locate/region_request.go
+++ b/internal/locate/region_request.go
@@ -499,6 +499,9 @@ func (s *RegionRequestSender) SendReqAsync(
 			et:       tikvrpc.TiKV,
 			opts:     opts,
 		},
+		invariants: reqInvariants{
+			staleRead: req.StaleRead,
+		},
 	}
 
 	cb.Inject(func(resp *tikvrpc.ResponseExt, err error) (*tikvrpc.ResponseExt, error) {
@@ -522,7 +525,7 @@ func (s *RegionRequestSender) SendReqAsync(
 		if retryTimes > 0 {
 			metrics.TiKVRequestRetryTimesHistogram.Observe(float64(retryTimes))
 		}
-		if req.StaleRead {
+		if state.invariants.staleRead {
 			if state.vars.sendTimes == 1 {
 				metrics.StaleReadHitCounter.Add(1)
 			} else {
@@ -573,7 +576,11 @@ func (s *RegionRequestSender) SendReqAsync(
 	cli.SendRequestAsync(ctx, sendToAddr, req, async.NewCallback(cb.Executor(), func(resp *tikvrpc.Response, err error) {
 		state.vars.sendTimes++
 		canceled := err != nil && hookCtx.Err() != nil && errors.Cause(hookCtx.Err()) == context.Canceled
-		if state.handleAsyncResponse(startTime, canceled, resp, err, cancels...) {
+		var execDetails *util.ExecDetails
+		if val := ctx.Value(util.ExecDetailsKey); val != nil {
+			execDetails = val.(*util.ExecDetails)
+		}
+		if state.handleAsyncResponse(startTime, canceled, resp, err, execDetails, cancels...) {
 			cb.Invoke(state.toResponseExt())
 			return
 		}
@@ -885,6 +892,15 @@ type sendReqState struct {
 		msg       string
 		sendTimes int
 	}
+
+	invariants reqInvariants
+}
+
+// reqInvariants holds the input state of the request.
+// If the tikvrpc.Request is changed during the retries or other operations.
+// the reqInvariants can tell the initial state.
+type reqInvariants struct {
+	staleRead bool
 }
 
 // next encapsulates one iteration of the retry loop. calling `next` will handle send error (s.vars.err) or region error
@@ -966,7 +982,6 @@ func (s *sendReqState) next() (done bool) {
 
 	if s.replicaSelector != nil &&
 		s.replicaSelector.target != nil &&
-		req.AccessLocation == kv.AccessUnknown &&
 		len(s.replicaSelector.option.labels) != 0 {
 		// patch the access location if it is not set under region request sender.
 		if s.replicaSelector.target.store.IsLabelsMatch(s.replicaSelector.option.labels) {
@@ -1114,6 +1129,18 @@ func (s *sendReqState) send() (canceled bool) {
 		if s.replicaSelector != nil {
 			recordAttemptedTime(s.replicaSelector, rpcDuration)
 		}
+
+		var execDetails *util.ExecDetails
+		if stmtExec := ctx.Value(util.ExecDetailsKey); stmtExec != nil {
+			execDetails := stmtExec.(*util.ExecDetails)
+			atomic.AddInt64(&execDetails.WaitKVRespDuration, int64(rpcDuration))
+		}
+		collector := networkCollector{
+			staleRead: s.invariants.staleRead,
+		}
+		collector.onReq(req, execDetails)
+		collector.onResp(req, s.vars.resp, execDetails)
+
 		// Record timecost of external requests on related Store when `ReplicaReadMode == "PreferLeader"`.
 		if rpcCtx.Store != nil && req.ReplicaReadType == kv.ReplicaReadPreferLeader && !util.IsInternalRequest(req.RequestSource) {
 			rpcCtx.Store.healthStatus.recordClientSideSlowScoreStat(rpcDuration)
@@ -1218,7 +1245,6 @@ func (s *sendReqState) initForAsyncRequest() (ok bool) {
 
 	if s.replicaSelector != nil &&
 		s.replicaSelector.target != nil &&
-		req.AccessLocation == kv.AccessUnknown &&
 		len(s.replicaSelector.option.labels) != 0 {
 		// patch the access location if it is not set under region request sender.
 		if s.replicaSelector.target.store.IsLabelsMatch(s.replicaSelector.option.labels) {
@@ -1251,7 +1277,7 @@ func (s *sendReqState) initForAsyncRequest() (ok bool) {
 }
 
 // handleAsyncResponse handles the response of an async request.
-func (s *sendReqState) handleAsyncResponse(start time.Time, canceled bool, resp *tikvrpc.Response, err error, cancels ...context.CancelFunc) (done bool) {
+func (s *sendReqState) handleAsyncResponse(start time.Time, canceled bool, resp *tikvrpc.Response, err error, execDetails *util.ExecDetails, cancels ...context.CancelFunc) (done bool) {
 	if len(cancels) > 0 {
 		defer func() {
 			for i := len(cancels) - 1; i >= 0; i-- {
@@ -1268,6 +1294,15 @@ func (s *sendReqState) handleAsyncResponse(start time.Time, canceled bool, resp
 	if s.Stats != nil {
 		s.Stats.RecordRPCRuntimeStats(req.Type, rpcDuration)
 	}
+	if execDetails != nil {
+		atomic.AddInt64(&execDetails.WaitKVRespDuration, int64(rpcDuration))
+	}
+	collector := networkCollector{
+		staleRead: s.invariants.staleRead,
+	}
+	collector.onReq(req, execDetails)
+	collector.onResp(req, resp, execDetails)
+
 	if s.vars.rpcCtx.Store != nil && req.ReplicaReadType == kv.ReplicaReadPreferLeader && !util.IsInternalRequest(req.RequestSource) {
 		s.vars.rpcCtx.Store.healthStatus.recordClientSideSlowScoreStat(rpcDuration)
 	}
@@ -1284,6 +1319,9 @@ func (s *sendReqState) handleAsyncResponse(start time.Time, canceled bool, resp
 	if err := s.vars.err; err != nil {
 		if isRPCError(err) {
 			s.rpcError = err
+			metrics.AsyncSendReqCounterWithRPCError.Inc()
+		} else {
+			metrics.AsyncSendReqCounterWithSendError.Inc()
 		}
 		if s.Stats != nil {
 			errStr := getErrMsg(err)
@@ -1292,17 +1330,18 @@ func (s *sendReqState) handleAsyncResponse(start time.Time, canceled bool, resp
 		}
 		if canceled {
 			metrics.TiKVRPCErrorCounter.WithLabelValues("context-canceled", storeIDLabel(s.vars.rpcCtx)).Inc()
-			return true
 		}
-		return false
+		return canceled
 	}
 
 	s.vars.regionErr, s.vars.err = s.vars.resp.GetRegionError()
 	if s.vars.err != nil {
 		s.vars.rpcCtx, s.vars.resp = nil, nil
+		metrics.AsyncSendReqCounterWithOtherError.Inc()
 		return true
 	} else if s.vars.regionErr != nil {
 		// need to handle region error
+		metrics.AsyncSendReqCounterWithRegionError.Inc()
 		return false
 	}
 
@@ -1310,6 +1349,7 @@ func (s *sendReqState) handleAsyncResponse(start time.Time, canceled bool, resp
 		s.replicaSelector.onSendSuccess(req)
 	}
 
+	metrics.AsyncSendReqCounterWithOK.Inc()
 	return true
 }
 
@@ -1394,13 +1434,16 @@ func (s *RegionRequestSender) SendReqCtx(
 			et:       et,
 			opts:     opts,
 		},
+		invariants: reqInvariants{
+			staleRead: req.StaleRead,
+		},
 	}
 
 	defer func() {
 		if retryTimes := state.vars.sendTimes - 1; retryTimes > 0 {
 			metrics.TiKVRequestRetryTimesHistogram.Observe(float64(retryTimes))
 		}
-		if req.StaleRead {
+		if state.invariants.staleRead {
 			if state.vars.sendTimes == 1 {
 				metrics.StaleReadHitCounter.Add(1)
 			} else {
diff --git a/metrics/metrics.go b/metrics/metrics.go
index 99da4b9..1d36c6d 100644
--- a/metrics/metrics.go
+++ b/metrics/metrics.go
@@ -118,6 +118,8 @@ var (
 	TiKVStaleRegionFromPDCounter                   prometheus.Counter
 	TiKVPipelinedFlushThrottleSecondsHistogram     prometheus.Histogram
 	TiKVTxnWriteConflictCounter                    prometheus.Counter
+	TiKVAsyncSendReqCounter                        *prometheus.CounterVec
+	TiKVAsyncBatchGetCounter                       *prometheus.CounterVec
 )
 
 // Label constants.
@@ -858,6 +860,24 @@ func initMetrics(namespace, subsystem string, constLabels prometheus.Labels) {
 			ConstLabels: constLabels,
 		})
 
+	TiKVAsyncSendReqCounter = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace:   namespace,
+			Subsystem:   subsystem,
+			Name:        "async_send_req_total",
+			Help:        "Counter of async send req by region request sender.",
+			ConstLabels: constLabels,
+		}, []string{LblResult})
+
+	TiKVAsyncBatchGetCounter = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace:   namespace,
+			Subsystem:   subsystem,
+			Name:        "async_batch_get_total",
+			Help:        "Counter of async batch get by txn snapshot.",
+			ConstLabels: constLabels,
+		}, []string{LblResult})
+
 	initShortcuts()
 }
 
@@ -954,6 +974,8 @@ func RegisterMetrics() {
 	prometheus.MustRegister(TiKVStaleRegionFromPDCounter)
 	prometheus.MustRegister(TiKVPipelinedFlushThrottleSecondsHistogram)
 	prometheus.MustRegister(TiKVTxnWriteConflictCounter)
+	prometheus.MustRegister(TiKVAsyncSendReqCounter)
+	prometheus.MustRegister(TiKVAsyncBatchGetCounter)
 }
 
 // readCounter reads the value of a prometheus.Counter.
diff --git a/metrics/shortcuts.go b/metrics/shortcuts.go
index 73c8107..90d7f25 100644
--- a/metrics/shortcuts.go
+++ b/metrics/shortcuts.go
@@ -176,6 +176,17 @@ var (
 	BatchRequestDurationSend prometheus.Observer
 	BatchRequestDurationRecv prometheus.Observer
 	BatchRequestDurationDone prometheus.Observer
+
+	AsyncSendReqCounterWithOK          prometheus.Counter
+	AsyncSendReqCounterWithRegionError prometheus.Counter
+	AsyncSendReqCounterWithRPCError    prometheus.Counter
+	AsyncSendReqCounterWithSendError   prometheus.Counter
+	AsyncSendReqCounterWithOtherError  prometheus.Counter
+
+	AsyncBatchGetCounterWithOK          prometheus.Counter
+	AsyncBatchGetCounterWithRegionError prometheus.Counter
+	AsyncBatchGetCounterWithLockError   prometheus.Counter
+	AsyncBatchGetCounterWithOtherError  prometheus.Counter
 )
 
 func initShortcuts() {
@@ -321,4 +332,15 @@ func initShortcuts() {
 	StaleReadLocalOutBytes = TiKVStaleReadBytes.WithLabelValues("local", "out")
 	StaleReadRemoteInBytes = TiKVStaleReadBytes.WithLabelValues("cross-zone", "in")
 	StaleReadRemoteOutBytes = TiKVStaleReadBytes.WithLabelValues("cross-zone", "out")
+
+	AsyncSendReqCounterWithOK = TiKVAsyncSendReqCounter.WithLabelValues("ok")
+	AsyncSendReqCounterWithRegionError = TiKVAsyncSendReqCounter.WithLabelValues("region_error")
+	AsyncSendReqCounterWithRPCError = TiKVAsyncSendReqCounter.WithLabelValues("rpc_error")
+	AsyncSendReqCounterWithSendError = TiKVAsyncSendReqCounter.WithLabelValues("send_error")
+	AsyncSendReqCounterWithOtherError = TiKVAsyncSendReqCounter.WithLabelValues("other_error")
+
+	AsyncBatchGetCounterWithOK = TiKVAsyncBatchGetCounter.WithLabelValues("ok")
+	AsyncBatchGetCounterWithRegionError = TiKVAsyncBatchGetCounter.WithLabelValues("region_error")
+	AsyncBatchGetCounterWithLockError = TiKVAsyncBatchGetCounter.WithLabelValues("lock_error")
+	AsyncBatchGetCounterWithOtherError = TiKVAsyncBatchGetCounter.WithLabelValues("other_error")
 }
diff --git a/txnkv/txnsnapshot/client_helper.go b/txnkv/txnsnapshot/client_helper.go
index c3008d1..8e30326 100644
--- a/txnkv/txnsnapshot/client_helper.go
+++ b/txnkv/txnsnapshot/client_helper.go
@@ -44,6 +44,7 @@ import (
 	"github.com/tikv/client-go/v2/tikvrpc"
 	"github.com/tikv/client-go/v2/txnkv/txnlock"
 	"github.com/tikv/client-go/v2/util"
+	"github.com/tikv/client-go/v2/util/async"
 )
 
 // ClientHelper wraps LockResolver and RegionRequestSender.
@@ -149,3 +150,19 @@ func (ch *ClientHelper) SendReqCtx(bo *retry.Backoffer, req *tikvrpc.Request, re
 	resp, ctx, _, err := sender.SendReqCtx(bo, req, regionID, timeout, et, opts...)
 	return resp, ctx, sender.GetStoreAddr(), err
 }
+
+// SendReqAsync wraps the SendReqAsync function and use the resolved lock result in the kvrpcpb.Context.
+func (ch *ClientHelper) SendReqAsync(
+	bo *retry.Backoffer,
+	req *tikvrpc.Request,
+	regionID locate.RegionVerID,
+	timeout time.Duration,
+	cb async.Callback[*tikvrpc.ResponseExt],
+	opts ...locate.StoreSelectorOption,
+) {
+	sender := locate.NewRegionRequestSender(ch.regionCache, ch.client, ch.oracle)
+	sender.Stats = ch.Stats
+	req.Context.ResolvedLocks = ch.resolvedLocks.GetAll()
+	req.Context.CommittedLocks = ch.committedLocks.GetAll()
+	sender.SendReqAsync(bo, req, regionID, timeout, cb, opts...)
+}
diff --git a/txnkv/txnsnapshot/snapshot.go b/txnkv/txnsnapshot/snapshot.go
index 4b34432..16f36ec 100644
--- a/txnkv/txnsnapshot/snapshot.go
+++ b/txnkv/txnsnapshot/snapshot.go
@@ -45,9 +45,11 @@ import (
 	"time"
 
 	"github.com/opentracing/opentracing-go"
+	"github.com/pingcap/kvproto/pkg/errorpb"
 	"github.com/pingcap/kvproto/pkg/kvrpcpb"
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pkg/errors"
+	"github.com/tikv/client-go/v2/config"
 	"github.com/tikv/client-go/v2/config/retry"
 	tikverr "github.com/tikv/client-go/v2/error"
 	"github.com/tikv/client-go/v2/internal/client"
@@ -101,6 +103,8 @@ type kvstore interface {
 	SendReq(bo *retry.Backoffer, req *tikvrpc.Request, regionID locate.RegionVerID, timeout time.Duration) (*tikvrpc.Response, error)
 	// GetOracle gets a timestamp oracle client.
 	GetOracle() oracle.Oracle
+	// Go schedules a function to be run in the goroutine pool.
+	Go(func()) error
 }
 
 // ReplicaReadAdjuster is a function that adjust the StoreSelectorOption and ReplicaReadType
@@ -256,7 +260,7 @@ func (s *KVSnapshot) BatchGetWithTier(ctx context.Context, keys [][]byte, readTi
 	s.mu.RUnlock()
 	// Create a map to collect key-values from region servers.
 	var mu sync.Mutex
-	err := s.batchGetKeysByRegions(bo, keys, readTier, func(k, v []byte) {
+	err := s.batchGetKeysByRegions(bo, keys, readTier, config.GetGlobalConfig().EnableAsyncBatchGet, func(k, v []byte) {
 		// when read buffer tier, empty value means a delete record, should also collect it.
 		if len(v) == 0 && readTier != BatchGetBufferTier {
 			return
@@ -323,6 +327,75 @@ func appendBatchKeysBySize(b []batchKeys, region locate.RegionVerID, keys [][]by
 	return b
 }
 
+type batchGetLockInfo struct {
+	lockedKeys [][]byte
+	locks      []*txnlock.Lock
+	keyErr     *kvrpcpb.KeyError
+}
+
+func collectBatchGetResponseData(
+	resp *tikvrpc.Response,
+	onKvPair func([]byte, []byte),
+	onDetails func(*kvrpcpb.ExecDetailsV2),
+) (*batchGetLockInfo, error) {
+	if resp.Resp == nil {
+		return nil, errors.WithStack(tikverr.ErrBodyMissing)
+	}
+	var (
+		data    = &batchGetLockInfo{}
+		pairs   []*kvrpcpb.KvPair
+		details *kvrpcpb.ExecDetailsV2
+	)
+	switch v := resp.Resp.(type) {
+	case *kvrpcpb.BatchGetResponse:
+		data.keyErr = v.GetError()
+		pairs = v.Pairs
+		details = v.GetExecDetailsV2()
+	case *kvrpcpb.BufferBatchGetResponse:
+		data.keyErr = v.GetError()
+		pairs = v.Pairs
+		details = v.GetExecDetailsV2()
+	default:
+		return nil, errors.Errorf("unknown response %T", v)
+	}
+	if data.keyErr != nil {
+		// If a response-level error happens, skip reading pairs.
+		lock, err := txnlock.ExtractLockFromKeyErr(data.keyErr)
+		if err != nil {
+			return nil, err
+		}
+		data.lockedKeys = append(data.lockedKeys, lock.Key)
+		data.locks = append(data.locks, lock)
+	} else {
+		for _, pair := range pairs {
+			keyErr := pair.GetError()
+			if keyErr == nil {
+				onKvPair(pair.GetKey(), pair.GetValue())
+				continue
+			}
+			lock, err := txnlock.ExtractLockFromKeyErr(keyErr)
+			if err != nil {
+				return nil, err
+			}
+			data.lockedKeys = append(data.lockedKeys, lock.Key)
+			data.locks = append(data.locks, lock)
+		}
+	}
+	if details != nil {
+		readKeys := len(pairs)
+		var readTime float64
+		if timeDetail := details.GetTimeDetailV2(); timeDetail != nil {
+			readTime = float64(timeDetail.GetKvReadWallTimeNs()) / 1000000000.
+		} else if timeDetail := details.GetTimeDetail(); timeDetail != nil {
+			readTime = float64(timeDetail.GetKvReadWallTimeMs()) / 1000.
+		}
+		readSize := float64(details.GetScanDetailV2().GetProcessedVersionsSize())
+		metrics.ObserveReadSLI(uint64(readKeys), readTime, readSize)
+		onDetails(details)
+	}
+	return data, nil
+}
+
 //go:noinline
 func growStackForBatchGetWorker() {
 	// A batch get worker typically needs 16KB stack space. So we pre-allocate 8KB here to let the stack grow to 16KB
@@ -331,7 +404,7 @@ func growStackForBatchGetWorker() {
 	runtime.KeepAlive(ballast[:])
 }
 
-func (s *KVSnapshot) batchGetKeysByRegions(bo *retry.Backoffer, keys [][]byte, readTier int, collectF func(k, v []byte)) error {
+func (s *KVSnapshot) batchGetKeysByRegions(bo *retry.Backoffer, keys [][]byte, readTier int, tryAsyncAPI bool, collectF func(k, v []byte)) error {
 	defer func(start time.Time) {
 		if s.IsInternal() {
 			metrics.TxnCmdHistogramWithBatchGetInternal.Observe(time.Since(start).Seconds())
@@ -361,6 +434,9 @@ func (s *KVSnapshot) batchGetKeysByRegions(bo *retry.Backoffer, keys [][]byte, r
 	if len(batches) == 1 {
 		return s.batchGetSingleRegion(bo, batches[0], readTier, collectF)
 	}
+	if tryAsyncAPI {
+		return s.asyncBatchGetByRegions(bo, batches, readTier, collectF)
+	}
 	ch := make(chan error, len(batches))
 	var lastForkedBo atomic.Pointer[retry.Backoffer]
 	forkedBo, cancel := bo.Fork()
@@ -425,6 +501,33 @@ func (s *KVSnapshot) buildBatchGetRequest(keys [][]byte, busyThresholdMs int64,
 	}
 }
 
+func (s *KVSnapshot) handleBatchGetRegionError(bo *retry.Backoffer, batch *batchKeys, regionCache *locate.RegionCache, regionError *errorpb.Error) (retriable bool, err error) {
+	if err = retry.MayBackoffForRegionError(regionError, bo); err != nil {
+		return false, err
+	}
+	return batch.relocate(bo, regionCache)
+}
+
+func (s *KVSnapshot) handleBatchGetLocks(bo *retry.Backoffer, lockInfo *batchGetLockInfo, cli *ClientHelper) error {
+	resolveLocksOpts := txnlock.ResolveLocksOptions{
+		CallerStartTS: s.version,
+		Locks:         lockInfo.locks,
+		Detail:        s.GetResolveLockDetail(),
+	}
+	resolveLocksRes, err := cli.ResolveLocksWithOpts(bo, resolveLocksOpts)
+	msBeforeExpired := resolveLocksRes.TTL
+	if err != nil {
+		return err
+	}
+	if msBeforeExpired > 0 {
+		err = bo.BackoffWithMaxSleepTxnLockFast(int(msBeforeExpired), errors.Errorf("BatchGetWithTier lockedKeys: %d", len(lockInfo.lockedKeys)))
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (s *KVSnapshot) batchGetSingleRegion(bo *retry.Backoffer, batch batchKeys, readTier int, collectF func(k, v []byte)) error {
 	cli := NewClientHelper(s.store, &s.resolvedLocks, &s.committedLocks, false)
 	s.mu.RLock()
@@ -494,109 +597,40 @@ func (s *KVSnapshot) batchGetSingleRegion(bo *retry.Backoffer, batch batchKeys,
 		}
 		readType = req.ReadType
 		if regionErr != nil {
-			if err = retry.MayBackoffForRegionError(regionErr, bo); err != nil {
-				return err
-			}
-			same, err := batch.relocate(bo, cli.regionCache)
+			retriable, err := s.handleBatchGetRegionError(bo, &batch, cli.regionCache, regionErr)
 			if err != nil {
 				return err
 			}
-			if same {
-				continue
-			}
-			return s.batchGetKeysByRegions(bo, pending, readTier, collectF)
-		}
-		if resp.Resp == nil {
-			return errors.WithStack(tikverr.ErrBodyMissing)
-		}
-		var (
-			lockedKeys [][]byte
-			locks      []*txnlock.Lock
-
-			keyErr  *kvrpcpb.KeyError
-			pairs   []*kvrpcpb.KvPair
-			details *kvrpcpb.ExecDetailsV2
-		)
-		switch v := resp.Resp.(type) {
-		case *kvrpcpb.BatchGetResponse:
-			keyErr = v.GetError()
-			pairs = v.Pairs
-			details = v.GetExecDetailsV2()
-		case *kvrpcpb.BufferBatchGetResponse:
-			keyErr = v.GetError()
-			pairs = v.Pairs
-			details = v.GetExecDetailsV2()
-		default:
-			return errors.Errorf("unknown response %T", v)
-		}
-		if keyErr != nil {
-			// If a response-level error happens, skip reading pairs.
-			lock, err := txnlock.ExtractLockFromKeyErr(keyErr)
-			if err != nil {
-				return err
-			}
-			lockedKeys = append(lockedKeys, lock.Key)
-			locks = append(locks, lock)
-		} else {
-			for _, pair := range pairs {
-				keyErr := pair.GetError()
-				if keyErr == nil {
-					collectF(pair.GetKey(), pair.GetValue())
-					continue
-				}
-				lock, err := txnlock.ExtractLockFromKeyErr(keyErr)
-				if err != nil {
-					return err
-				}
-				lockedKeys = append(lockedKeys, lock.Key)
-				locks = append(locks, lock)
+			if !retriable {
+				return s.batchGetKeysByRegions(bo, pending, readTier, false, collectF)
 			}
+			continue
 		}
-		if details != nil {
-			readKeys := len(pairs)
-			var readTime float64
-			if timeDetail := details.GetTimeDetailV2(); timeDetail != nil {
-				readTime = float64(timeDetail.GetKvReadWallTimeNs()) / 1000000000.
-			} else if timeDetail := details.GetTimeDetail(); timeDetail != nil {
-				readTime = float64(timeDetail.GetKvReadWallTimeMs()) / 1000.
-			}
-			readSize := float64(details.GetScanDetailV2().GetProcessedVersionsSize())
-			metrics.ObserveReadSLI(uint64(readKeys), readTime, readSize)
-			s.mergeExecDetail(details)
+
+		lockInfo, err := collectBatchGetResponseData(resp, collectF, s.mergeExecDetail)
+		if err != nil {
+			return err
 		}
-		if len(lockedKeys) > 0 {
+		if len(lockInfo.lockedKeys) > 0 {
 			if resolvingRecordToken == nil {
-				token := cli.RecordResolvingLocks(locks, s.version)
+				token := cli.RecordResolvingLocks(lockInfo.locks, s.version)
 				resolvingRecordToken = &token
 				defer cli.ResolveLocksDone(s.version, *resolvingRecordToken)
 			} else {
-				cli.UpdateResolvingLocks(locks, s.version, *resolvingRecordToken)
+				cli.UpdateResolvingLocks(lockInfo.locks, s.version, *resolvingRecordToken)
 			}
 			// we need to read from leader after resolving the lock.
 			if isStaleness {
 				isStaleness = false
 				busyThresholdMs = 0
 			}
-			resolveLocksOpts := txnlock.ResolveLocksOptions{
-				CallerStartTS: s.version,
-				Locks:         locks,
-				Detail:        s.GetResolveLockDetail(),
-			}
-			resolveLocksRes, err := cli.ResolveLocksWithOpts(bo, resolveLocksOpts)
-			msBeforeExpired := resolveLocksRes.TTL
-			if err != nil {
+			if err := s.handleBatchGetLocks(bo, lockInfo, cli); err != nil {
 				return err
 			}
-			if msBeforeExpired > 0 {
-				err = bo.BackoffWithMaxSleepTxnLockFast(int(msBeforeExpired), errors.Errorf("BatchGetWithTier lockedKeys: %d", len(lockedKeys)))
-				if err != nil {
-					return err
-				}
-			}
 			// Only reduce pending keys when there is no response-level error. Otherwise,
 			// lockedKeys may be incomplete.
-			if keyErr == nil {
-				pending = lockedKeys
+			if lockInfo.keyErr == nil {
+				pending = lockInfo.lockedKeys
 			}
 			continue
 		}
diff --git a/txnkv/txnsnapshot/snapshot_async.go b/txnkv/txnsnapshot/snapshot_async.go
new file mode 100644
index 0000000..dbe441c
--- /dev/null
+++ b/txnkv/txnsnapshot/snapshot_async.go
@@ -0,0 +1,297 @@
+// Copyright 2025 TiKV Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package txnsnapshot
+
+import (
+	"github.com/pingcap/kvproto/pkg/errorpb"
+	"github.com/pkg/errors"
+	"github.com/tikv/client-go/v2/config/retry"
+	"github.com/tikv/client-go/v2/internal/client"
+	"github.com/tikv/client-go/v2/internal/locate"
+	"github.com/tikv/client-go/v2/internal/logutil"
+	"github.com/tikv/client-go/v2/metrics"
+	"github.com/tikv/client-go/v2/tikvrpc"
+	"github.com/tikv/client-go/v2/util/async"
+	"go.uber.org/zap"
+)
+
+type poolWrapper struct {
+	pool interface{ Go(func()) error }
+}
+
+func (p *poolWrapper) Go(f func()) {
+	err := p.pool.Go(f)
+	if err != nil {
+		// fallback to native go
+		go f()
+	}
+}
+
+func (s *KVSnapshot) asyncBatchGetByRegions(
+	bo *retry.Backoffer,
+	batches []batchKeys,
+	readTier int,
+	collectF func(k, v []byte),
+) (err error) {
+	var (
+		runloop      = async.NewRunLoop()
+		completed    = 0
+		lastForkedBo *retry.Backoffer
+	)
+	runloop.Pool = &poolWrapper{pool: s.store}
+	forkedBo, cancel := bo.Fork()
+	defer cancel()
+	for i, batch1 := range batches {
+		var backoffer *retry.Backoffer
+		if i == 0 {
+			backoffer = forkedBo
+		} else {
+			backoffer = forkedBo.Clone()
+		}
+		batch := batch1
+		s.tryBatchGetSingleRegionUsingAsyncAPI(backoffer, batch, readTier, collectF, async.NewCallback(runloop, func(_ struct{}, e error) {
+			// The callback is designed to be executed in the runloop's goroutine thus it should be safe to update the
+			// following variables without locks.
+			completed++
+			lastForkedBo = backoffer
+			if e != nil {
+				logutil.BgLogger().Debug("snapshot BatchGetWithTier failed",
+					zap.Error(e),
+					zap.Uint64("txnStartTS", s.version))
+				err = errors.WithStack(e)
+			}
+		}))
+	}
+	for completed < len(batches) {
+		_, err = runloop.Exec(bo.GetCtx())
+		if err != nil {
+			break
+		}
+	}
+	if lastForkedBo != nil {
+		bo.UpdateUsingForked(lastForkedBo)
+	}
+	return err
+}
+
+func (s *KVSnapshot) tryBatchGetSingleRegionUsingAsyncAPI(
+	bo *retry.Backoffer,
+	batch batchKeys,
+	readTier int,
+	collectF func(k, v []byte),
+	cb async.Callback[struct{}],
+) {
+	cli := NewClientHelper(s.store, &s.resolvedLocks, &s.committedLocks, false)
+	s.mu.RLock()
+	if s.mu.stats != nil {
+		cli.Stats = locate.NewRegionRequestRuntimeStats()
+		cb.Inject(func(res struct{}, err error) (struct{}, error) {
+			s.mergeRegionRequestStats(cli.Stats)
+			return res, err
+		})
+	}
+	req, err := s.buildBatchGetRequest(batch.keys, s.mu.busyThreshold.Milliseconds(), readTier)
+	if err != nil {
+		s.mu.RUnlock()
+		cb.Invoke(struct{}{}, err)
+		return
+	}
+	req.InputRequestSource = s.GetRequestSource()
+	if s.mu.resourceGroupTag == nil && s.mu.resourceGroupTagger != nil {
+		s.mu.resourceGroupTagger(req)
+	}
+	isStaleness := s.mu.isStaleness
+	readReplicaScope := s.mu.readReplicaScope
+	matchStoreLabels := s.mu.matchStoreLabels
+	replicaReadAdjuster := s.mu.replicaReadAdjuster
+	s.mu.RUnlock()
+
+	req.TxnScope = readReplicaScope
+	req.ReadReplicaScope = readReplicaScope
+	if isStaleness {
+		req.EnableStaleWithMixedReplicaRead()
+	}
+
+	timeout := client.ReadTimeoutMedium
+	if s.readTimeout > 0 {
+		timeout = s.readTimeout
+	}
+	req.MaxExecutionDurationMs = uint64(timeout.Milliseconds())
+
+	ops := make([]locate.StoreSelectorOption, 0, 2)
+	if len(matchStoreLabels) > 0 {
+		ops = append(ops, locate.WithMatchLabels(matchStoreLabels))
+	}
+	if req.ReplicaReadType.IsFollowerRead() && replicaReadAdjuster != nil {
+		op, readType := replicaReadAdjuster(len(batch.keys))
+		if op != nil {
+			ops = append(ops, op)
+		}
+		req.ReplicaReadType = readType
+	}
+
+	onResp := func(resp *tikvrpc.ResponseExt, err error) {
+		if err != nil {
+			cb.Invoke(struct{}{}, err)
+			metrics.AsyncBatchGetCounterWithOtherError.Inc()
+			return
+		}
+
+		regionErr, err := resp.GetRegionError()
+		if err != nil {
+			cb.Invoke(struct{}{}, err)
+			metrics.AsyncBatchGetCounterWithOtherError.Inc()
+			return
+		}
+		if regionErr != nil {
+			cb.Executor().Go(func() {
+				growStackForBatchGetWorker()
+				err := s.retryBatchGetSingleRegionAfterAsyncAPI(bo, cli, batch, readTier, req.ReadType, regionErr, nil, collectF)
+				cb.Schedule(struct{}{}, err)
+			})
+			metrics.AsyncBatchGetCounterWithRegionError.Inc()
+			return
+		}
+
+		lockInfo, err := collectBatchGetResponseData(&resp.Response, collectF, s.mergeExecDetail)
+		if err != nil {
+			cb.Invoke(struct{}{}, err)
+			metrics.AsyncBatchGetCounterWithOtherError.Inc()
+			return
+		}
+		if len(lockInfo.lockedKeys) > 0 {
+			cb.Executor().Go(func() {
+				growStackForBatchGetWorker()
+				err := s.retryBatchGetSingleRegionAfterAsyncAPI(bo, cli, batch, readTier, req.ReadType, nil, lockInfo, collectF)
+				cb.Schedule(struct{}{}, err)
+			})
+			metrics.AsyncBatchGetCounterWithLockError.Inc()
+			return
+		}
+
+		cb.Invoke(struct{}{}, nil)
+		metrics.AsyncBatchGetCounterWithOK.Inc()
+	}
+
+	cli.SendReqAsync(bo, req, batch.region, timeout, async.NewCallback(cb.Executor(), onResp), ops...)
+}
+
+func (s *KVSnapshot) retryBatchGetSingleRegionAfterAsyncAPI(
+	bo *retry.Backoffer,
+	cli *ClientHelper,
+	batch batchKeys,
+	readTier int,
+	readType string,
+	regionErr *errorpb.Error,
+	lockInfo *batchGetLockInfo,
+	collectF func(k, v []byte),
+) error {
+	var (
+		resolvingRecordToken  *int
+		readAfterResolveLocks bool
+	)
+	for {
+		if regionErr != nil {
+			retriable, err := s.handleBatchGetRegionError(bo, &batch, cli.regionCache, regionErr)
+			if err != nil {
+				return err
+			}
+			if !retriable {
+				return s.batchGetKeysByRegions(bo, batch.keys, readTier, false, collectF)
+			}
+		} else if lockInfo != nil && len(lockInfo.lockedKeys) > 0 {
+			if resolvingRecordToken == nil {
+				token := cli.RecordResolvingLocks(lockInfo.locks, s.version)
+				resolvingRecordToken = &token
+				defer cli.ResolveLocksDone(s.version, *resolvingRecordToken)
+			} else {
+				cli.UpdateResolvingLocks(lockInfo.locks, s.version, *resolvingRecordToken)
+			}
+			readAfterResolveLocks = true
+			if err := s.handleBatchGetLocks(bo, lockInfo, cli); err != nil {
+				return err
+			}
+			// Only reduce pending keys when there is no response-level error. Otherwise,
+			// lockedKeys may be incomplete.
+			if lockInfo.keyErr == nil {
+				batch.keys = lockInfo.lockedKeys
+			}
+		}
+
+		s.mu.RLock()
+		isStaleness := s.mu.isStaleness
+		busyThresholdMs := s.mu.busyThreshold.Milliseconds()
+		if isStaleness && readAfterResolveLocks {
+			isStaleness = false
+			busyThresholdMs = 0
+		}
+		req, err := s.buildBatchGetRequest(batch.keys, busyThresholdMs, readTier)
+		if err != nil {
+			s.mu.RUnlock()
+			return err
+		}
+		req.InputRequestSource = s.GetRequestSource()
+		if readType != "" {
+			req.ReadType = readType
+			req.IsRetryRequest = true
+		}
+		if s.mu.resourceGroupTag == nil && s.mu.resourceGroupTagger != nil {
+			s.mu.resourceGroupTagger(req)
+		}
+		readReplicaScope := s.mu.readReplicaScope
+		matchStoreLabels := s.mu.matchStoreLabels
+		replicaReadAdjuster := s.mu.replicaReadAdjuster
+		s.mu.RUnlock()
+
+		req.TxnScope = readReplicaScope
+		req.ReadReplicaScope = readReplicaScope
+		if isStaleness {
+			req.EnableStaleWithMixedReplicaRead()
+		}
+		timeout := client.ReadTimeoutMedium
+		req.MaxExecutionDurationMs = uint64(timeout.Milliseconds())
+		ops := make([]locate.StoreSelectorOption, 0, 2)
+		if len(matchStoreLabels) > 0 {
+			ops = append(ops, locate.WithMatchLabels(matchStoreLabels))
+		}
+		if req.ReplicaReadType.IsFollowerRead() && replicaReadAdjuster != nil {
+			op, readType := replicaReadAdjuster(len(batch.keys))
+			if op != nil {
+				ops = append(ops, op)
+			}
+			req.ReplicaReadType = readType
+		}
+		resp, _, _, err := cli.SendReqCtx(bo, req, batch.region, timeout, tikvrpc.TiKV, "", ops...)
+		if err != nil {
+			return err
+		}
+		regionErr, err = resp.GetRegionError()
+		if err != nil {
+			return err
+		}
+		readType = req.ReadType
+		if regionErr != nil {
+			continue
+		}
+		lockInfo, err = collectBatchGetResponseData(resp, collectF, s.mergeExecDetail)
+		if err != nil {
+			return err
+		}
+		if len(lockInfo.lockedKeys) > 0 {
+			continue
+		}
+		return nil
+	}
+}


Test patch:
diff --git a/integration_tests/split_test.go b/integration_tests/split_test.go
index 5f86079..56957bf 100644
--- a/integration_tests/split_test.go
+++ b/integration_tests/split_test.go
@@ -46,6 +46,7 @@ import (
 	rmpb "github.com/pingcap/kvproto/pkg/resource_manager"
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/suite"
+	"github.com/tikv/client-go/v2/config"
 	"github.com/tikv/client-go/v2/testutils"
 	"github.com/tikv/client-go/v2/tikv"
 	"github.com/tikv/client-go/v2/txnkv/transaction"
@@ -114,6 +115,65 @@ func (s *testSplitSuite) TestSplitBatchGet() {
 	s.Nil(err)
 }
 
+func (s *testSplitSuite) TestBatchGetUsingAsyncAPI() {
+	defer config.UpdateGlobal(func(conf *config.Config) {
+		conf.EnableAsyncBatchGet = true
+	})()
+
+	// init data
+	txn := s.begin()
+	s.Nil(txn.Set([]byte("a"), []byte("a")))
+	s.Nil(txn.Set([]byte("c"), []byte("c")))
+	s.Nil(txn.Commit(context.TODO()))
+
+	// split by 'b' and invalidate the region so that BatchGet('a', 'b', 'c') has to handle 2 batches.
+	loc, err := s.store.GetRegionCache().LocateKey(s.bo, []byte("b"))
+	s.Nil(err)
+	s.True(loc.Contains([]byte("a")))
+	s.True(loc.Contains([]byte("c")))
+	s.split(loc.Region.GetID(), []byte("b"))
+	s.store.GetRegionCache().InvalidateCachedRegion(loc.Region)
+
+	txn = s.begin()
+	m, err := txn.GetSnapshot().BatchGet(context.TODO(), [][]byte{{'a'}, {'b'}, {'c'}})
+	s.Nil(err)
+	s.Len(m, 2)
+	s.Equal([]byte("a"), m[string([]byte{'a'})])
+	s.Equal([]byte("c"), m[string([]byte{'c'})])
+	s.NotContains(m, string([]byte{'b'}))
+
+	// split by 'c' without invalidating the old region, then BatchGet('a', 'b', 'c') will meet region error when handling batch{keys: ['b', 'c']}.
+	loc, err = s.store.GetRegionCache().LocateKey(s.bo, []byte("b"))
+	s.Nil(err)
+	s.False(loc.Contains([]byte("a")))
+	s.True(loc.Contains([]byte("b")))
+	s.True(loc.Contains([]byte("c")))
+	s.split(loc.Region.GetID(), []byte("c"))
+
+	txn = s.begin()
+	m, err = txn.GetSnapshot().BatchGet(context.TODO(), [][]byte{{'a'}, {'b'}, {'c'}})
+	s.Nil(err)
+	s.Len(m, 2)
+	s.Equal([]byte("a"), m[string([]byte{'a'})])
+	s.Equal([]byte("c"), m[string([]byte{'c'})])
+	s.NotContains(m, string([]byte{'b'}))
+
+	// leave a lock on 'a' so that BatchGet('a', 'b', 'c') will meet lock error.
+	txnW := s.begin()
+	s.Nil(txnW.Set([]byte("a"), []byte("a")))
+	commiter, err := txnW.NewCommitter(1)
+	s.Nil(err)
+	s.Nil(commiter.PrewriteAllMutations(context.TODO()))
+
+	txn = s.begin()
+	m, err = txn.GetSnapshot().BatchGet(context.TODO(), [][]byte{{'a'}, {'b'}, {'c'}})
+	s.Nil(err)
+	s.Len(m, 2)
+	s.Equal([]byte("a"), m[string([]byte{'a'})])
+	s.Equal([]byte("c"), m[string([]byte{'c'})])
+	s.NotContains(m, string([]byte{'b'}))
+}
+
 func (s *testSplitSuite) TestStaleEpoch() {
 	mockPDClient := &mockPDClient{client: s.store.GetRegionCache().PDClient()}
 	s.store.SetRegionCachePDClient(mockPDClient)
diff --git a/internal/client/metrics_collector_test.go b/internal/locate/metrics_collector_test.go
similarity index 97%
rename from internal/client/metrics_collector_test.go
rename to internal/locate/metrics_collector_test.go
index b1f32c7..75cdde2 100644
--- a/internal/client/metrics_collector_test.go
+++ b/internal/locate/metrics_collector_test.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package client
+package locate
 
 import (
 	"sync/atomic"
@@ -85,9 +85,6 @@ func TestNetworkCollectorOnReq(t *testing.T) {
 }
 
 func TestNetworkCollectorOnResp(t *testing.T) {
-	// Initialize the collector and dependencies
-	collector := &networkCollector{}
-
 	// Construct requests and responses
 	reqs := []*tikvrpc.Request{
 		tikvrpc.NewRequest(
@@ -140,6 +137,10 @@ func TestNetworkCollectorOnResp(t *testing.T) {
 	for _, cas := range testCases {
 		// Call the method
 		cas.req.AccessLocation = kv.AccessLocalZone
+		// Initialize the collector and dependencies
+		collector := &networkCollector{
+			staleRead: cas.req.StaleRead,
+		}
 		collector.onResp(cas.req, cas.resp, details)
 
 		// Verify metrics
diff --git a/internal/locate/region_request3_test.go b/internal/locate/region_request3_test.go
index a299c1d..03aa1f3 100644
--- a/internal/locate/region_request3_test.go
+++ b/internal/locate/region_request3_test.go
@@ -50,6 +50,8 @@ import (
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/log"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/suite"
 	"github.com/tikv/client-go/v2/config/retry"
 	tikverr "github.com/tikv/client-go/v2/error"
@@ -57,8 +59,11 @@ import (
 	"github.com/tikv/client-go/v2/internal/client"
 	"github.com/tikv/client-go/v2/internal/mockstore/mocktikv"
 	"github.com/tikv/client-go/v2/kv"
+	"github.com/tikv/client-go/v2/metrics"
 	"github.com/tikv/client-go/v2/oracle"
 	"github.com/tikv/client-go/v2/tikvrpc"
+	"github.com/tikv/client-go/v2/util"
+	"github.com/tikv/client-go/v2/util/async"
 	"go.uber.org/zap"
 )
 
@@ -1639,3 +1644,99 @@ func (s *testRegionRequestToThreeStoresSuite) TestTiKVRecoveredFromDown() {
 	}
 	s.Require().Fail("should access recovered peer after region reloading within RegionCacheTTL")
 }
+
+func (s *testRegionRequestToThreeStoresSuite) TestStaleReadMetrics() {
+	readMetric := func(col prometheus.Collector) int {
+		ch := make(chan prometheus.Metric, 1)
+		col.Collect(ch)
+		var m dto.Metric
+		s.Nil((<-ch).Write(&m))
+		return int(*m.Counter.Value + 0.000001) // round to int and avoid floating point precision issues
+	}
+
+	for _, staleReadHit := range []bool{false, true} {
+		for _, asyncReq := range []bool{false, true} {
+			caseName := fmt.Sprintf("async=%t, staleReadHit=%t", asyncReq, staleReadHit)
+			// Delete all vectors and recreate them before each test case.
+			metrics.TiKVStaleReadCounter.Reset()
+			metrics.TiKVStaleReadReqCounter.Reset()
+			metrics.TiKVStaleReadBytes.Reset()
+			metrics.StaleReadHitCounter = metrics.TiKVStaleReadCounter.WithLabelValues("hit")
+			metrics.StaleReadMissCounter = metrics.TiKVStaleReadCounter.WithLabelValues("miss")
+			metrics.StaleReadReqLocalCounter = metrics.TiKVStaleReadReqCounter.WithLabelValues("local")
+			metrics.StaleReadReqCrossZoneCounter = metrics.TiKVStaleReadReqCounter.WithLabelValues("cross-zone")
+			metrics.StaleReadLocalInBytes = metrics.TiKVStaleReadBytes.WithLabelValues("local", "in")
+			metrics.StaleReadLocalOutBytes = metrics.TiKVStaleReadBytes.WithLabelValues("local", "out")
+			metrics.StaleReadRemoteInBytes = metrics.TiKVStaleReadBytes.WithLabelValues("cross-zone", "in")
+			metrics.StaleReadRemoteOutBytes = metrics.TiKVStaleReadBytes.WithLabelValues("cross-zone", "out")
+
+			key := []byte("key")
+			value := []byte("value")
+
+			s.regionRequestSender.client = &fnClient{fn: func(ctx context.Context, addr string, req *tikvrpc.Request, timeout time.Duration) (resp *tikvrpc.Response, err error) {
+				if req.StaleRead && !staleReadHit {
+					return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{RegionError: &errorpb.Error{
+						DataIsNotReady: &errorpb.DataIsNotReady{},
+					}}}, nil
+				}
+				return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{Value: value}}, nil
+			}}
+
+			req := tikvrpc.NewReplicaReadRequest(tikvrpc.CmdGet, &kvrpcpb.GetRequest{Key: key}, kv.ReplicaReadLeader, nil)
+			req.ReadReplicaScope = oracle.GlobalTxnScope
+			req.EnableStaleWithMixedReplicaRead()
+
+			ctx := context.WithValue(context.Background(), util.ExecDetailsKey, &util.ExecDetails{})
+			bo := retry.NewBackoffer(ctx, -1)
+			loc, err := s.cache.LocateKey(bo, key)
+			s.Require().Nil(err)
+			var resp *tikvrpc.Response
+			if asyncReq {
+				complete := false
+				rl := async.NewRunLoop()
+				s.regionRequestSender.SendReqAsync(bo, req, loc.Region, time.Second, async.NewCallback(rl, func(innerResp *tikvrpc.ResponseExt, innerErr error) {
+					resp, err = &innerResp.Response, innerErr
+					complete = true
+				}), WithMatchLabels(s.cluster.GetStore(s.storeIDs[0]).Labels))
+				for !complete {
+					_, err := rl.Exec(ctx)
+					s.Nil(err)
+				}
+			} else {
+				resp, _, _, err = s.regionRequestSender.SendReqCtx(bo, req, loc.Region, time.Second, tikvrpc.TiKV, WithMatchLabels(s.cluster.GetStore(s.storeIDs[0]).Labels))
+			}
+			s.Require().Nil(err)
+			s.Equal(value, resp.Resp.(*kvrpcpb.GetResponse).Value)
+
+			hits, misses := readMetric(metrics.StaleReadHitCounter), readMetric(metrics.StaleReadMissCounter)
+			localReq, remoteReq := readMetric(metrics.StaleReadReqLocalCounter), readMetric(metrics.StaleReadReqCrossZoneCounter)
+			localInBytes, localOutBytes := readMetric(metrics.StaleReadLocalInBytes), readMetric(metrics.StaleReadLocalOutBytes)
+			remoteInBytes, remoteOutBytes := readMetric(metrics.StaleReadRemoteInBytes), readMetric(metrics.StaleReadRemoteOutBytes)
+			if staleReadHit {
+				// when stale read hitting
+				// local metrics should be counted
+				s.Equal(1, hits, caseName)
+				s.Equal(1, localReq, caseName)
+				s.Greater(localInBytes, 0, caseName)
+				s.Greater(localOutBytes, 0, caseName)
+				// remote metrics should be zero
+				s.Zero(misses, caseName)
+				s.Zero(remoteReq, caseName)
+				s.Zero(remoteInBytes, caseName)
+				s.Zero(remoteOutBytes, caseName)
+			} else {
+				// when stale read missing
+				s.Zero(hits, caseName)
+				// local replica is tried first, so local metrics will also be counted
+				s.Greater(localReq, 0, caseName)
+				s.Greater(localInBytes, 0, caseName)
+				s.Greater(localOutBytes, 0, caseName)
+				// remote metrics should be counted
+				s.Equal(1, misses, caseName)
+				s.Equal(1, remoteReq, caseName)
+				s.Greater(remoteInBytes, 0, caseName)
+				s.Greater(remoteInBytes, 0, caseName)
+			}
+		}
+	}
+}
diff --git a/internal/locate/replica_selector_test.go b/internal/locate/replica_selector_test.go
index a9213c1..35cc249 100644
--- a/internal/locate/replica_selector_test.go
+++ b/internal/locate/replica_selector_test.go
@@ -2864,18 +2864,36 @@ func (ca *replicaSelectorAccessPathCase) run(s *testReplicaSelectorSuite) *Regio
 				s.NotNil(rc)
 				regionErr, err := ca.genAccessErr(s.cache, rc, ca.accessErr[idx])
 				if regionErr != nil {
-					return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{
-						RegionError: regionErr,
-					}}, nil
+					switch req.Type {
+					case tikvrpc.CmdGet:
+						return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{
+							RegionError: regionErr,
+						}}, nil
+					case tikvrpc.CmdPrewrite:
+						return &tikvrpc.Response{Resp: &kvrpcpb.PrewriteResponse{
+							RegionError: regionErr,
+						}}, nil
+					default:
+						s.FailNow("unsupported reqType " + req.Type.String())
+						return nil, fmt.Errorf("unsupported reqType %v", req.Type)
+					}
 				}
 				if err != nil {
 					return nil, err
 				}
 			}
 		}
-		return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{
-			Value: []byte("hello world"),
-		}}, nil
+		switch req.Type {
+		case tikvrpc.CmdGet:
+			return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{
+				Value: []byte("hello world"),
+			}}, nil
+		case tikvrpc.CmdPrewrite:
+			return &tikvrpc.Response{Resp: &kvrpcpb.PrewriteResponse{}}, nil
+		default:
+			s.FailNow("unsupported reqType " + req.Type.String())
+			return nil, fmt.Errorf("unsupported reqType %v", req.Type)
+		}
 	}}
 	sender := NewRegionRequestSender(s.cache, fnClient, oracle.NoopReadTSValidator{})
 	req, opts, timeout := ca.buildRequest(s)
diff --git a/tikv/test_util.go b/tikv/test_util.go
index eb1c6d0..4430114 100644
--- a/tikv/test_util.go
+++ b/tikv/test_util.go
@@ -44,6 +44,7 @@ import (
 	"github.com/tikv/client-go/v2/internal/apicodec"
 	"github.com/tikv/client-go/v2/internal/locate"
 	"github.com/tikv/client-go/v2/tikvrpc"
+	"github.com/tikv/client-go/v2/util/async"
 	pd "github.com/tikv/pd/client"
 )
 
@@ -66,6 +67,28 @@ func (c *CodecClient) SendRequest(ctx context.Context, addr string, req *tikvrpc
 	return c.codec.DecodeResponse(req, resp)
 }
 
+func (c *CodecClient) SendRequestAsync(ctx context.Context, addr string, req *tikvrpc.Request, cb async.Callback[*tikvrpc.Response]) {
+	cli, ok := c.Client.(ClientAsync)
+	if ok {
+		req, err := c.codec.EncodeRequest(req)
+		if err != nil {
+			cb.Invoke(nil, err)
+			return
+		}
+		cb.Inject(func(resp *tikvrpc.Response, err error) (*tikvrpc.Response, error) {
+			if err != nil {
+				return nil, err
+			}
+			return c.codec.DecodeResponse(req, resp)
+		})
+		cli.SendRequestAsync(ctx, addr, req, cb)
+	} else {
+		go func() {
+			cb.Schedule(c.SendRequest(ctx, addr, req, 0))
+		}()
+	}
+}
+
 // NewTestTiKVStore creates a test store with Option
 func NewTestTiKVStore(client Client, pdClient pd.Client, clientHijack func(Client) Client, pdClientHijack func(pd.Client) pd.Client, txnLocalLatches uint, opt ...Option) (*KVStore, error) {
 	codec := apicodec.NewCodecV1(apicodec.ModeTxn)