update
parent
95880712b8
commit
510629a424
|
@ -6,7 +6,9 @@ import (
|
||||||
"git.makemake.in/kzkzzzz/mycommon/mylog"
|
"git.makemake.in/kzkzzzz/mycommon/mylog"
|
||||||
"github.com/goccy/go-json"
|
"github.com/goccy/go-json"
|
||||||
"github.com/rs/xid"
|
"github.com/rs/xid"
|
||||||
|
"gorm.io/gorm"
|
||||||
"gorm.io/gorm/clause"
|
"gorm.io/gorm/clause"
|
||||||
|
"gorm.io/gorm/logger"
|
||||||
"log"
|
"log"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
@ -43,6 +45,7 @@ type BatchWriterConfig struct {
|
||||||
asyncWorkerNum int
|
asyncWorkerNum int
|
||||||
clauseExpr []clause.Expression
|
clauseExpr []clause.Expression
|
||||||
debug bool
|
debug bool
|
||||||
|
noPrepare bool
|
||||||
}
|
}
|
||||||
|
|
||||||
type BatchWriter[T any] struct {
|
type BatchWriter[T any] struct {
|
||||||
|
@ -63,48 +66,6 @@ type BatchWriter[T any] struct {
|
||||||
|
|
||||||
type BatchWriterOpt func(c *BatchWriterConfig)
|
type BatchWriterOpt func(c *BatchWriterConfig)
|
||||||
|
|
||||||
func WithWriteJobNum(v int) BatchWriterOpt {
|
|
||||||
return func(c *BatchWriterConfig) {
|
|
||||||
c.jobNum = v
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func WithWriteChannelBuffer(v int) BatchWriterOpt {
|
|
||||||
return func(c *BatchWriterConfig) {
|
|
||||||
c.channelBuffer = v
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func WithWriteBatchSize(v int) BatchWriterOpt {
|
|
||||||
return func(c *BatchWriterConfig) {
|
|
||||||
c.batchSize = v
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func WithWriteIntervalTime(v time.Duration) BatchWriterOpt {
|
|
||||||
return func(c *BatchWriterConfig) {
|
|
||||||
c.batchInterval = v
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func WithAsyncWorkerNum(v int) BatchWriterOpt {
|
|
||||||
return func(c *BatchWriterConfig) {
|
|
||||||
c.asyncWorkerNum = v
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func WithClause(v ...clause.Expression) BatchWriterOpt {
|
|
||||||
return func(c *BatchWriterConfig) {
|
|
||||||
c.clauseExpr = v
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func WithDebug(v bool) BatchWriterOpt {
|
|
||||||
return func(c *BatchWriterConfig) {
|
|
||||||
c.debug = v
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func NewBatchWrite[T any](db *MysqlDb, tableName, jobName string, opts ...BatchWriterOpt) *BatchWriter[T] {
|
func NewBatchWrite[T any](db *MysqlDb, tableName, jobName string, opts ...BatchWriterOpt) *BatchWriter[T] {
|
||||||
config := &BatchWriterConfig{}
|
config := &BatchWriterConfig{}
|
||||||
for _, opt := range opts {
|
for _, opt := range opts {
|
||||||
|
@ -262,6 +223,8 @@ func (bw *BatchWriter[T]) writeToDb(bd *batchData[T]) {
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var disableGormLog = logger.Default.LogMode(logger.Silent)
|
||||||
|
|
||||||
func (bw *BatchWriter[T]) asyncWriteToDb(jobIndex int, copyDataList []T) {
|
func (bw *BatchWriter[T]) asyncWriteToDb(jobIndex int, copyDataList []T) {
|
||||||
if len(copyDataList) == 0 {
|
if len(copyDataList) == 0 {
|
||||||
return
|
return
|
||||||
|
@ -273,7 +236,20 @@ func (bw *BatchWriter[T]) asyncWriteToDb(jobIndex int, copyDataList []T) {
|
||||||
query.Clauses(bw.config.clauseExpr...)
|
query.Clauses(bw.config.clauseExpr...)
|
||||||
}
|
}
|
||||||
|
|
||||||
err := query.Create(copyDataList).Error
|
var err error
|
||||||
|
if bw.config.noPrepare {
|
||||||
|
_sql := query.ToSQL(func(tx *gorm.DB) *gorm.DB {
|
||||||
|
return tx.Session(&gorm.Session{
|
||||||
|
Logger: disableGormLog,
|
||||||
|
}).Create(©DataList)
|
||||||
|
})
|
||||||
|
|
||||||
|
err = bw.db.Table(bw.tableName).Exec(_sql).Error
|
||||||
|
|
||||||
|
} else {
|
||||||
|
err = query.Create(copyDataList).Error
|
||||||
|
}
|
||||||
|
|
||||||
if err == nil {
|
if err == nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,6 +1,10 @@
|
||||||
package mymysql
|
package mymysql
|
||||||
|
|
||||||
import "gorm.io/gorm"
|
import (
|
||||||
|
"gorm.io/gorm"
|
||||||
|
"gorm.io/gorm/clause"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
type Opt func(m *MysqlDb)
|
type Opt func(m *MysqlDb)
|
||||||
|
|
||||||
|
@ -15,3 +19,51 @@ func WithGormConfig(v *gorm.Config) Opt {
|
||||||
m.gormConfig = v
|
m.gormConfig = v
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func WithWriteJobNum(v int) BatchWriterOpt {
|
||||||
|
return func(c *BatchWriterConfig) {
|
||||||
|
c.jobNum = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func WithWriteChannelBuffer(v int) BatchWriterOpt {
|
||||||
|
return func(c *BatchWriterConfig) {
|
||||||
|
c.channelBuffer = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func WithWriteBatchSize(v int) BatchWriterOpt {
|
||||||
|
return func(c *BatchWriterConfig) {
|
||||||
|
c.batchSize = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func WithWriteIntervalTime(v time.Duration) BatchWriterOpt {
|
||||||
|
return func(c *BatchWriterConfig) {
|
||||||
|
c.batchInterval = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func WithAsyncWorkerNum(v int) BatchWriterOpt {
|
||||||
|
return func(c *BatchWriterConfig) {
|
||||||
|
c.asyncWorkerNum = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func WithClause(v ...clause.Expression) BatchWriterOpt {
|
||||||
|
return func(c *BatchWriterConfig) {
|
||||||
|
c.clauseExpr = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func WithDebug(v bool) BatchWriterOpt {
|
||||||
|
return func(c *BatchWriterConfig) {
|
||||||
|
c.debug = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func WithNoPrepare() BatchWriterOpt {
|
||||||
|
return func(c *BatchWriterConfig) {
|
||||||
|
c.noPrepare = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
Loading…
Reference in New Issue