PV UV 采集代码

PV埋点异步处理方案0519-2.md

+
+
+
+
+          
+## 小程序 PV/UV 统计完整方案（队列异步实现）
+
+### 一、数据库表设计
+
+```sql
+-- 1. 埋点数据表
+CREATE TABLE `fa_shopro_miniprogram_track` (
+    `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
+    `user_id` int(11) DEFAULT '0' COMMENT '用户ID',
+    `openid` varchar(128) DEFAULT '' COMMENT '微信openid',
+    `session_id` varchar(64) DEFAULT '' COMMENT '会话ID',
+    `event_type` varchar(32) NOT NULL COMMENT '事件类型',
+    `page_path` varchar(255) DEFAULT '' COMMENT '页面路径',
+    `page_title` varchar(100) DEFAULT '' COMMENT '页面标题',
+    `params` text COMMENT '自定义参数',
+    `ip` varchar(50) DEFAULT '' COMMENT '用户IP',
+    `create_time` int(11) NOT NULL COMMENT '创建时间',
+    PRIMARY KEY (`id`),
+    KEY `session_id` (`session_id`),
+    KEY `event_type` (`event_type`),
+    KEY `create_time` (`create_time`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='小程序行为埋点表';
+
+-- 2. 每日统计汇总表
+CREATE TABLE `fa_shopro_miniprogram_stats` (
+    `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
+    `date` date NOT NULL COMMENT '统计日期',
+    `uv` int(11) DEFAULT '0' COMMENT '独立访客数',
+    `pv` int(11) DEFAULT '0' COMMENT '页面浏览量',
+    `new_uv` int(11) DEFAULT '0' COMMENT '新访客数',
+    `avg_duration` float DEFAULT '0' COMMENT '平均停留时长',
+    `create_time` int(11) DEFAULT '0',
+    `update_time` int(11) DEFAULT '0',
+    PRIMARY KEY (`id`),
+    UNIQUE KEY `date` (`date`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='小程序每日统计表';
+
+-- 3. 埋点失败数据表
+CREATE TABLE `fa_shopro_miniprogram_track_failed` (
+                                       `id` int(11) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键ID',
+                                       `data` text COMMENT '原始失败数据(JSON格式)',
+                                       `error` varchar(500) DEFAULT '' COMMENT '错误信息',
+                                       `attempts` int(11) DEFAULT '0' COMMENT '重试次数',
+                                       `status` tinyint(1) DEFAULT '0' COMMENT '状态：0=待处理，1=已处理，2=已放弃',
+                                       `create_time` int(11) DEFAULT '0' COMMENT '创建时间',
+                                       `update_time` int(11) DEFAULT '0' COMMENT '更新时间',
+                                       PRIMARY KEY (`id`),
+                                       KEY `status` (`status`),
+                                       KEY `create_time` (`create_time`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='埋点失败数据表';
+
+```
+
+---
+
+### 二、小程序端埋点代码
+
+#### 2.1 创建 tracker.js
+
+```javascript
+// miniprogram/utils/tracker.js
+class Tracker {
+  constructor() {
+    this.baseUrl = 'https://your-domain.com/addons/shopro/api/track';
+    this.sessionId = this.getSessionId();
+    this.userId = 0;
+    this.openid = '';
+    this.queue = [];
+    this.timer = null;
+    this.maxQueueSize = 20;
+  }
+
+  /**
+   * 获取会话ID
+   */
+  getSessionId() {
+    let id = wx.getStorageSync('track_session');
+    if (!id) {
+      id = Date.now().toString(36) + Math.random().toString(36).substr(2);
+      wx.setStorageSync('track_session', id);
+      // 会话有效期2小时
+      setTimeout(() => wx.removeStorageSync('track_session'), 2 * 60 * 60 * 1000);
+    }
+    return id;
+  }
+
+  /**
+   * 设置用户信息
+   */
+  setUser(userId, openid) {
+    this.userId = userId || 0;
+    this.openid = openid || '';
+  }
+
+  /**
+   * 页面访问埋点
+   */
+  trackPage(pagePath, pageTitle = '') {
+    this.track('page_view', { page_path: pagePath, page_title: pageTitle });
+  }
+
+  /**
+   * 自定义事件埋点
+   */
+  track(eventType, params = {}) {
+    const data = {
+      session_id: this.sessionId,
+      user_id: this.userId,
+      openid: this.openid,
+      event_type: eventType,
+      create_time: Date.now(),
+      ...params
+    };
+    
+    this.queue.push(data);
+    
+    // 队列满了立即发送
+    if (this.queue.length >= this.maxQueueSize) {
+      this.upload();
+    } else {
+      // 否则延迟发送
+      this.scheduleUpload();
+    }
+  }
+
+  /**
+   * 点击事件埋点
+   */
+  trackClick(elementId, elementName, params = {}) {
+    this.track('click', {
+      element_id: elementId,
+      element_name: elementName,
+      ...params
+    });
+  }
+
+  /**
+   * 延迟上传
+   */
+  scheduleUpload() {
+    if (this.timer) clearTimeout(this.timer);
+    this.timer = setTimeout(() => this.upload(), 5000);
+  }
+
+  /**
+   * 上传数据
+   */
+  upload() {
+    if (this.queue.length === 0) return;
+
+    const data = [...this.queue];
+    this.queue = [];
+
+    wx.request({
+      url: this.baseUrl,
+      method: 'POST',
+      data: { list: data },
+      header: { 'Content-Type': 'application/json' },
+      timeout: 30000,
+      success: () => {
+        // 上传成功，清除缓存
+        wx.removeStorageSync('track_cache');
+      },
+      fail: () => {
+        // 失败则保存到本地缓存
+        const cache = wx.getStorageSync('track_cache') || [];
+        wx.setStorageSync('track_cache', [...cache, ...data]);
+      }
+    });
+  }
+
+  /**
+   * 同步缓存数据
+   */
+  syncCache() {
+    const cache = wx.getStorageSync('track_cache');
+    if (cache && cache.length > 0) {
+      wx.request({
+        url: this.baseUrl,
+        method: 'POST',
+        data: { list: cache },
+        success: () => wx.removeStorageSync('track_cache')
+      });
+    }
+  }
+
+  /**
+   * 页面卸载时调用
+   */
+  onUnload() {
+    if (this.timer) clearTimeout(this.timer);
+    this.upload();
+  }
+}
+
+module.exports = new Tracker();
+```
+
+#### 2.2 在页面中使用
+
+```javascript
+// miniprogram/pages/index/index.js
+const tracker = require('../../utils/tracker.js');
+
+Page({
+  onLoad(options) {
+    // 设置用户信息（登录后）
+    const user = wx.getStorageSync('user');
+    if (user) {
+      tracker.setUser(user.id, user.openid);
+    }
+
+    // 页面访问埋点
+    tracker.trackPage('/pages/index/index', '首页');
+  },
+
+  onShow() {
+    // 同步缓存数据（网络恢复时）
+    tracker.syncCache();
+  },
+
+  onUnload() {
+    // 页面卸载时上传剩余数据
+    tracker.onUnload();
+  },
+
+  handleProductClick(e) {
+    const product = e.currentTarget.dataset.product;
+    tracker.trackClick('product_card', '商品卡片', {
+      product_id: product.id,
+      product_name: product.name
+    });
+  }
+});
+```
+
+---
+
+### 三、后端接口代码
+
+#### 3.1 API控制器
+
+```php
+<?php
+// addons/shopro/controller/api/Track.php
+
+namespace addons\shopro\controller\api;
+
+use think\Queue;
+use app\admin\controller\shopro\Common;
+
+class Track extends Common
+{
+    /**
+     * 接收埋点数据（推入队列）
+     */
+    public function index() {
+        // 限制请求大小
+        if ($_SERVER['CONTENT_LENGTH'] > 5 * 1024 * 1024) {
+            $this->error('请求数据过大');
+        }
+
+        // 获取原始数据
+        $rawData = file_get_contents('php://input');
+        if (empty($rawData)) {
+            $this->error('请求数据为空');
+        }
+
+        // 解析JSON
+        $requestData = json_decode($rawData, true);
+        if (json_last_error() !== JSON_ERROR_NONE) {
+            $this->error('JSON解析失败');
+        }
+
+        // 验证必要参数
+        if (!isset($requestData['list']) || !is_array($requestData['list'])) {
+            $this->error('参数错误');
+        }
+
+        // 推入队列
+        $jobId = Queue::push('addons\shopro\job\TrackProcessJob', [
+            'data' => $requestData['list'],
+            'ip' => $this->request->ip(),
+            'timestamp' => time()
+        ]);
+
+        if ($jobId) {
+            $this->success('接收成功');
+        } else {
+            $this->error('队列推送失败');
+        }
+    }
+
+    /**
+     * 获取统计数据
+     */
+    public function stats() {
+        $startDate = $this->request->param('start', date('Y-m-d', strtotime('-7 days')));
+        $endDate = $this->request->param('end', date('Y-m-d'));
+
+        // 参数验证
+        if (!preg_match('/^\d{4}-\d{2}-\d{2}$/', $startDate) || !preg_match('/^\d{4}-\d{2}-\d{2}$/', $endDate)) {
+            $this->error('日期格式错误');
+        }
+
+        $stats = \app\admin\model\shopro\MiniProgramStats::whereBetween('date', [$startDate, $endDate])
+            ->order('date ASC')
+            ->select();
+
+        $this->success('获取成功', $stats);
+    }
+}
+```
+
+#### 3.2 队列任务
+
+```php
+<?php
+// addons/shopro/job/TrackProcessJob.php
+
+namespace addons\shopro\job;
+
+use think\queue\Job;
+use think\Db;
+
+class TrackProcessJob
+{
+    /**
+     * 执行队列任务
+     */
+    public function fire(Job $job, $data) {
+        try {
+            $trackData = $data['data'] ?? [];
+            $clientIp = $data['ip'] ?? '';
+
+            if (empty($trackData)) {
+                $job->delete();
+                return;
+            }
+
+            // 1. 批量插入埋点数据
+            $this->insertTracks($trackData, $clientIp);
+
+            // 2. 更新统计数据
+            $this->updateStats();
+
+            // 3. 删除任务
+            $job->delete();
+
+        } catch (\Exception $e) {
+            // 重试机制
+            if ($job->attempts() < 3) {
+                $job->release(60); // 60秒后重试
+            } else {
+                // 记录失败数据
+                $this->saveFailed($data, $e->getMessage());
+                $job->delete();
+            }
+        }
+    }
+
+    /**
+     * 批量插入埋点数据
+     */
+    protected function insertTracks($trackData, $clientIp) {
+        $insertData = [];
+        $now = time();
+
+        foreach ($trackData as $item) {
+            $insertData[] = [
+                'user_id' => $item['user_id'] ?? 0,
+                'openid' => $item['openid'] ?? '',
+                'session_id' => $item['session_id'] ?? '',
+                'event_type' => $item['event_type'] ?? '',
+                'page_path' => $item['page_path'] ?? '',
+                'page_title' => $item['page_title'] ?? '',
+                'params' => !empty($item['params']) ? json_encode($item['params']) : '',
+                'ip' => $clientIp,
+                'create_time' => isset($item['create_time']) ? (int)($item['create_time'] / 1000) : $now
+            ];
+        }
+
+        // 分批插入（每批500条）
+        $batches = array_chunk($insertData, 500);
+        foreach ($batches as $batch) {
+            Db::name('shopro_miniprogram_track')->insertAll($batch);
+        }
+    }
+
+    /**
+     * 更新统计数据
+     */
+    protected function updateStats() {
+        $date = date('Y-m-d');
+
+        // 统计UV（按openid去重）
+        $uv = Db::name('shopro_miniprogram_track')
+            ->whereDate('create_time', $date)
+            ->distinct(true)
+            ->count('openid');
+
+        // 统计PV（页面浏览量）
+        $pv = Db::name('shopro_miniprogram_track')
+            ->whereDate('create_time', $date)
+            ->where('event_type', 'page_view')
+            ->count();
+
+        // 更新或插入统计记录
+        Db::name('shopro_miniprogram_stats')->updateOrCreate(
+            ['date' => $date],
+            [
+                'uv' => $uv,
+                'pv' => $pv,
+                'update_time' => time()
+            ]
+        );
+    }
+
+    /**
+     * 保存失败数据
+     */
+    protected function saveFailed($data, $error) {
+        Db::name('shopro_track_failed')->insert([
+            'data' => json_encode($data),
+            'error' => $error,
+            'attempts' => 3,
+            'create_time' => time()
+        ]);
+    }
+}
+```
+
+---
+
+### 四、请求入参说明
+
+#### 4.1 接口地址
+
+```
+POST https://your-domain.com/addons/shopro/api/track
+```
+
+#### 4.2 请求体格式
+
+```json
+{
+    "list": [
+        {
+            "session_id": "abc123xyz",
+            "user_id": 123,
+            "openid": "o1234567890abcdef",
+            "event_type": "page_view",
+            "page_path": "/pages/index/index",
+            "page_title": "首页",
+            "params": {"product_id": 1},
+            "create_time": 1705315200000
+        },
+        {
+            "session_id": "abc123xyz",
+            "event_type": "click",
+            "element_id": "product_card",
+            "element_name": "商品卡片"
+        }
+    ]
+}
+```
+
+#### 4.3 字段说明
+
+| 字段 | 类型 | 必填 | 说明 |
+|------|------|------|------|
+| `list` | array | 是 | 埋点数据列表 |
+| `list[].session_id` | string | 是 | 会话ID |
+| `list[].user_id` | int | 否 | 用户ID（登录后） |
+| `list[].openid` | string | 否 | 微信openid |
+| `list[].event_type` | string | 是 | 事件类型 |
+| `list[].page_path` | string | 否 | 页面路径 |
+| `list[].page_title` | string | 否 | 页面标题 |
+| `list[].params` | object | 否 | 自定义参数 |
+| `list[].create_time` | int | 否 | 创建时间戳(毫秒) |
+
+#### 4.4 事件类型说明
+
+| 类型 | 说明 | 场景 |
+|------|------|------|
+| `page_view` | 页面访问 | 页面加载时 |
+| `click` | 点击事件 | 按钮/卡片点击 |
+| `share` | 分享事件 | 用户分享 |
+| `form_submit` | 表单提交 | 表单提交 |
+
+---
+
+### 五、队列配置与启动
+
+#### 5.1 配置文件
+
+```php
+// config/queue.php
+return [
+    'default' => 'redis',
+    'connections' => [
+        'redis' => [
+            'type' => 'redis',
+            'queue' => 'default',
+            'host' => '127.0.0.1',
+            'port' => 6379,
+            'password' => '',
+            'select' => 0,
+        ],
+    ],
+    'failed' => [
+        'type' => 'database',
+        'table' => 'shopro_track_failed',
+    ],
+];
+```
+
+#### 5.2 启动命令
+
+```bash
+# 开发环境
+php think queue:listen
+
+# 生产环境（后台运行）
+nohup php think queue:listen > runtime/log/queue.log 2>&1 &
+```
+
+---
+
+### 六、架构流程图
+
+```
+小程序端              API控制器              队列处理器              数据库
+    |                     |                     |                    |
+    |--POST /api/track---->|                     |                    |
+    |                     |                     |                    |
+    |                     |--push Queue-------->|                    |
+    |                     |                     |                    |
+    |<--200 OK------------|                     |                    |
+    |                     |                     |                    |
+    |                     |                     |--insert----------->|
+    |                     |                     |--update stats----->|
+```
+
+---
+
+### 七、关键特性
+
+| 特性 | 说明 |
+|------|------|
+| **异步处理** | 数据推入队列后立即返回 |
+| **批量插入** | 每批500条，提高效率 |
+| **重试机制** | 失败自动重试3次 |
+| **本地缓存** | 网络异常时数据暂存 |
+| **自动统计** | 队列处理后更新UV/PV |
+
+---
+
+### 八、使用注意事项
+
+1. **队列依赖**：需要安装 Redis 作为队列驱动
+2. **进程管理**：生产环境建议使用 Supervisor 管理队列进程
+3. **数据清理**：定期清理过期的埋点数据
+4. **错误排查**：失败数据记录在 `shopro_track_failed` 表中
+        
\ No newline at end of file

addons/shopro/controller/api/Track.php

+<?php
+namespace addons\shopro\controller\api;
+
+use think\Queue;
+//use app\admin\controller\shopro\Common;
+use addons\shopro\controller\Common;
+
+class Track extends Common
+{
+
+    /**
+     * 无需登录的方法
+     * @var array
+     */
+    protected $noNeedLogin = ['upload', 'stats'];
+    protected $noNeedRight = ['*'];
+
+    /**
+     * 接收埋点数据（推入队列）
+     */
+    public function upload() {
+
+        // 限制请求大小
+        if ($_SERVER['CONTENT_LENGTH'] > 5 * 1024 * 1024) {
+            $this->error('请求数据过大');
+        }
+
+        // 获取原始数据
+        $rawData = file_get_contents('php://input');
+        if (empty($rawData)) {
+            $this->error('请求数据为空');
+        }
+
+        // 解析JSON
+        $requestData = json_decode($rawData, true);
+        if (json_last_error() !== JSON_ERROR_NONE) {
+            $this->error('JSON解析失败');
+        }
+
+        // 验证必要参数
+        if (!isset($requestData['list']) || !is_array($requestData['list'])) {
+            $this->error('参数错误');
+        }
+//print_r([
+//    'data' => $requestData['list'],
+//    'ip' => $this->request->ip(),
+//    'timestamp' => time()
+//]);
+        // 推入队列
+        $jobId = Queue::push('addons\shopro\job\TrackProcessJob@fire', [
+            'data' => $requestData['list'],
+            'ip' => $this->request->ip(),
+            'timestamp' => time()
+        ], 'shopro-high');
+
+        if ($jobId) {
+            $this->success('接收成功');
+        } else {
+            $this->error('队列推送失败');
+        }
+    }
+
+    /**
+     * 获取统计数据
+     */
+    public function stats() {
+        $startDate = $this->request->param('start', date('Y-m-d', strtotime('-7 days')));
+        $endDate = $this->request->param('end', date('Y-m-d'));
+
+        // 参数验证
+        if (!preg_match('/^\d{4}-\d{2}-\d{2}$/', $startDate) || !preg_match('/^\d{4}-\d{2}-\d{2}$/', $endDate)) {
+            $this->error('日期格式错误');
+        }
+
+        $stats = \app\admin\model\shopro\MiniProgramStats::whereBetween('date', [$startDate, $endDate])
+            ->order('date ASC')
+            ->select();
+
+        $this->success('获取成功', $stats);
+    }
+}
\ No newline at end of file

addons/shopro/job/TrackProcessJob.php

+<?php
+// addons/shopro/job/TrackProcessJob.php
+
+namespace addons\shopro\job;
+
+use think\queue\Job;
+use think\Db;
+use think\Queue;
+
+class TrackProcessJob
+{
+    /**
+     * 执行队列任务
+     */
+    public function fire(Job $job, $data) {
+        try {
+            $trackData = $data['data'] ?? [];
+            $clientIp = $data['ip'] ?? '';
+
+            if (empty($trackData)) {
+                $job->delete();
+                return;
+            }
+
+            // 1. 批量插入埋点数据
+            $this->insertTracks($trackData, $clientIp);
+
+            // 2. 更新统计数据
+            $this->updateStats();
+
+            // 3. 删除任务
+            $job->delete();
+
+        } catch (\Exception $e) {
+            // 重试机制
+            if ($job->attempts() < 3) {
+                $job->release(60); // 60秒后重试
+            } else {
+                // 记录失败数据
+                $this->saveFailed($data, $e->getMessage());
+                $job->delete();
+            }
+        }
+    }
+
+    /**
+     * 批量插入埋点数据
+     */
+    protected function insertTracks($trackData, $clientIp) {
+        $insertData = [];
+        $now = time();
+
+        foreach ($trackData as $item) {
+            $insertData[] = [
+                'user_id' => $item['user_id'] ?? 0,
+                'openid' => $item['openid'] ?? '',
+                'session_id' => $item['session_id'] ?? '',
+                'event_type' => $item['event_type'] ?? '',
+                'page_path' => $item['page_path'] ?? '',
+                'page_title' => $item['page_title'] ?? '',
+                'params' => !empty($item['params']) ? json_encode($item['params']) : '',
+                'ip' => $clientIp,
+                'create_time' => isset($item['create_time']) ? (int)($item['create_time'] / 1000) : $now
+            ];
+        }
+
+        // 分批插入（每批500条）
+        $batches = array_chunk($insertData, 500);
+        foreach ($batches as $batch) {
+            Db::name('shopro_miniprogram_track')->insertAll($batch);
+        }
+    }
+
+    /**
+     * 更新统计数据
+     */
+    protected function updateStats() {
+        $date = date('Y-m-d');
+
+        // 统计UV（按openid去重）
+        $uv = Db::name('shopro_miniprogram_track')
+            ->whereDate('create_time', $date)
+            ->distinct(true)
+            ->count('openid');
+
+        // 统计PV（页面浏览量）
+        $pv = Db::name('shopro_miniprogram_track')
+            ->whereDate('create_time', $date)
+            ->where('event_type', 'page_view')
+            ->count();
+
+        // 更新或插入统计记录
+        Db::name('shopro_miniprogram_stats')->updateOrCreate(
+            ['date' => $date],
+            [
+                'uv' => $uv,
+                'pv' => $pv,
+                'update_time' => time()
+            ]
+        );
+    }
+
+    /**
+     * 保存失败数据
+     */
+    protected function saveFailed($data, $error) {
+        Db::name('shopro_track_failed')->insert([
+            'data' => json_encode($data),
+            'error' => $error,
+            'attempts' => 3,
+            'create_time' => time()
+        ]);
+    }
+
+
+    // 恢复失败数据的脚本
+    public function retryFailed()
+    {
+        $failedList = Db::name('shopro_track_failed')
+            ->where('status', 0)
+            ->select();
+
+        foreach ($failedList as $item) {
+            try {
+                $data = json_decode($item['data'], true);
+
+                // 重新推入队列
+                Queue::push('\addons\shopro\job\TrackProcessJob@fire', $data, 'shopro-high');
+
+                // 更新状态为已处理
+                Db::name('shopro_track_failed')
+                    ->where('id', $item['id'])
+                    ->update(['status' => 1, 'update_time' => time()]);
+
+            } catch (\Exception $e) {
+                // 更新状态为已放弃
+                Db::name('shopro_track_failed')
+                    ->where('id', $item['id'])
+                    ->update(['status' => 2, 'error' => $e->getMessage(), 'update_time' => time()]);
+            }
+        }
+    }
+}
\ No newline at end of file