Merge branch 'dev' of http://git.legu.cc/liwei_3d/go_dreamfactory into liwei

comm/core.go

@@ -57,7 +57,7 @@ type ISC_GateRouteComp interface {
 }
 
 type Autogenerated struct {
-	ID  string `json:"_id"`
+	ID  string `json:"ID,omitempty" bson:"_id"`
 	UID string `json:"uid"`
 	Act string `json:"act"` // insert update delete
 	D   []interface{}

lego/sys/mgo/mgo.go

@@ -3,7 +3,6 @@ package mgo
 import (
 	"context"
 	"fmt"
-	"time"
 
 	"go_dreamfactory/lego/core"
 
@@ -12,7 +11,6 @@ import (
 	"go.mongodb.org/mongo-driver/mongo/options"
 	"go.mongodb.org/mongo-driver/mongo/readconcern"
 	"go.mongodb.org/mongo-driver/mongo/readpref"
-	"go.mongodb.org/mongo-driver/mongo/writeconcern"
 )
 
 func newSys(options Options) (sys *Mongodb, err error) {
@@ -28,20 +26,20 @@ type Mongodb struct {
 }
 
 func (this *Mongodb) init() (err error) {
-	want, err := readpref.New(readpref.SecondaryMode) //表示只使用辅助节点
+	// want, err := readpref.New(readpref.SecondaryMode) //表示只使用辅助节点
-	if err != nil {
+	// if err != nil {
-		return fmt.Errorf("数据库设置辅助节点 err=%s", err.Error())
+	// 	return fmt.Errorf("数据库设置辅助节点 err=%s", err.Error())
-	}
+	// }
-	wc := writeconcern.New(writeconcern.WMajority())
+	// wc := writeconcern.New(writeconcern.W(1))
 	readconcern.Majority()
 	//链接mongo服务
 	opt := options.Client().ApplyURI(this.options.MongodbUrl)
-	opt.SetLocalThreshold(3 * time.Second)       //只使用与mongo操作耗时小于3秒的
+	// opt.SetLocalThreshold(3 * time.Second)       //只使用与mongo操作耗时小于3秒的
-	opt.SetMaxConnIdleTime(5 * time.Second)      //指定连接可以保持空闲的最大毫秒数
+	// opt.SetMaxConnIdleTime(5 * time.Second)      //指定连接可以保持空闲的最大毫秒数
 	opt.SetMaxPoolSize(this.options.MaxPoolSize) //使用最大的连接数
-	opt.SetReadPreference(want)                  //表示只使用辅助节点
+	// opt.SetReadPreference(want)                  //表示只使用辅助节点
-	opt.SetReadConcern(readconcern.Majority())   //指定查询应返回实例的最新数据确认为，已写入副本集中的大多数成员
+	// opt.SetReadConcern(readconcern.Majority())   //指定查询应返回实例的最新数据确认为，已写入副本集中的大多数成员
-	opt.SetWriteConcern(wc)                      //请求确认写操作传播到大多数mongod实例
+	// opt.SetWriteConcern(wc)                      //请求确认写操作传播到大多数mongod实例
 	if client, err := mongo.Connect(this.getContext(), opt); err != nil {
 		return fmt.Errorf("连接数据库错误 err=%s", err.Error())
 	} else {

lego/sys/mgo/options.go

@@ -40,7 +40,7 @@ func SetTimeOut(v time.Duration) Option {
 
 func newOptions(config map[string]interface{}, opts ...Option) Options {
 	options := Options{
-		MaxPoolSize: 1000,
+		MaxPoolSize: 100,
 		TimeOut:     time.Second * 3,
 	}
 	if config != nil {
@@ -54,7 +54,7 @@ func newOptions(config map[string]interface{}, opts ...Option) Options {
 
 func newOptionsByOption(opts ...Option) Options {
 	options := Options{
-		MaxPoolSize: 1000,
+		MaxPoolSize: 100,
 		TimeOut:     time.Second * 3,
 	}
 	for _, o := range opts {

lego/sys/redis/options.go

@@ -63,7 +63,7 @@ func SetRedis_Single_PoolSize(v int) Option {
 		o.Redis_Single_PoolSize = v
 	}
 }
-func Redis_Cluster_Addr(v []string) Option {
+func SetRedis_Cluster_Addr(v []string) Option {
 	return func(o *Options) {
 		o.Redis_Cluster_Addr = v
 	}

modules/model/api.go

@@ -18,3 +18,8 @@ func (this *Api_Comp) Init(service core.IService, module core.IModule, comp core
 	this.module = module.(*Model)
 	return
 }
+
+func (this *Api_Comp) Start() (err error) {
+
+	return
+}

modules/model/db_comp.go

@@ -5,9 +5,9 @@ import (
 	"errors"
 	"go_dreamfactory/comm"
 	"go_dreamfactory/lego/core"
-	"go_dreamfactory/lego/core/cbase"
 	"go_dreamfactory/lego/sys/log"
-	"go_dreamfactory/lego/sys/mgo"
+	"go_dreamfactory/modules"
+	"time"
 
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/mongo/options"
@@ -18,8 +18,7 @@ const (
 )
 
 type DB_Comp struct {
-	cbase.ModuleCompBase
+	modules.MComp_DBComp
-	mgo mgo.ISys
 }
 
 // type data struct {
@@ -44,13 +43,16 @@ type IModel interface {
 }
 
 func (this *DB_Comp) Model_UpdateDBByLog() (err error) {
+	//_startTime := time.Now().UnixNano()
 
 	_delID := make([]string, 0) // 处理完成要删除的id
-	_data, err := this.mgo.Find(DB_ModelTable, bson.M{}, options.Find().SetLimit(int64(WriteMaxNum)))
+	_data, err := this.DB.Find(DB_ModelTable, bson.M{}, options.Find().SetLimit(int64(WriteMaxNum)))
 	if err != nil {
 		return err
 	}
+
 	for _data.Next(context.TODO()) { // 处理删除逻辑
+
 		data := &comm.Autogenerated{}
 		if err = _data.Decode(data); err == nil {
 			_delID = append(_delID, data.ID)
@@ -59,23 +61,34 @@ func (this *DB_Comp) Model_UpdateDBByLog() (err error) {
 			continue
 		}
 
-		log.Debugf("======= insert log : %+v =======", data)
+		//log.Debugf("======= insert log : %+v =======", data)
 		if data.Act == string(comm.LogHandleType_Insert) {
 			if len(data.D) < 2 { // 参数校验
 				log.Errorf("parameter len _id : %s,uid : %s d.len:%v", data.ID, data.UID, len(data.D))
 				return errors.New("parameter len err")
 			}
-			query := data.D[1].([]interface{})
+
-			_, err := this.mgo.InsertMany(data.D[0].(core.SqlTable), query)
+			query := data.D[1]
-			if err != nil {
+			_obj := bson.M{}
-				log.Errorf("insert %s db err:%v", data.D[0].(core.SqlTable), err)
+			for _, v := range query.(bson.D) {
+				_obj[v.Key] = v.Value
 			}
+			_key := data.D[0].(string)
+
+			_, err := this.DB.InsertOne(core.SqlTable(_key), _obj)
+			if err != nil {
+				log.Errorf("insert %s db err:%v", "mail", err)
+			}
+			// _, err = this.DB.DeleteOne(DB_ModelTable, bson.M{"_id": data.ID})
+			// if err != nil {
+			// 	log.Errorf("insert %s db err:%v", data.ID, err)
+			// }
 		} else if data.Act == string(comm.LogHandleType_Delete) {
 			if len(data.D) < 2 { // 参数校验
 				log.Errorf("parameter len _id : %s,uid : %s d.len:%v", data.ID, data.UID, len(data.D))
 				return errors.New("parameter len err")
 			}
-			_, err := this.mgo.DeleteMany(data.D[0].(core.SqlTable), data.D[1].(map[string]interface{}))
+			_, err := this.DB.DeleteMany(data.D[0].(core.SqlTable), data.D[1].(map[string]interface{}))
 			if err != nil {
 				log.Errorf("delete %s db err:%v", data.D[0].(core.SqlTable), err)
 			}
@@ -93,20 +106,22 @@ func (this *DB_Comp) Model_UpdateDBByLog() (err error) {
 			for k, v := range query {
 				_obj.Query[k] = v
 			}
-			this.mgo.UpdateMany(data.D[0].(core.SqlTable), _obj.Selector, _obj.Query)
+			this.DB.UpdateMany(data.D[0].(core.SqlTable), _obj.Selector, _obj.Query)
 		}
 	}
 	// 批量删除已处理的数据
-	_, err = this.mgo.DeleteMany(DB_ModelTable, bson.M{"_id": bson.M{"$inc": _delID}}, options.Delete())
+	_, err = this.DB.DeleteMany(DB_ModelTable, bson.M{"_id": bson.M{"$in": _delID}}, options.Delete())
 	if err != nil {
 		log.Errorf("del err %v", err)
 	}
+	// _endTime := time.Now().UnixNano()
+	// log.Debugf("==============subTime=%d===========del len = %d", _endTime-_startTime, len(_delID))
 	return
 }
 
 func (this *DB_Comp) Model_InsertDBByLog(data *comm.Autogenerated) (err error) {
 
-	_, err = this.mgo.InsertOne(DB_ModelTable, data)
+	_, err = this.DB.InsertOne(DB_ModelTable, data)
 	if err != nil {
 		log.Errorf("insert model db err %v", err)
 	}
@@ -114,9 +129,10 @@ func (this *DB_Comp) Model_InsertDBByLog(data *comm.Autogenerated) (err error) {
 }
 
 func (this *DB_Comp) Model_UpdateUserDataByUid(uid string) (err error) {
+	_startTime := time.Now().UnixNano()
 
 	_delID := make([]string, 0) // 处理完成要删除的id
-	_data, err := this.mgo.Find(DB_ModelTable, bson.M{"userid": uid}, options.Find())
+	_data, err := this.DB.Find(DB_ModelTable, bson.M{"userid": uid}, options.Find())
 	if err != nil {
 		return err
 	}
@@ -135,7 +151,7 @@ func (this *DB_Comp) Model_UpdateUserDataByUid(uid string) (err error) {
 				log.Errorf("parameter len _id : %s,uid : %s d.len:%v", data.ID, data.UID, len(data.D))
 				return errors.New("parameter len err")
 			}
-			_, err := this.mgo.InsertMany(data.D[0].(core.SqlTable), data.D[1].([]interface{}))
+			_, err := this.DB.InsertMany(data.D[0].(core.SqlTable), data.D[1].([]interface{}))
 			if err != nil {
 				log.Errorf("insert %s db err:%v", data.D[0].(core.SqlTable), err)
 			}
@@ -144,7 +160,7 @@ func (this *DB_Comp) Model_UpdateUserDataByUid(uid string) (err error) {
 				log.Errorf("parameter len _id : %s,uid : %s d.len:%v", data.ID, data.UID, len(data.D))
 				return errors.New("parameter len err")
 			}
-			_, err := this.mgo.DeleteMany(data.D[0].(core.SqlTable), data.D[1].(map[string]interface{}))
+			_, err := this.DB.DeleteMany(data.D[0].(core.SqlTable), data.D[1].(map[string]interface{}))
 			if err != nil {
 				log.Errorf("delete %s db err:%v", data.D[0].(core.SqlTable), err)
 			}
@@ -162,13 +178,16 @@ func (this *DB_Comp) Model_UpdateUserDataByUid(uid string) (err error) {
 			for k, v := range query {
 				_obj.Query[k] = v
 			}
-			this.mgo.FindOneAndUpdate(data.D[0].(core.SqlTable), _obj.Selector, _obj.Query)
+			this.DB.FindOneAndUpdate(data.D[0].(core.SqlTable), _obj.Selector, _obj.Query)
 		}
 	}
 	// 批量删除已处理的数据
-	_, err = this.mgo.DeleteMany(DB_ModelTable, bson.M{"_id": bson.M{"$inc": _delID}}, options.Delete())
+	_, err = this.DB.DeleteMany(DB_ModelTable, bson.M{"_id": bson.M{"$inc": _delID}}, options.Delete())
 	if err != nil {
 		log.Errorf("del err %v", err)
 	}
+
+	_endTime := time.Now().UnixNano()
+	log.Debugf("==============subTime============del len = %d", _endTime-_startTime, len(_delID))
 	return
 }

modules/model/dbservice_comp.go

@@ -8,28 +8,29 @@ import (
 
 type DBService_Comp struct {
 	cbase.ModuleCompBase
-	task chan string
+	task   chan string
-	DB_Comp
+	module *Model
 }
 
 func (this *DBService_Comp) Init(service core.IService, module core.IModule, comp core.IModuleComp, options core.IModuleOptions) (err error) {
 	this.ModuleCompBase.Init(service, module, comp, options)
-
+	this.module = module.(*Model)
 	return
 }
 
-func (this *DBService_Comp) Start(err error) {
+func (this *DBService_Comp) Start() (err error) {
-	this.ModuleCompBase.Start()
+	err = this.ModuleCompBase.Start()
 	go this.run()
+	return
 }
 
 func (this *DBService_Comp) run() {
 	for {
 		select {
 		case v := <-this.task:
-			this.Model_UpdateUserDataByUid(v)
+			this.module.db_comp.Model_UpdateUserDataByUid(v)
-		case <-time.After(time.Second):
+		case <-time.After(time.Second * 2):
-			this.Model_UpdateDBByLog()
+			this.module.db_comp.Model_UpdateDBByLog()
 		}
 	}
 }

modules/model/mail_test.go

@@ -1,4 +1,4 @@
-package modules
+package model
 
 import (
 	"go_dreamfactory/lego/sys/log"
@@ -7,10 +7,6 @@ import (
 	"time"
 )
 
-var (
-	obj MComp_DBComp
-)
-
 func TestCreatemoudles(t *testing.T) {
 	_mail := &pb.DB_MailData{
 
@@ -23,7 +19,7 @@ func TestCreatemoudles(t *testing.T) {
 		Check:      false,
 		Reward:     false,
 	}
-	obj.InsertModelLogs("mail", "uid123", _mail)
+	// obj.InsertModelLogs("mail", "uid123", _mail)
 
 	log.Debugf("insert : %v", _mail)
 }

modules/model/module.go

@@ -4,7 +4,6 @@ import (
 	"go_dreamfactory/comm"
 	"go_dreamfactory/lego/core"
 	"go_dreamfactory/modules"
-	"time"
 )
 
 func NewModule() core.IModule {
@@ -15,6 +14,8 @@ func NewModule() core.IModule {
 type Model struct {
 	modules.ModuleBase
 	api_comp       *Api_Comp
+	db_comp        *DB_Comp
+	db_service     *DBService_Comp
 	configure_comp *Configure_Comp
 }
 
@@ -24,12 +25,6 @@ func (this *Model) Init(service core.IService, module core.IModule, options core
 	return
 }
 
-func (this *Model) Start() (err error) {
-	err = this.ModuleBase.Start()
-	//go this.RunWriteDB()
-	return
-}
-
 func (this *Model) GetType() core.M_Modules {
 	return comm.SM_LogModelModule
 }
@@ -37,12 +32,7 @@ func (this *Model) GetType() core.M_Modules {
 func (this *Model) OnInstallComp() {
 	this.ModuleBase.OnInstallComp()
 	this.api_comp = this.RegisterComp(new(Api_Comp)).(*Api_Comp)
+	this.db_comp = this.RegisterComp(new(DB_Comp)).(*DB_Comp)
+	this.db_service = this.RegisterComp(new(DBService_Comp)).(*DBService_Comp)
 	this.configure_comp = this.RegisterComp(new(Configure_Comp)).(*Configure_Comp)
 }
-
-func (this *Model) RunWriteDB() {
-	for {
-		time.Sleep(time.Second)
-
-	}
-}

modules/pack/pack_test.go

@@ -1,27 +1,72 @@
-package pack_test
+package pack
 
 import (
 	"fmt"
+	"go_dreamfactory/lego"
+	"go_dreamfactory/lego/base/rpcx"
+	"go_dreamfactory/lego/core"
+	"go_dreamfactory/lego/sys/log"
+	"go_dreamfactory/services"
 	"go_dreamfactory/sys/cache"
 	"go_dreamfactory/sys/configure"
 	"go_dreamfactory/sys/db"
 	"os"
 	"testing"
+	"time"
 )
 
+func newService(ops ...rpcx.Option) core.IService {
+	s := new(TestService)
+	s.Configure(ops...)
+	return s
+}
+
+//梦工厂基础服务对象
+type TestService struct {
+	rpcx.RPCXService
+}
+
+//初始化相关系统
+func (this *TestService) InitSys() {
+	this.RPCXService.InitSys()
+	if err := cache.OnInit(this.GetSettings().Sys["cache"]); err != nil {
+		panic(fmt.Sprintf("init sys.cache err: %s", err.Error()))
+	} else {
+		log.Infof("init sys.cache success!")
+	}
+	if err := db.OnInit(this.GetSettings().Sys["db"]); err != nil {
+		panic(fmt.Sprintf("init sys.db err: %s", err.Error()))
+	} else {
+		log.Infof("init sys.db success!")
+	}
+	if err := configure.OnInit(this.GetSettings().Sys["configure"]); err != nil {
+		panic(fmt.Sprintf("init sys.configure err: %s", err.Error()))
+	} else {
+		log.Infof("init sys.configure success!")
+	}
+}
+
+var module = new(Pack)
+
 //测试环境下初始化db和cache 系统
 func TestMain(m *testing.M) {
-	if err := db.OnInit(nil, db.Set_MongodbUrl("mongodb://admin:123456@10.0.0.9:27018"), db.Set_MongodbDatabase("dreamfactory")); err != nil {
+	s := newService(
-		fmt.Printf("err:%v\n", err)
+		rpcx.SetConfPath("../../bin/conf/worker_1.yaml"),
-		return
+		rpcx.SetVersion("1.0.0.0"),
-	}
+	)
-	if err := cache.OnInit(nil, cache.Set_Redis_Addr([]string{"10.0.0.9:9001", "10.0.0.9:9002", "10.0.0.9:9003", "10.0.1.45:9004", "10.0.1.45:9005", "10.0.1.45:9006"}), cache.Set_Redis_Password("")); err != nil {
+	s.OnInstallComp( //装备组件
-		fmt.Printf("err:%v\n", err)
+		services.NewGateRouteComp(), //此服务需要接受用户的消息 需要装备网关组件
-		return
+	)
-	}
+	go func() {
-	if err := configure.OnInit(nil); err != nil {
+		lego.Run(s, //运行模块
-		fmt.Printf("err:%v\n", err)
+			module,
-		return
+		)
-	}
+	}()
+	time.Sleep(time.Second)
 	defer os.Exit(m.Run())
 }
+
+func Test_Log(t *testing.T) {
+	items, err := module.db_comp.Pack_QueryUserPack("liwei1dao")
+	log.Debugf("item:%v err:%v", items, err)
+}

sys/cache/cache.go

@@ -21,7 +21,7 @@ type Cache struct {
 func (this *Cache) init() (err error) {
 	this.redis, err = redis.NewSys(
 		redis.SetRedisType(redis.Redis_Cluster),
-		redis.Redis_Cluster_Addr(this.options.Redis_Addr),
+		redis.SetRedis_Cluster_Addr(this.options.Redis_Addr),
 		redis.SetRedis_Cluster_Password(this.options.Redis_Password))
 	return
 }

sys/cache/init_test.go

@@ -2,10 +2,16 @@ package cache_test
 
 import (
 	"fmt"
+	"go_dreamfactory/comm"
+	"go_dreamfactory/lego/sys/log"
+	"go_dreamfactory/pb"
 	"go_dreamfactory/sys/cache"
 	"go_dreamfactory/sys/db"
 	"os"
 	"testing"
+	"time"
+
+	"go.mongodb.org/mongo-driver/bson/primitive"
 )
 
 //测试环境下初始化db和cache 系统
@@ -18,7 +24,36 @@ func TestMain(m *testing.M) {
 		fmt.Printf("err:%v\n", err)
 		return
 	}
+	for i := 0; i < 50000; i++ {
+		//go func() {
+		_mail := &pb.DB_MailData{
+			ObjId:  primitive.NewObjectID().Hex(),
+			UserId: "uid123",
+			Title:  "系统邮件",
 
+			Contex:     "恭喜获得专属礼包一份",
+			CreateTime: uint64(time.Now().Unix()),
+			DueTime:    uint64(time.Now().Unix()) + 30*24*3600,
+			Check:      false,
+			Reward:     false,
+		}
+		//db.InsertModelLogs("mail", "uid123", _mail)
+		//InsertModelLogs("mail", "uid123", _mail)
+		data := &comm.Autogenerated{
+			ID:  primitive.NewObjectID().Hex(),
+			UID: "uid123",
+			Act: string(comm.LogHandleType_Insert),
+		}
+		data.D = append(data.D, "mail") // D[0]
+		data.D = append(data.D, _mail)  // D[1]
+
+		_, err1 := db.Defsys.Mgo().InsertOne("model", data)
+		if err1 != nil {
+			log.Errorf("insert model db err %v", err1)
+		}
+		//}()
+	}
+	time.Sleep(time.Second * 10)
 	defer os.Exit(m.Run())
 
 }