component.go 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172
  1. package mhayaMongo
  2. import (
  3. "context"
  4. "crypto/tls"
  5. "fmt"
  6. "time"
  7. cfacade "github.com/mhaya/facade"
  8. clog "github.com/mhaya/logger"
  9. cprofile "github.com/mhaya/profile"
  10. "go.mongodb.org/mongo-driver/mongo"
  11. "go.mongodb.org/mongo-driver/mongo/options"
  12. "go.mongodb.org/mongo-driver/mongo/readpref"
  13. )
  14. const (
  15. Name = "mongo_component"
  16. )
  17. type (
  18. Component struct {
  19. cfacade.Component
  20. dbMap map[string]map[string]*mongo.Database
  21. }
  22. // HashDb hash by group id
  23. HashDb func(dbMaps map[string]*mongo.Database) string
  24. )
  25. func NewComponent() *Component {
  26. return &Component{
  27. dbMap: make(map[string]map[string]*mongo.Database),
  28. }
  29. }
  30. func (*Component) Name() string {
  31. return Name
  32. }
  33. func (s *Component) Init() {
  34. // load only the database contained in the `db_id_list`
  35. mongoIdList := s.App().Settings().Get("db_id_list")
  36. if mongoIdList.LastError() != nil || mongoIdList.Size() < 1 {
  37. clog.Warnf("[nodeId = %s] `mongo_id_list` property not exists.", s.App().NodeId())
  38. return
  39. }
  40. mongoConfig := cprofile.GetConfig("mongo")
  41. if mongoConfig.LastError() != nil {
  42. panic("`mongo` property not exists in profile file.")
  43. }
  44. for _, groupId := range mongoConfig.Keys() {
  45. s.dbMap[groupId] = make(map[string]*mongo.Database)
  46. dbGroup := mongoConfig.GetConfig(groupId)
  47. for i := 0; i < dbGroup.Size(); i++ {
  48. item := dbGroup.GetConfig(i)
  49. var (
  50. enable = item.GetBool("enable", true)
  51. id = item.GetString("db_id")
  52. dbName = item.GetString("db_name")
  53. uri = item.GetString("uri")
  54. timeout = time.Duration(item.GetInt64("timeout", 3)) * time.Second
  55. tlsEnable = item.GetInt("tls")
  56. maxClient = item.GetInt("maxClient")
  57. )
  58. for _, key := range mongoIdList.Keys() {
  59. dbId := mongoIdList.Get(key).ToString()
  60. if id != dbId {
  61. continue
  62. }
  63. if !enable {
  64. panic(fmt.Sprintf("[dbName = %s] is disabled!", dbName))
  65. }
  66. db, err := CreateDatabase(uri, dbName, tlsEnable, uint64(maxClient), timeout)
  67. if err != nil {
  68. panic(fmt.Sprintf("[dbName = %s] create mongodb fail. error = %s", dbName, err))
  69. }
  70. s.dbMap[groupId][id] = db
  71. clog.Infof("[dbGroup =%s, dbName = %s] is connected.", groupId, id)
  72. }
  73. }
  74. }
  75. }
  76. func CreateDatabase(uri, dbName string, tlsEnable int, maxClient uint64, timeout ...time.Duration) (*mongo.Database, error) {
  77. tt := 5 * time.Second
  78. if len(timeout) > 0 && timeout[0].Seconds() > 3 {
  79. tt = timeout[0]
  80. }
  81. var o *options.ClientOptions
  82. if tlsEnable == 1 {
  83. tlsConfig := &tls.Config{
  84. MinVersion: tls.VersionTLS12,
  85. PreferServerCipherSuites: true,
  86. }
  87. o = options.Client().ApplyURI(uri).SetMaxPoolSize(500). //最大连接
  88. SetMinPoolSize(20). //最小连接
  89. SetMaxConnIdleTime(3 * time.Minute). //连接空闲时间
  90. SetMaxConnIdleTime(5 * time.Minute). //连接生命周期
  91. SetConnectTimeout(10 * time.Second). //连接超时时间
  92. SetSocketTimeout(10 * time.Second).SetTLSConfig(tlsConfig) //套接字超时时间
  93. } else {
  94. o = options.Client().ApplyURI(uri).SetMaxPoolSize(500). //最大连接
  95. SetMinPoolSize(20). //最小连接
  96. SetMaxConnIdleTime(3 * time.Minute). //连接空闲时间
  97. SetMaxConnIdleTime(5 * time.Minute). //连接生命周期
  98. SetConnectTimeout(10 * time.Second). //连接超时时间
  99. SetSocketTimeout(10 * time.Second)
  100. }
  101. if err := o.Validate(); err != nil {
  102. return nil, err
  103. }
  104. ctx, cancel := context.WithTimeout(context.Background(), tt)
  105. defer cancel()
  106. client, err := mongo.Connect(ctx, o)
  107. if err != nil {
  108. return nil, err
  109. }
  110. err = client.Ping(context.Background(), readpref.Primary())
  111. if err != nil {
  112. return nil, err
  113. }
  114. clog.Infof("ping db [uri = %s] is ok", uri)
  115. return client.Database(dbName), nil
  116. }
  117. func (s *Component) GetDb(id string) *mongo.Database {
  118. for _, group := range s.dbMap {
  119. for k, v := range group {
  120. if k == id {
  121. return v
  122. }
  123. }
  124. }
  125. return nil
  126. }
  127. func (s *Component) GetHashDb(groupId string, hashFn HashDb) (*mongo.Database, bool) {
  128. dbGroup, found := s.GetDbMap(groupId)
  129. if !found {
  130. clog.Warnf("groupId = %s not found.", groupId)
  131. return nil, false
  132. }
  133. dbId := hashFn(dbGroup)
  134. db, found := dbGroup[dbId]
  135. return db, found
  136. }
  137. func (s *Component) GetDbMap(groupId string) (map[string]*mongo.Database, bool) {
  138. dbGroup, found := s.dbMap[groupId]
  139. return dbGroup, found
  140. }