ShTcp.php 4.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123
  1. <?php
  2. namespace app\workerman;
  3. use app\common\server\ShOneanalysis;
  4. use think\facade\Log;
  5. use think\worker\Server;
  6. use Workerman\Lib\Timer;
  7. class ShTcp extends Server
  8. {
  9. protected $socket = 'tcp://0.0.0.0:21444';
  10. //protected $socket = 'tcp://0.0.0.0:9501';
  11. protected function init()
  12. {
  13. $this->count = 4; //线程数
  14. }
  15. public function onConnect($connection)
  16. {
  17. }
  18. public function onWorkerStart($worker)
  19. {
  20. define('HEARTBEAT_TIME', 1000);
  21. //异步处理
  22. //发送
  23. $server_sa = new ShOneanalysis();
  24. Timer::add(1, function () use ($worker, $server_sa) {
  25. if (isset($worker->device_id_code)) {
  26. $device_id_code = $worker->device_id_code;
  27. foreach ($worker->connections as $connection) {
  28. $info = $server_sa->getsendqueuelist($device_id_code);
  29. $totalpage = $info['totalpage'];
  30. if ($totalpage > 0) {
  31. $size = $info['size'];
  32. for ($page = 1; $page <= $totalpage; $page++) {
  33. $list = $server_sa->getsendqueuelist($device_id_code, 1, $size);
  34. foreach ($list as $key => $value) {
  35. $connection->send($value['msg']);
  36. $server_sa->showfortest($value['msg']);
  37. $server_sa->sendmsgsuccess($value);
  38. }
  39. }
  40. }
  41. }
  42. }
  43. });
  44. //解析
  45. $server_sa->initanalysisShoneReceiveMsg();
  46. Timer::add(10, function () use ($worker, $server_sa) {
  47. $page = 1;
  48. $size = 30;
  49. $list = $server_sa->getreceivequeuelist($page, $size);
  50. if ($list) {
  51. $list = (array) $list;
  52. foreach ($list as $key => $value) {
  53. try {
  54. $linedata = $server_sa->getline($value['msg']);
  55. //长度相等可解析
  56. $content = $linedata['content_str'];
  57. $device_id_code = $linedata['device_id_code'];
  58. $fid = $server_sa->getfacilityid($device_id_code);
  59. $content = $server_sa->getcommandcontent($content);
  60. $result = $server_sa->saveshinfo($fid, $device_id_code, $content, $value['msg']);
  61. if ($result) {
  62. $command = $content[0];
  63. $data = $server_sa->createbacksendmsg($fid, $device_id_code, $content, $command);
  64. }
  65. $server_sa->receivemsgsuccess($value);
  66. } catch (\Exception $e) {
  67. $msg = $e->getMessage();
  68. echo $msg . "\n";
  69. Log::write($msg, 'shouhuan');
  70. }
  71. }
  72. }
  73. });
  74. Timer::add(1, function () use ($worker) {
  75. $time_now = time();
  76. foreach ($worker->connections as $connection) {
  77. // 有可能该connection还没收到过消息,则lastMessageTime设置为当前时间
  78. if (empty($connection->lastMessageTime)) {
  79. $connection->lastMessageTime = $time_now;
  80. continue;
  81. }
  82. // 上次通讯时间间隔大于心跳间隔,则认为客户端已经下线,关闭连接
  83. if ($time_now - $connection->lastMessageTime > HEARTBEAT_TIME) {
  84. //$connection->close();
  85. //设置对应设备下线
  86. }
  87. }
  88. });
  89. }
  90. public function onMessage($connection, $data)
  91. {
  92. //Log::write($data, 'shouhuan');
  93. $server_sa = new ShOneanalysis();
  94. $datas = (array) $server_sa->unpack($data);
  95. foreach ($datas as $key => $data) {
  96. $server_sa->showfortest($data);
  97. try {
  98. $result = $server_sa->savereceivequeue($data);
  99. if (empty($result)) {
  100. Log::write("手环接收信息队列,数据保存失败", 'shouhuan');
  101. }
  102. $line = $server_sa->getline($data);
  103. $this->device_id_code = $line['device_id_code'];
  104. //$this->device_id_code = "358800006072996";
  105. } catch (\Exception $e) {
  106. Log::write($data, 'shouhuan');
  107. Log::write($e->getMessage(), 'shouhuan');
  108. var_dump($e->getMessage());
  109. }
  110. }
  111. }
  112. public function onClose($connection)
  113. {
  114. }
  115. }