Skip to content

Commit 032873d

Browse files
author
fudongbiao
committed
optimization
1 parent 238de23 commit 032873d

File tree

7 files changed

+234
-129
lines changed

7 files changed

+234
-129
lines changed

‎CHANGELOG.md

+10
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,16 @@
66

77
### Bugfixes
88

9+
## v2.17.08.0 [2017-09-04]
10+
11+
### Release Notes
12+
13+
该版本是一个更新版本，包含一些性能上的优化，以及完善功能细节
14+
15+
### Features
16+
17+
### Bugfixes
18+
919
## v2.17.07.0 [2017-08-03]
1020

1121
### Release Notes

‎internal/client/driver/mysql/applier.go

+146-29
Original file line numberDiff line numberDiff line change
@@ -52,11 +52,12 @@ type Applier struct {
5252
rowCopyCompleteFlag int64
5353
// copyRowsQueue should not be buffered; if buffered some non-damaging but
5454
// excessive work happens at the end of the iteration as new copy-jobs arrive befroe realizing the copy is complete
55-
copyRowsQueue chan *dumpEntry
56-
applyDataEntryQueue chan *binlog.BinlogEntry
57-
applyBinlogTxQueue chan *binlog.BinlogTx
58-
applyBinlogGroupTxQueue chan []*binlog.BinlogTx
59-
lastAppliedBinlogTx *binlog.BinlogTx
55+
copyRowsQueue chan *dumpEntry
56+
applyDataEntryQueue chan *binlog.BinlogEntry
57+
applyGrouDataEntrypQueue chan []*binlog.BinlogEntry
58+
applyBinlogTxQueue chan *binlog.BinlogTx
59+
applyBinlogGroupTxQueue chan []*binlog.BinlogTx
60+
lastAppliedBinlogTx *binlog.BinlogTx
6061

6162
natsConn *gonats.Conn
6263
waitCh chan *models.WaitResult
@@ -84,6 +85,7 @@ func NewApplier(subject, tp string, cfg *config.MySQLDriverConfig, logger *log.L
8485
allEventsUpToLockProcessed: make(chan string),
8586
copyRowsQueue: make(chan *dumpEntry, cfg.ReplChanBufferSize),
8687
applyDataEntryQueue: make(chan *binlog.BinlogEntry, cfg.ReplChanBufferSize),
88+
applyGrouDataEntrypQueue: make(chan []*binlog.BinlogEntry, cfg.ReplChanBufferSize),
8789
applyBinlogTxQueue: make(chan *binlog.BinlogTx, cfg.ReplChanBufferSize),
8890
applyBinlogGroupTxQueue: make(chan []*binlog.BinlogTx, cfg.ReplChanBufferSize),
8991
waitCh: make(chan *models.WaitResult, 1),
@@ -224,8 +226,9 @@ func (a *Applier) Run() {
224226
if err != nil {
225227
a.onError(TaskStateDead, err)
226228
}
229+
227230
for {
228-
if completeFlag != "" {
231+
if completeFlag != "" && a.rowCopyCompleteFlag == 1 {
229232
switch completeFlag {
230233
case "0":
231234
a.onError(TaskStateComplete, nil)
@@ -638,6 +641,7 @@ func (a *Applier) executeWriteFuncs() {
638641
if rowCount == string(a.applyRowCount) {
639642
a.logger.Printf("mysql.applier: Rows copy complete.number of rows:%d", a.applyRowCount)
640643
a.rowCopyComplete <- true
644+
atomic.StoreInt64(&a.rowCopyCompleteFlag, 1)
641645
break
642646
}
643647
if a.shutdown {
@@ -651,17 +655,31 @@ func (a *Applier) executeWriteFuncs() {
651655
OUTER:
652656
for {
653657
select {
654-
case binlogEntry := <-a.applyDataEntryQueue:
658+
case groupEntry := <-a.applyGrouDataEntrypQueue:
655659
{
656-
if nil == binlogEntry {
660+
if len(groupEntry) == 0 {
657661
continue
658662
}
659663
/*if a.mysqlContext.MySQLServerUuid == binlogEntry.Coordinates.OSID {
660664
continue
661665
}*/
662-
if err := a.ApplyBinlogEvent(a.db, binlogEntry); err != nil {
663-
a.onError(TaskStateDead, err)
664-
break OUTER
666+
667+
for idx, binlogEntry := range groupEntry {
668+
dbApplier = a.dbs[idx%a.mysqlContext.ParallelWorkers]
669+
//go func(entry *binlog.BinlogEntry) {
670+
//a.wg.Add(1)
671+
if err := a.ApplyBinlogEvent(dbApplier, binlogEntry); err != nil {
672+
a.onError(TaskStateDead, err)
673+
}
674+
//a.wg.Done()
675+
//}(binlogEntry)
676+
}
677+
//a.wg.Wait() // Waiting for all goroutines to finish
678+
679+
//a.logger.Debugf("mysql.applier: apply binlogEntry: %+v", groupEntry[len(groupEntry)-1].Coordinates.GNO)
680+
681+
if !a.shutdown {
682+
a.mysqlContext.Gtid = fmt.Sprintf("%s:1-%d", groupEntry[len(groupEntry)-1].Coordinates.SID, groupEntry[len(groupEntry)-1].Coordinates.GNO)
665683
}
666684
}
667685
case groupTx := <-a.applyBinlogGroupTxQueue:
@@ -743,7 +761,7 @@ func (a *Applier) initiateStreaming() error {
743761
if err := Decode(m.Data, &binlogEntry); err != nil {
744762
a.onError(TaskStateDead, err)
745763
}
746-
//a.logger.Debugf("mysql.applier: received binlogEntry: %+v", binlogEntry.Coordinates.GNO)
764+
//a.logger.Debugf("mysql.applier: received binlogEntry GNO: %+v,LastCommitted:%+v", binlogEntry.Coordinates.GNO,binlogEntry.Coordinates.LastCommitted)
747765
a.applyDataEntryQueue <- binlogEntry
748766
a.currentCoordinates.RetrievedGtidSet = fmt.Sprintf("%s:%d", binlogEntry.Coordinates.SID, binlogEntry.Coordinates.GNO)
749767

@@ -754,6 +772,48 @@ func (a *Applier) initiateStreaming() error {
754772
if err != nil {
755773
return err
756774
}
775+
776+
go func() {
777+
var lastCommitted int64
778+
//timeout := time.After(100 * time.Millisecond)
779+
groupEntry := []*binlog.BinlogEntry{}
780+
OUTER:
781+
for {
782+
select {
783+
case binlogEntry := <-a.applyDataEntryQueue:
784+
if nil == binlogEntry {
785+
continue
786+
}
787+
/*if a.mysqlContext.MySQLServerUuid == binlogTx.SID {
788+
continue
789+
}*/
790+
if a.mysqlContext.ParallelWorkers <= 1 {
791+
if err := a.ApplyBinlogEvent(a.dbs[0], binlogEntry); err != nil {
792+
a.onError(TaskStateDead, err)
793+
break OUTER
794+
}
795+
} else {
796+
if binlogEntry.Coordinates.LastCommitted == lastCommitted {
797+
groupEntry = append(groupEntry, binlogEntry)
798+
} else {
799+
if len(groupEntry) != 0 {
800+
a.applyGrouDataEntrypQueue <- groupEntry
801+
groupEntry = []*binlog.BinlogEntry{}
802+
}
803+
groupEntry = append(groupEntry, binlogEntry)
804+
}
805+
lastCommitted = binlogEntry.Coordinates.LastCommitted
806+
}
807+
case <-time.After(100 * time.Millisecond):
808+
if len(groupEntry) != 0 {
809+
a.applyGrouDataEntrypQueue <- groupEntry
810+
groupEntry = []*binlog.BinlogEntry{}
811+
}
812+
case <-a.shutdownCh:
813+
break OUTER
814+
}
815+
}
816+
}()
757817
} else {
758818
_, err := a.natsConn.Subscribe(fmt.Sprintf("%s_incr", a.subject), func(m *gonats.Msg) {
759819
var binlogTx []*binlog.BinlogTx
@@ -1477,33 +1537,79 @@ func (a *Applier) getSharedColumns(originalColumns, columns *umconf.ColumnList,
14771537
// buildDMLEventQuery creates a query to operate on the ghost table, based on an intercepted binlog
14781538
// event entry on the original table.
14791539
func (a *Applier) buildDMLEventQuery(dmlEvent binlog.DataEvent) (query string, args []interface{}, rowsDelta int64, err error) {
1480-
destTableColumns, _, err := a.InspectTableColumnsAndUniqueKeys(dmlEvent.DatabaseName, dmlEvent.TableName)
1481-
if err != nil {
1482-
return "", args, 0, err
1540+
/*var destTableColumns *umconf.ColumnList
1541+
if len(a.mysqlContext.ReplicateDoDb) ==0 {
1542+
tableColumns, _, err := a.InspectTableColumnsAndUniqueKeys(dmlEvent.DatabaseName, dmlEvent.TableName)
1543+
if err != nil {
1544+
return "", args, 0, err
1545+
}
1546+
tb:=&config.Table{
1547+
TableSchema:dmlEvent.DatabaseName,
1548+
TableName:dmlEvent.TableName,
1549+
OriginalTableColumns:tableColumns,
1550+
}
1551+
db:= &config.DataSource{
1552+
TableSchema:dmlEvent.DatabaseName,
1553+
Tables:[]*config.Table{tb},
1554+
}
1555+
a.mysqlContext.ReplicateDoDb = append(a.mysqlContext.ReplicateDoDb,db)
1556+
}else{
1557+
L:
1558+
for _,db:=range a.mysqlContext.ReplicateDoDb{
1559+
if db.TableSchema != dmlEvent.DatabaseName {
1560+
continue
1561+
}
1562+
for _,tb:=range db.Tables {
1563+
if tb.TableName == dmlEvent.TableName && tb.OriginalTableColumns!=nil{
1564+
break L
1565+
}
1566+
}
1567+
tableColumns, _, err := a.InspectTableColumnsAndUniqueKeys(dmlEvent.DatabaseName, dmlEvent.TableName)
1568+
if err != nil {
1569+
return "", args, 0, err
1570+
}
1571+
tb:=&config.Table{
1572+
TableSchema:dmlEvent.DatabaseName,
1573+
TableName:dmlEvent.TableName,
1574+
OriginalTableColumns:tableColumns,
1575+
}
1576+
db.Tables = append(db.Tables,tb)
1577+
}
14831578
}
1484-
/*_, err = getSharedUniqueKeys(destTableUniqueKeys, destTableUniqueKeys)
1485-
if err != nil {
1486-
return "", args, 0, err
1487-
}*/
1488-
/*if len(sharedUniqueKeys) == 0 {
1489-
return "", args, 0, fmt.Errorf("No shared unique key can be found after ALTER! Bailing out")
1579+
for _,db:=range a.mysqlContext.ReplicateDoDb{
1580+
if db.TableSchema == dmlEvent.DatabaseName {
1581+
for _,tb:=range db.Tables {
1582+
if tb.TableName == dmlEvent.TableName {
1583+
destTableColumns = tb.OriginalTableColumns
1584+
}
1585+
}
1586+
}
1587+
}
1588+
if destTableColumns ==nil{
1589+
destTableColumns, _, err = a.InspectTableColumnsAndUniqueKeys(dmlEvent.DatabaseName, dmlEvent.TableName)
1590+
if err != nil {
1591+
return "", args, 0, err
1592+
}
14901593
}*/
1491-
sharedColumns, mappedSharedColumns := a.getSharedColumns(destTableColumns, destTableColumns, a.parser.GetNonTrivialRenames())
1492-
//a.logger.Printf("mysql.applier: shared columns are %s", sharedColumns)
1594+
14931595
switch dmlEvent.DML {
14941596
case binlog.DeleteDML:
14951597
{
1496-
query, uniqueKeyArgs, err := sql.BuildDMLDeleteQuery(dmlEvent.DatabaseName, dmlEvent.TableName, destTableColumns, destTableColumns, dmlEvent.WhereColumnValues.GetAbstractValues())
1598+
tableColumns, err := base.GetTableColumns(a.db, dmlEvent.DatabaseName, dmlEvent.TableName)
1599+
query, uniqueKeyArgs, err := sql.BuildDMLDeleteQuery(dmlEvent.DatabaseName, dmlEvent.TableName, tableColumns, dmlEvent.WhereColumnValues.GetAbstractValues())
14971600
return query, uniqueKeyArgs, -1, err
14981601
}
14991602
case binlog.InsertDML:
15001603
{
1501-
query, sharedArgs, err := sql.BuildDMLInsertQuery(dmlEvent.DatabaseName, dmlEvent.TableName, destTableColumns, sharedColumns, mappedSharedColumns, dmlEvent.NewColumnValues.GetAbstractValues())
1604+
//query, sharedArgs,err := sql.BuildDMLInsertQuery(dmlEvent.DatabaseName, dmlEvent.TableName,dmlEvent.ColumnCount,dmlEvent.NewColumnValues)
1605+
tableColumns, err := base.GetTableColumns(a.db, dmlEvent.DatabaseName, dmlEvent.TableName)
1606+
query, sharedArgs, err := sql.BuildDMLInsertQuery(dmlEvent.DatabaseName, dmlEvent.TableName, tableColumns, tableColumns, tableColumns, dmlEvent.NewColumnValues)
15021607
return query, sharedArgs, 1, err
15031608
}
15041609
case binlog.UpdateDML:
15051610
{
1506-
query, sharedArgs, uniqueKeyArgs, err := sql.BuildDMLUpdateQuery(dmlEvent.DatabaseName, dmlEvent.TableName, destTableColumns, sharedColumns, mappedSharedColumns, destTableColumns, dmlEvent.NewColumnValues.GetAbstractValues(), dmlEvent.WhereColumnValues.GetAbstractValues())
1611+
tableColumns, err := base.GetTableColumns(a.db, dmlEvent.DatabaseName, dmlEvent.TableName)
1612+
query, sharedArgs, uniqueKeyArgs, err := sql.BuildDMLUpdateQuery(dmlEvent.DatabaseName, dmlEvent.TableName, tableColumns, tableColumns, tableColumns, tableColumns, dmlEvent.NewColumnValues[0].GetAbstractValues(), dmlEvent.WhereColumnValues.GetAbstractValues())
15071613
args = append(args, sharedArgs...)
15081614
args = append(args, uniqueKeyArgs...)
15091615
return query, args, 0, err
@@ -1513,18 +1619,19 @@ func (a *Applier) buildDMLEventQuery(dmlEvent binlog.DataEvent) (query string, a
15131619
}
15141620

15151621
// ApplyEventQueries applies multiple DML queries onto the dest table
1516-
func (a *Applier) ApplyBinlogEvent(db *gosql.DB, binlogEntry *binlog.BinlogEntry) error {
1622+
func (a *Applier) ApplyBinlogEvent(dbApplier *sql.DB, binlogEntry *binlog.BinlogEntry) error {
15171623
var totalDelta int64
15181624

1519-
interval, err := base.SelectGtidExecuted(db, binlogEntry.Coordinates.SID, binlogEntry.Coordinates.GNO)
1625+
interval, err := base.SelectGtidExecuted(dbApplier.Db, binlogEntry.Coordinates.SID, binlogEntry.Coordinates.GNO)
15201626
if err != nil {
15211627
return err
15221628
}
15231629
if interval == "" {
15241630
return nil
15251631
}
15261632

1527-
tx, err := db.Begin()
1633+
dbApplier.DbMutex.Lock()
1634+
tx, err := dbApplier.Db.Begin()
15281635
if err != nil {
15291636
return err
15301637
}
@@ -1538,6 +1645,7 @@ func (a *Applier) ApplyBinlogEvent(db *gosql.DB, binlogEntry *binlog.BinlogEntry
15381645
a.currentCoordinates.ExecutedGtidSet = fmt.Sprintf("%s:%d", binlogEntry.Coordinates.SID, binlogEntry.Coordinates.GNO)
15391646
a.mysqlContext.Gtid = fmt.Sprintf("%s:1-%d", binlogEntry.Coordinates.SID, binlogEntry.Coordinates.GNO)
15401647
}
1648+
dbApplier.DbMutex.Unlock()
15411649
}()
15421650
sessionQuery := `SET @@session.foreign_key_checks = 0`
15431651
if _, err := tx.Exec(sessionQuery); err != nil {
@@ -1573,6 +1681,15 @@ func (a *Applier) ApplyBinlogEvent(db *gosql.DB, binlogEntry *binlog.BinlogEntry
15731681
a.logger.Warnf("mysql.applier: Ignore error: %v", err)
15741682
}
15751683
}
1684+
/*for _,db:=range a.mysqlContext.ReplicateDoDb{
1685+
for _,tb:=range db.Tables {
1686+
tableColumns, _, err := a.InspectTableColumnsAndUniqueKeys(tb.TableSchema, tb.TableName)
1687+
if err != nil {
1688+
return err
1689+
}
1690+
tb.OriginalTableColumns = tableColumns
1691+
}
1692+
}*/
15761693
default:
15771694
query, args, rowDelta, err := a.buildDMLEventQuery(event)
15781695
if err != nil {

‎internal/client/driver/mysql/binlog/binlog_event.go

+4-2
Original file line numberDiff line numberDiff line change
@@ -78,16 +78,18 @@ type DataEvent struct {
7878
DatabaseName string
7979
TableName string
8080
DML EventDML
81+
ColumnCount int
8182
OriginalTableColumns *mysql.ColumnList
8283
WhereColumnValues *mysql.ColumnValues
83-
NewColumnValues *mysql.ColumnValues
84+
NewColumnValues []*mysql.ColumnValues
8485
}
8586

86-
func NewDataEvent(databaseName, tableName string, dml EventDML) DataEvent {
87+
func NewDataEvent(databaseName, tableName string, dml EventDML, columnCount int) DataEvent {
8788
event := DataEvent{
8889
DatabaseName: databaseName,
8990
TableName: tableName,
9091
DML: dml,
92+
ColumnCount: columnCount,
9193
}
9294
return event
9395
}

‎internal/client/driver/mysql/binlog/binlog_reader.go

+7-3
Original file line numberDiff line numberDiff line change
@@ -140,6 +140,7 @@ func (b *BinlogReader) handleEvent(ev *replication.BinlogEvent, entriesChannel c
140140
u, _ := uuid.FromBytes(evt.GTID.SID)
141141
b.currentCoordinates.SID = u.String()
142142
b.currentCoordinates.GNO = evt.GTID.GNO
143+
b.currentCoordinates.LastCommitted = evt.GTID.LastCommitted
143144
b.currentBinlogEntry = NewBinlogEntryAt(b.currentCoordinates)
144145
} else {
145146
evt := ev.Event.(*replication.GTIDEvent)
@@ -214,6 +215,7 @@ func (b *BinlogReader) handleEvent(ev *replication.BinlogEvent, entriesChannel c
214215
string(rowsEvent.Table.Schema),
215216
string(rowsEvent.Table.Table),
216217
dml,
218+
int(rowsEvent.ColumnCount),
217219
)
218220

219221
/*originalTableColumns, _, err := b.InspectTableColumnsAndUniqueKeys(string(rowsEvent.Table.Schema), string(rowsEvent.Table.Table))
@@ -231,24 +233,26 @@ func (b *BinlogReader) handleEvent(ev *replication.BinlogEvent, entriesChannel c
231233
switch dml {
232234
case InsertDML:
233235
{
234-
dmlEvent.NewColumnValues = mysql.ToColumnValues(row)
236+
//dmlEvent.NewColumnValues = mysql.ToColumnValues(row)
237+
dmlEvent.NewColumnValues = append(dmlEvent.NewColumnValues, mysql.ToColumnValues(row))
235238
}
236239
case UpdateDML:
237240
{
238241
dmlEvent.WhereColumnValues = mysql.ToColumnValues(row)
239-
dmlEvent.NewColumnValues = mysql.ToColumnValues(rowsEvent.Rows[i+1])
242+
//dmlEvent.NewColumnValues = mysql.ToColumnValues(rowsEvent.Rows[i+1])
243+
dmlEvent.NewColumnValues = append(dmlEvent.NewColumnValues, mysql.ToColumnValues(rowsEvent.Rows[i+1]))
240244
}
241245
case DeleteDML:
242246
{
243247
dmlEvent.WhereColumnValues = mysql.ToColumnValues(row)
244248
}
245249
}
246-
b.currentBinlogEntry.Events = append(b.currentBinlogEntry.Events, dmlEvent)
247250
// The channel will do the throttling. Whoever is reding from the channel
248251
// decides whether action is taken sycnhronously (meaning we wait before
249252
// next iteration) or asynchronously (we keep pushing more events)
250253
// In reality, reads will be synchronous
251254
}
255+
b.currentBinlogEntry.Events = append(b.currentBinlogEntry.Events, dmlEvent)
252256
return nil
253257
}
254258
}

‎internal/client/driver/mysql/extractor.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -708,7 +708,7 @@ func (e *Extractor) validateAndReadTimeZone() error {
708708
return err
709709
}
710710

711-
e.logger.Printf("mysql.applier: Will use time_zone='%s' on extractor", e.mysqlContext.TimeZone)
711+
e.logger.Printf("mysql.extractor: Will use time_zone='%s' on extractor", e.mysqlContext.TimeZone)
712712
return nil
713713
}
714714

0 commit comments

Comments
 (0)