func (db *DB) Prepared(query string) (s *Stmts) {
	s = &Stmts{query: query, sts: make(map[string]*Stmt), db: db}
	for _, c := range db.conns {
		st := c.prepared(query)
		s.mu.Lock()
		s.sts[c.addr] = st
		s.mu.Unlock()
	}
	return
}
Prepared creates a prepared statement for later queries or executions.
Multiple queries or executions may be run concurrently from the returned
statement. The caller must call the statement's Close method when the
statement is no longer needed.
package tidb
import (
	"context"
	"database/sql"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	"github.com/mapgoo-lab/atreus/pkg/log"
	"github.com/mapgoo-lab/atreus/pkg/naming"
	"github.com/mapgoo-lab/atreus/pkg/net/netutil/breaker"
	"github.com/mapgoo-lab/atreus/pkg/net/trace"
	"github.com/go-sql-driver/mysql"
	"github.com/pkg/errors"
)
const (
	_family          = "tidb_client"
	_slowLogDuration = time.Millisecond * 250
)
var (
	ErrStmtNil = errors.New("sql: prepare failed and stmt nil")
	ErrNoRows = sql.ErrNoRows
	ErrTxDone = sql.ErrTxDone
)
type DB struct {
	conf         *Config
	conns        []*conn
	idx          int64
	dis          naming.Resolver
	appid        string
	mutex        sync.RWMutex
	breakerGroup *breaker.Group
}
type conn struct {
	*sql.DB
	breaker breaker.Breaker
	conf    *Config
	addr    string
}
type Tx struct {
	db     *conn
	tx     *sql.Tx
	t      trace.Trace
	c      context.Context
	cancel func()
}
type Row struct {
	err error
	*sql.Row
	db     *conn
	query  string
	args   []interface{}
	t      trace.Trace
	cancel func()
}
func (r *Row) Scan(dest ...interface{}) (err error) {
	defer slowLog(fmt.Sprintf("Scan addr: %s query(%s) args(%+v)", r.db.addr, r.query, r.args), time.Now())
	if r.t != nil {
		defer r.t.Finish(&err)
	}
	if r.err != nil {
		err = r.err
	} else if r.Row == nil {
		err = ErrStmtNil
	}
	if err != nil {
		return
	}
	err = r.Row.Scan(dest...)
	if r.cancel != nil {
		r.cancel()
	}
	r.db.onBreaker(&err)
	if err != ErrNoRows {
		err = errors.Wrapf(err, "addr: %s, query %s args %+v", r.db.addr, r.query, r.args)
	}
	return
}
type Rows struct {
	*sql.Rows
	cancel func()
}
func (rs *Rows) Close() (err error) {
	err = errors.WithStack(rs.Rows.Close())
	if rs.cancel != nil {
		rs.cancel()
	}
	return
}
type Stmt struct {
	db    *conn
	tx    bool
	query string
	stmt  atomic.Value
	t     trace.Trace
}
type Stmts struct {
	query string
	sts   map[string]*Stmt
	mu    sync.RWMutex
	db    *DB
}
func Open(c *Config) (db *DB, err error) {
	db = &DB{conf: c, breakerGroup: breaker.NewGroup(c.Breaker)}
	cfg, err := mysql.ParseDSN(c.DSN)
	if err != nil {
		return
	}
	var dsns []string
	if cfg.Net == "discovery" {
		db.appid = cfg.Addr
		for _, addr := range db.disc() {
			dsns = append(dsns, genDSN(c.DSN, addr))
		}
	} else {
		dsns = append(dsns, c.DSN)
	}
	cs := make([]*conn, 0, len(dsns))
	for _, dsn := range dsns {
		r, err := db.connectDSN(dsn)
		if err != nil {
			return db, err
		}
		cs = append(cs, r)
	}
	db.conns = cs
	return
}
func (db *DB) connectDSN(dsn string) (c *conn, err error) {
	d, err := connect(db.conf, dsn)
	if err != nil {
		return
	}
	addr := parseDSNAddr(dsn)
	brk := db.breakerGroup.Get(addr)
	c = &conn{DB: d, breaker: brk, conf: db.conf, addr: addr}
	return
}
func connect(c *Config, dataSourceName string) (*sql.DB, error) {
	d, err := sql.Open("mysql", dataSourceName)
	if err != nil {
		err = errors.WithStack(err)
		return nil, err
	}
	d.SetMaxOpenConns(c.Active)
	d.SetMaxIdleConns(c.Idle)
	d.SetConnMaxLifetime(time.Duration(c.IdleTimeout))
	return d, nil
}
func (db *DB) conn() (c *conn) {
	db.mutex.RLock()
	c = db.conns[db.index()]
	db.mutex.RUnlock()
	return
}
func (db *DB) Begin(c context.Context) (tx *Tx, err error) {
	return db.conn().begin(c)
}
func (db *DB) Exec(c context.Context, query string, args ...interface{}) (res sql.Result, err error) {
	return db.conn().exec(c, query, args...)
}
func (db *DB) Prepare(query string) (*Stmt, error) {
	return db.conn().prepare(query)
}