修复在线用户并发重复记录问题

app/Library/Utils/Lock.php

@@ -0,0 +1,79 @@
+<?php
+
+namespace App\Library\Utils;
+
+use App\Library\Cache\Backend\Redis as RedisCache;
+use Phalcon\Di;
+use Phalcon\Text;
+
+class Lock
+{
+
+    /**
+     * @param string $itemId
+     * @param int $expire
+     * @return bool|string
+     */
+    public static function addLock($itemId, $expire = 600)
+    {
+        if (!$itemId || $expire <= 0) {
+            return false;
+        }
+
+        /**
+         * @var RedisCache $cache
+         */
+        $cache = Di::getDefault()->getShared('cache');
+
+        $redis = $cache->getRedis();
+
+        $lockId = Text::random(Text::RANDOM_ALNUM, 16);
+
+        $keyName = self::getLockKey($itemId);
+
+        $result = $redis->set($keyName, $lockId, ['nx', 'ex' => $expire]);
+
+        return $result ? $lockId : false;
+    }
+
+    /**
+     * @param string $itemId
+     * @param string $lockId
+     * @return bool
+     */
+    public static function releaseLock($itemId, $lockId)
+    {
+        if (!$itemId || !$lockId) {
+            return false;
+        }
+
+        /**
+         * @var RedisCache $cache
+         */
+        $cache = Di::getDefault()->getShared('cache');
+
+        $redis = $cache->getRedis();
+
+        $keyName = self::getLockKey($itemId);
+
+        $redis->watch($keyName);
+
+        /**
+         * 监听key防止被修改或删除，提交事务后会自动取消监控，其他情况需手动解除监控
+         */
+        if ($lockId == $redis->get($keyName)) {
+            $redis->multi()->del($keyName)->exec();
+            return true;
+        }
+
+        $redis->unwatch();
+
+        return false;
+    }
+
+    public static function getLockKey($itemId)
+    {
+        return sprintf('kg_lock:%s', $itemId);
+    }
+
+}

app/Listeners/User.php

@@ -2,6 +2,7 @@
 
 namespace App\Listeners;
 
+use App\Library\Utils\Lock as LockUtil;
 use App\Models\Online as OnlineModel;
 use App\Models\User as UserModel;
 use App\Repos\Online as OnlineRepo;
@@ -15,7 +16,13 @@ class User extends Listener
 
     public function online(Event $event, $source, UserModel $user)
     {
+        $itemId = "user:{$user->id}";
+
+        $lockId = LockUtil::addLock($itemId);
+
         $now = time();
+        $clientType = $this->getClientType();
+        $clientIp = $this->getClientIp();
 
         if ($now - $user->active_time > 600) {
 
@@ -25,28 +32,46 @@ class User extends Listener
 
             $onlineRepo = new OnlineRepo();
 
-            $online = $onlineRepo->findByUserDate($user->id, date('Y-m-d'));
+            $records = $onlineRepo->findByUserDate($user->id, date('Y-m-d'));
+
+            if ($records->count() > 0) {
+
+                $online = null;
+
+                foreach ($records as $record) {
+                    if ($record->client_type == $clientType && $record->client_ip == $clientIp) {
+                        $online = $record;
+                        break;
+                    }
+                }
 
                 if ($online) {
-
                     $online->active_time = $now;
-                $online->client_type = $this->getClientType();
-                $online->client_ip = $this->getClientIp();
-
                     $online->update();
+                } else {
+                    $this->createOnline($user->id, $clientType, $clientIp);
+                }
 
             } else {
+                $this->createOnline($user->id, $clientType, $clientIp);
+            }
+        }
 
+        LockUtil::releaseLock($itemId, $lockId);
+    }
+
+    protected function createOnline($userId, $clientType, $clientIp)
+    {
         $online = new OnlineModel();
 
-                $online->user_id = $user->id;
-                $online->active_time = $now;
-                $online->client_type = $this->getClientType();
-                $online->client_ip = $this->getClientIp();
+        $online->user_id = $userId;
+        $online->client_type = $clientType;
+        $online->client_ip = $clientIp;
+        $online->active_time = time();
 
         $online->create();
-            }
-        }
+
+        return $online;
     }
 
 }

app/Repos/Online.php

@@ -3,7 +3,8 @@
 namespace App\Repos;
 
 use App\Models\Online as OnlineModel;
-use Phalcon\Mvc\Model;
+use Phalcon\Mvc\Model\Resultset;
+use Phalcon\Mvc\Model\ResultsetInterface;
 
 class Online extends Repository
 {
@@ -11,20 +12,18 @@ class Online extends Repository
     /**
      * @param int $userId
      * @param string $activeDate
-     * @return OnlineModel|Model|bool
+     * @return ResultsetInterface|Resultset|OnlineModel[]
      */
     public function findByUserDate($userId, $activeDate)
     {
-        $activeTime = strtotime($activeDate);
+        $startTime = strtotime($activeDate);
 
-        return OnlineModel::findFirst([
-            'conditions' => 'user_id = ?1 AND active_time BETWEEN ?2 AND ?3',
-            'bind' => [
-                1 => $userId,
-                2 => $activeTime,
-                3 => $activeTime + 86400,
-            ],
-        ]);
+        $endTime = $startTime + 86400;
+
+        return OnlineModel::query()
+            ->where('user_id = :user_id:', ['user_id' => $userId])
+            ->betweenWhere('active_time', $startTime, $endTime)
+            ->execute();
     }
 
 }