ShTcp.php 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121
  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->sendmsgsuccess($value);
  37. }
  38. }
  39. }
  40. }
  41. }
  42. });
  43. //解析
  44. $server_sa->initanalysisShoneReceiveMsg();
  45. Timer::add(10, function () use ($worker, $server_sa) {
  46. $page = 1;
  47. $size = 30;
  48. $list = $server_sa->getreceivequeuelist($page, $size);
  49. if ($list) {
  50. $list = (array) $list;
  51. foreach ($list as $key => $value) {
  52. try {
  53. $linedata = $server_sa->getline($value['msg']);
  54. //长度相等可解析
  55. $content = $linedata['content_str'];
  56. $device_id_code = $linedata['device_id_code'];
  57. $fid = $server_sa->getfacilityid($device_id_code);
  58. $content = $server_sa->getcommandcontent($content);
  59. $result = $server_sa->saveshinfo($fid, $device_id_code, $content, $value['msg']);
  60. if ($result) {
  61. $command = $content[0];
  62. $data = $server_sa->createbacksendmsg($fid, $device_id_code, $content, $command);
  63. }
  64. $server_sa->receivemsgsuccess($value);
  65. } catch (\Exception $e) {
  66. $msg = $e->getMessage();
  67. echo $msg . "\n";
  68. Log::write($msg, 'shouhuan');
  69. }
  70. }
  71. }
  72. });
  73. Timer::add(1, function () use ($worker) {
  74. $time_now = time();
  75. foreach ($worker->connections as $connection) {
  76. // 有可能该connection还没收到过消息,则lastMessageTime设置为当前时间
  77. if (empty($connection->lastMessageTime)) {
  78. $connection->lastMessageTime = $time_now;
  79. continue;
  80. }
  81. // 上次通讯时间间隔大于心跳间隔,则认为客户端已经下线,关闭连接
  82. if ($time_now - $connection->lastMessageTime > HEARTBEAT_TIME) {
  83. //$connection->close();
  84. //设置对应设备下线
  85. }
  86. }
  87. });
  88. }
  89. public function onMessage($connection, $data)
  90. {
  91. Log::write($data, 'shouhuan');
  92. $server_sa = new ShOneanalysis();
  93. $datas = (array) $server_sa->unpack($data);
  94. foreach ($datas as $key => $data) {
  95. try {
  96. $result = $server_sa->savereceivequeue($data);
  97. if (empty($result)) {
  98. Log::write("手环接收信息队列,数据保存失败", 'shouhuan');
  99. }
  100. $line = $server_sa->getline($data);
  101. $this->device_id_code = $line['device_id_code'];
  102. //$this->device_id_code = "358800006072996";
  103. } catch (\Exception $e) {
  104. Log::write($data, 'shouhuan');
  105. Log::write($e->getMessage(), 'shouhuan');
  106. var_dump($e->getMessage());
  107. }
  108. }
  109. }
  110. public function onClose($connection)
  111. {
  112. }
  113. }