DbConnectionManager.php 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117
  1. <?php
  2. class LtDbConnectionManager
  3. {
  4. /**
  5. * Connection management
  6. * array(
  7. * "connection" => connection resource id,
  8. * "expire_time" => expire time,
  9. * "schema" => default schema name,
  10. * "charset" => char set / encoding
  11. * )
  12. */
  13. static public $connectionPool;
  14. public $configHandle;
  15. protected $connectionAdapter;
  16. protected $sqlAdapter;
  17. private $servers;
  18. public function getConnection($group, $node, $role = "master")
  19. {
  20. if(empty($this->servers))
  21. {
  22. $this->servers = $this->configHandle->get("db.servers");
  23. }
  24. if (($connection = $this->getNewConnection($group, $node, $role)) ||($connection = $this->getCachedConnection($group, $node, $role)))
  25. {
  26. return array(
  27. "connectionAdapter" => $this->connectionAdapter,
  28. "connectionResource" => $connection
  29. );
  30. }
  31. else
  32. {
  33. trigger_error("db server can not be connected: group=$group, node=$node, role=$role", E_USER_ERROR);
  34. return false;
  35. }
  36. }
  37. protected function getConnectionKey($connConf)
  38. {
  39. return $connConf['adapter'] . $connConf['host'] . $connConf['port'] . $connConf['username'] . $connConf['dbname'];
  40. }
  41. protected function saveConnection($connConf, $connection, $ttl)
  42. {
  43. $connectionInfo = array(
  44. "connection" => $connection,
  45. "expire_time" => time() + $ttl,
  46. "schema" => $connConf["schema"],
  47. "charset" => $connConf["charset"],
  48. );
  49. self::$connectionPool[$this->getConnectionKey($connConf)] = $connectionInfo;
  50. }
  51. protected function getCachedConnection($group, $node, $role)
  52. {
  53. foreach($this->servers[$group][$node][$role] as $hostConfig)
  54. {
  55. $key = $this->getConnectionKey($hostConfig);
  56. if(isset(self::$connectionPool[$key]) && time() < self::$connectionPool[$key]['expire_time'])
  57. {//cached connection resource FOUND
  58. $connectionInfo = self::$connectionPool[$key];
  59. if ($connectionInfo["schema"] != $hostConfig["schema"] || $connectionInfo["charset"] != $hostConfig["charset"])
  60. {//检查当前schema和charset与用户要操作的目标不一致
  61. $hostConfig = $this->servers[$group][$node][$role][$hostIndexArray[$hashNumber]];
  62. $dbFactory = new LtDbAdapterFactory;
  63. $this->connectionAdapter = $dbFactory->getConnectionAdapter($hostConfig["connection_adapter"]);
  64. $this->sqlAdapter = $dbFactory->getSqlAdapter($hostConfig["sql_adapter"]);
  65. if ($connectionInfo["schema"] != $hostConfig["schema"])
  66. {
  67. $this->connectionAdapter->exec($this->sqlAdapter->setSchema($hostConfig["schema"]), $connectionInfo["connection"]);
  68. }
  69. if ($connectionInfo["charset"] != $hostConfig["charset"])
  70. {
  71. $this->connectionAdapter->exec($this->sqlAdapter->setCharset($hostConfig["charset"]), $connectionInfo["connection"]);
  72. }
  73. $this->saveConnection($hostConfig, $connectionInfo["connection"], $hostConfig["connection_ttl"]);
  74. }
  75. return $connectionInfo["connection"];
  76. }
  77. }
  78. return false;
  79. }
  80. protected function getNewConnection($group, $node, $role)
  81. {
  82. $hostTotal = count($this->servers[$group][$node][$role]);
  83. $hostIndexArray = array_keys($this->servers[$group][$node][$role]);
  84. while ($hostTotal)
  85. {
  86. $hashNumber = substr(microtime(),7,1) % $hostTotal;
  87. $hostConfig = $this->servers[$group][$node][$role][$hostIndexArray[$hashNumber]];
  88. $dbFactory = new LtDbAdapterFactory;
  89. $this->connectionAdapter = $dbFactory->getConnectionAdapter($hostConfig["connection_adapter"]);
  90. $this->sqlAdapter = $dbFactory->getSqlAdapter($hostConfig["sql_adapter"]);
  91. if ($connection = $this->connectionAdapter->connect($hostConfig))
  92. {
  93. $this->connectionAdapter->exec($this->sqlAdapter->setSchema($hostConfig["schema"]), $connection);
  94. $this->connectionAdapter->exec($this->sqlAdapter->setCharset($hostConfig["charset"]), $connection);
  95. $this->saveConnection($hostConfig, $connection, $hostConfig["connection_ttl"]);
  96. return $connection;
  97. }
  98. else
  99. {
  100. //trigger_error('connection fail', E_USER_WARNING);
  101. //delete the unavailable server
  102. for ($i = $hashNumber; $i < $hostTotal - 1; $i ++)
  103. {
  104. $hostIndexArray[$i] = $hostIndexArray[$i+1];
  105. }
  106. unset($hostIndexArray[$hostTotal-1]);
  107. $hostTotal --;
  108. }//end else
  109. }//end while
  110. return false;
  111. }
  112. }