DataCollectionManager.cs 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445
  1. using System;
  2. using System.Collections.Generic;
  3. using System.Linq;
  4. using System.Text;
  5. using Aitex.Core.Util;
  6. using Npgsql;
  7. using System.Threading;
  8. using Aitex.Core.RT.Log;
  9. using Aitex.Core.RT.Event;
  10. using Aitex.Core.RT.DBCore;
  11. using System.IO;
  12. using Aitex.Common.Util;
  13. using Aitex.Core.RT.DataCenter;
  14. using System.Collections.Concurrent;
  15. using MECF.Framework.Common.Communications;
  16. using Aitex.Core.RT.SCCore;
  17. namespace Aitex.Core.RT.DataCollection
  18. {
  19. public class DataCollectionManager : Singleton<DataCollectionManager>, IConnection
  20. {
  21. public string Address
  22. {
  23. get { return "DB:DataCollection"; }
  24. }
  25. public bool IsConnected { get; set; }
  26. NpgsqlConnection _conn;
  27. bool _bAlive = true;
  28. Dictionary<string, Func<object>> _preSubscribedRecordedData; //for other thread to subscribe data
  29. Dictionary<string, Func<object>> _subscribedRecordedData; //internal use
  30. object _lock = new object();
  31. F_TRIG _connFailTrig = new F_TRIG();
  32. private int dataCollectionInterval;
  33. IDataCollectionCallback _callback;
  34. private string[] _modules = new string[] { "Data" };
  35. public DataCollectionManager()
  36. {
  37. _preSubscribedRecordedData = new Dictionary<string, Func<object>>();
  38. _subscribedRecordedData = new Dictionary<string, Func<object>>();
  39. dataCollectionInterval = SC.GetValue<int>("System.DataCollectionInterval");
  40. }
  41. public void Initialize(IDataCollectionCallback callback)
  42. {
  43. _callback = callback == null ? new DefaultDataCollectionCallback() : callback;
  44. Connect();
  45. System.Threading.Tasks.Task.Factory.StartNew(DataRecorderThread);
  46. }
  47. public void Initialize(string[] modules, string dbName)
  48. {
  49. _callback = new DefaultDataCollectionCallback(dbName);
  50. if (modules != null && modules.Length > 0 && !string.IsNullOrEmpty(modules[0]))
  51. _modules = modules;
  52. Connect();
  53. System.Threading.Tasks.Task.Factory.StartNew(DataRecorderThread);
  54. }
  55. public void Initialize(string dbName)
  56. {
  57. _callback = new DefaultDataCollectionCallback(dbName);
  58. Connect();
  59. System.Threading.Tasks.Task.Factory.StartNew(DataRecorderThread);
  60. }
  61. public void Terminate()
  62. {
  63. _bAlive = false;
  64. }
  65. /// <summary>
  66. /// initialization
  67. /// </summary>
  68. public void SubscribeData(string dataName, string alias, Func<object> dataValue)
  69. {
  70. lock (_lock)
  71. {
  72. if (!_preSubscribedRecordedData.Keys.Contains(dataName))
  73. {
  74. _preSubscribedRecordedData[dataName] = dataValue;
  75. }
  76. }
  77. }
  78. private void MonitorDataCenter()
  79. {
  80. lock (_lock)
  81. {
  82. SortedDictionary<string, Func<object>> data = DATA.GetDBRecorderList();
  83. foreach (var item in data)
  84. {
  85. if (!_subscribedRecordedData.ContainsKey(item.Key))
  86. {
  87. object o = item.Value.Invoke();
  88. if (o == null)
  89. continue;
  90. Type dataType = o.GetType();
  91. if (dataType == typeof(Boolean) || dataType == typeof(double) || dataType == typeof(float) ||
  92. dataType == typeof(int) || dataType == typeof(ushort) || dataType == typeof(short))
  93. {
  94. _subscribedRecordedData[item.Key] = item.Value;
  95. _preSubscribedRecordedData[item.Key] = item.Value;
  96. }
  97. }
  98. }
  99. }
  100. }
  101. public bool Connect()
  102. {
  103. bool result = true;
  104. try
  105. {
  106. if (_conn != null)
  107. _conn.Close();
  108. _conn = new NpgsqlConnection(PostgresqlHelper.ConnectionString);
  109. _conn.Open();
  110. _conn.ChangeDatabase(_callback.GetDBName());
  111. EV.PostInfoLog("DataCollection", "Connected with database");
  112. //Invoke operation
  113. //EV.PostMessage("DataCollection", EventEnum.da, "Connected with database");
  114. }
  115. catch (Exception ex)
  116. {
  117. if (_conn != null)
  118. {
  119. _conn.Close();
  120. _conn = null;
  121. }
  122. result = false;
  123. EV.PostInfoLog("DataCollection", "Can not connect database");
  124. LOG.WriteExeption(ex);
  125. }
  126. IsConnected = result;
  127. return result;
  128. }
  129. public bool Disconnect()
  130. {
  131. try
  132. {
  133. if (_conn != null)
  134. {
  135. _conn.Close();
  136. _conn = null;
  137. EV.PostInfoLog("DataCollection", "Disconnected with database");
  138. }
  139. }
  140. catch (Exception ex)
  141. {
  142. LOG.WriteExeption(ex);
  143. }
  144. IsConnected = false;
  145. return true;
  146. }
  147. /// <summary>
  148. /// 重置数据库连接出错事件
  149. /// </summary>
  150. public void Reset()
  151. {
  152. _connFailTrig.CLK = false;
  153. }
  154. /// <summary>
  155. /// data recorder thread
  156. /// </summary>
  157. void DataRecorderThread()
  158. {
  159. while (_bAlive)
  160. {
  161. try
  162. {
  163. Thread.Sleep(dataCollectionInterval);
  164. _connFailTrig.CLK = IsConnected;// Connect();
  165. if (_connFailTrig.Q)
  166. {
  167. EV.PostWarningLog("DataCollection", "Can not connect with database");
  168. }
  169. if (!_connFailTrig.M)
  170. continue;
  171. MonitorDataCenter();
  172. var moduleDataItem = new Dictionary<string, Dictionary<string, Func<object>>>();
  173. var moduleInsertSql = new Dictionary<string, string>();
  174. foreach (var module in _modules)
  175. {
  176. moduleDataItem[module] = new Dictionary<string, Func<object>>();
  177. }
  178. string defaultModule = _modules.Contains("System") ? "System" : (_modules.Contains("Data") ? "Data" : (_modules[0]));
  179. if (_subscribedRecordedData.Count > 0)
  180. {
  181. lock (_lock)
  182. {
  183. bool foundModule;
  184. foreach (var dataName in _subscribedRecordedData.Keys)
  185. {
  186. foundModule = false;
  187. foreach (var module in _modules)
  188. {
  189. if (dataName.StartsWith(module + "."))
  190. {
  191. moduleDataItem[module][dataName] = _subscribedRecordedData[dataName];
  192. foundModule = true;
  193. break;
  194. }
  195. }
  196. if (!foundModule)
  197. {
  198. moduleDataItem[defaultModule][dataName] = _subscribedRecordedData[dataName];
  199. }
  200. }
  201. _preSubscribedRecordedData.Clear();
  202. }
  203. }
  204. DateTime dtToday = DateTime.Now.Date;
  205. foreach (var module in _modules)
  206. {
  207. string tableName = $"{dtToday:yyyyMMdd}.{module}";
  208. UpdateTableSchema(tableName, moduleDataItem[module]);
  209. string preCreatedInsertSQL = string.Format("INSERT INTO \"{0}\"(\"time\" ", tableName);
  210. foreach (var dataName in moduleDataItem[module].Keys)
  211. {
  212. preCreatedInsertSQL += string.Format(",\"{0}\"", dataName);
  213. }
  214. preCreatedInsertSQL += ")";
  215. moduleInsertSql[module] = preCreatedInsertSQL;
  216. }
  217. //insert data into database
  218. StringBuilder sb = new StringBuilder(10000);
  219. while (_bAlive)
  220. {
  221. //Thread.Sleep(990);
  222. Thread.Sleep((int)(dataCollectionInterval * 0.99)); //for time delay in sleep function
  223. foreach (var module in _modules)
  224. {
  225. sb.Remove(0, sb.Length);
  226. sb.Append("Values(");
  227. sb.Append(DateTime.Now.Ticks.ToString());
  228. foreach (var dataName in moduleDataItem[module].Keys)
  229. {
  230. sb.Append(",");
  231. var v1 = moduleDataItem[module][dataName].Invoke();
  232. if (v1 == null)
  233. {
  234. sb.Append("0");
  235. }
  236. else
  237. {
  238. if (v1 is double || v1 is float)
  239. {
  240. double v2 = Convert.ToDouble(v1);
  241. if (double.IsNaN(v2))
  242. v2 = 0;
  243. sb.Append(v2.ToString());
  244. }
  245. else
  246. {
  247. sb.Append(v1.ToString());
  248. }
  249. }
  250. }
  251. sb.Append(");");
  252. try
  253. {
  254. if (_conn.State == System.Data.ConnectionState.Open)
  255. {
  256. NpgsqlCommand cmd = new NpgsqlCommand(moduleInsertSql[module] + sb.ToString(), _conn);
  257. cmd.ExecuteNonQuery();
  258. }
  259. else
  260. {
  261. Connect();
  262. }
  263. }
  264. catch
  265. {
  266. Connect();
  267. }
  268. //try
  269. //{
  270. //}
  271. //catch (Exception ex)
  272. //{
  273. // LOG.WriteExeption("数据记录发生异常" + moduleInsertSql[module], ex);
  274. //}
  275. }
  276. //if alert to another day, create a new table
  277. if (DateTime.Now.Date != dtToday)
  278. break;
  279. //if preSubscribed data not empty, alert current table's structure
  280. MonitorDataCenter();
  281. if (_preSubscribedRecordedData.Count > 0)
  282. break;
  283. }//end while
  284. }//end while
  285. catch (Exception ex)
  286. {
  287. LOG.WriteExeption("数据库操作记录发生异常", ex);
  288. }
  289. }
  290. }
  291. private string UpdateTableSchema(string tblName, Dictionary<string, Func<object>> dataItem)
  292. {
  293. //query if table already exist?
  294. string sqlTblDefine = string.Format("select column_name from information_schema.columns where table_name = '{0}';", tblName);
  295. NpgsqlCommand cmdTblDefine = new NpgsqlCommand(sqlTblDefine, _conn);
  296. var tblDefineData = cmdTblDefine.ExecuteReader();
  297. string tblAlertString = string.Empty;
  298. List<string> colNameList = new List<string>();
  299. while (tblDefineData.Read())
  300. {
  301. for (int i = 0; i < tblDefineData.FieldCount; i++)
  302. colNameList.Add(tblDefineData[i].ToString());
  303. }
  304. tblDefineData.Close();
  305. if (colNameList.Count > 0)
  306. {
  307. //table exist
  308. foreach (var dataName in dataItem.Keys)
  309. {
  310. if (!colNameList.Contains(dataName))
  311. {
  312. var dataType = dataItem[dataName].Invoke().GetType();
  313. if (dataType == typeof(Boolean))
  314. {
  315. tblAlertString += string.Format("ALTER TABLE \"{0}\" ADD COLUMN \"{1}\" {2};", tblName, dataName, "Boolean");
  316. }
  317. else if (dataType == typeof(double) || dataType == typeof(float) || dataType == typeof(int) || dataType == typeof(ushort) || dataType == typeof(short))
  318. {
  319. tblAlertString += string.Format("ALTER TABLE \"{0}\" ADD COLUMN \"{1}\" {2};", tblName, dataName, "Real");
  320. }
  321. else
  322. {
  323. }
  324. }
  325. }
  326. if (!string.IsNullOrEmpty(tblAlertString))
  327. {
  328. try
  329. {
  330. NpgsqlCommand alertTblCmd = new NpgsqlCommand(tblAlertString, _conn);
  331. alertTblCmd.ExecuteNonQuery();
  332. }
  333. catch (Exception ex)
  334. {
  335. LOG.WriteExeption(ex);
  336. }
  337. }
  338. }
  339. else
  340. {
  341. //table not exist, auto generate a table
  342. string createTble = string.Format("CREATE TABLE \"{0}\"(Time bigint NOT NULL,", tblName);
  343. string commentStr = "";
  344. foreach (var dataName in dataItem.Keys)
  345. {
  346. Type dataType = dataItem[dataName].Invoke().GetType();
  347. if (dataType == typeof(Boolean))
  348. {
  349. createTble += string.Format("\"{0}\" Boolean,\n", dataName);
  350. }
  351. else if (dataType == typeof(double) || dataType == typeof(float) ||
  352. dataType == typeof(int) || dataType == typeof(ushort) || dataType == typeof(short))
  353. {
  354. createTble += string.Format("\"{0}\" Real,\n", dataName);
  355. }
  356. //if (!string.IsNullOrEmpty(alias) && ((dataType == typeof(Boolean)) || (dataType == typeof(double) || dataType == typeof(float) ||
  357. // dataType == typeof(int) || dataType == typeof(ushort) || dataType == typeof(short))))
  358. // commentStr += string.Format("COMMENT ON COLUMN \"{0}\".\"{1}\" IS '{2}';\n", tblName, dataName, alias);
  359. }
  360. createTble += string.Format("CONSTRAINT \"{0}_pkey\" PRIMARY KEY (Time));", tblName);
  361. createTble += string.Format("GRANT SELECT ON TABLE \"{0}\" TO postgres;", tblName);
  362. // createTble += string.Format("CREATE INDEX \"{0}_time_idx\" ON \"{0}\" USING btree (\"time\" );", tblName);
  363. createTble += commentStr;
  364. try
  365. {
  366. NpgsqlCommand cmd = new NpgsqlCommand(createTble.ToString(), _conn);
  367. cmd.ExecuteNonQuery();
  368. }
  369. catch (Exception ex1)
  370. {
  371. LOG.WriteExeption("创建数据库表格失败", ex1);
  372. throw;
  373. }
  374. }
  375. return tblName;
  376. }
  377. }
  378. }