基于Flume+Spark Streaming的实时日志入侵检测系统

简介本资源是一个基于Flume、Spark与Flask构建的分布式实时日志分析与入侵检测系统面向大数据初学者、毕业设计学生及安全分析实践者聚焦于Web服务器日志如access_log的采集、流式处理、异常行为识别与可视化展示可支撑课程设计、毕设开发与小型安全监控场景。压缩包共108个文件含14个编译后class文件、16个导出配置export、7个PNG图表、5个说明类txt与1个核心data样本辅以Scala/Java源码、Flask Web接口、Spark Streaming作业及配置文件conf、properties整体18.91MB结构完整、模块清晰便于分层理解数据流转链路。目前已有325人学习下载资源经本地全链路编译验证附详细环境配置文档与助教审定保障提供从日志接入、特征提取、规则匹配到结果呈现的端到端可运行方案特别适合夯实实时计算与安全日志分析双重能力。1. 这不是又一个“日志可视化看板”它用 Flume 实时接 Apache/Nginx 日志Spark Streaming 做滑动窗口统计 规则匹配含 SQL 式 IOC 提取Flask 暴露 REST API 简洁 Web 界面真正跑在三节点伪分布式环境里、能复现入侵行为链的毕业设计级系统你可能已经下载过十几个“日志分析系统”压缩包解压后发现是 Flask 写的静态页面 本地 CSV 加载 三个 matplotlib 图表——那叫日志展示不叫实时分析。而这个.zip包里的东西从 Flume agent 配置开始就踩在真实生产逻辑上它监听/var/log/nginx/access.log的 tail -F 行为通过 Avro Sink 推给 Spark Streaming 的 ReceiverInputDStreamSpark 侧不是简单 count()而是用windowDuration30s, slideDuration10s构建滑动窗口对每个窗口内 IP 做请求频次、404 比率、User-Agent 异常熵值、SQL 注入关键词如union select, or 11正则命中数四维聚合结果存入内存 MapState 并触发阈值告警比如 10s 内单 IP 404 50 且含注入词再经 Flask REST 接口推送到前端 ECharts 动态拓扑图。它不依赖 HDFS 高可用但明确要求 Spark Standalone 三节点master 2 worker所有配置文件flume-conf.properties、spark-defaults.conf、application.py都带注释标注端口/路径/序列化方式。适合需要答辩演示“数据从产生到告警闭环”的本科毕设也足够作为校招项目讲清楚流式架构分层采集层→计算层→服务层的边界与协作。2. 从零搭起三节点 Spark Standalone 集群为什么必须用spark-submit --master spark://master:7077而不是 local[*]以及如何绕过 YARN 依赖直接跑通 Streaming2.1 为什么毕业设计必须坚持 Standalone 模式避开 Hadoop 生态的“伪分布式陷阱”很多同学在毕设里写“基于 Spark 的实时分析”实际代码里全是SparkContext(sc SparkContext(local[4], LogAnalysis))——这根本不算分布式连单机多线程都算不上只是把 RDD 分片在本机 CPU 核上跑。而本系统要求的Standalone 模式是 Spark 官方推荐的轻量级集群管理器它不依赖 Hadoop 生态HDFS/YARN却能真实体现 Driver 与 Executor 的网络通信、Shuffle 数据拉取、Executor 内存隔离等核心机制。尤其对 Streaming 场景StreamingContext必须连接到集群 Masterspark://master:7077否则ReceiverInputDStream无法在 Worker 上启动 NetcatReceiver 或 FlumePollingReceiver。本系统spark-streaming-flume_2.12-3.3.2.jar已预编译进 lib 目录但若你用 Spark 3.4需手动替换为对应 Scala 版本的 jar见后文避坑章。提示本系统默认使用 Spark 3.3.2 Scala 2.12所有 pom.xml 和 build.sbt 中 scalaVersion 必须严格匹配否则ClassNotFoundException: org.apache.spark.streaming.flume.FlumeUtils会直接中断启动。2.2 三节点部署实操master、worker1、worker2 的角色分工与关键配置项我们以三台虚拟机或同一物理机的三个终端模拟最小可行集群。假设 IP 分配如下节点主机名IP 地址角色mastermaster192.168.56.101Spark Master Flume Agent Flask Web Serverworker1node1192.168.56.102Spark Worker Flume Agent可选worker2node2192.168.56.103Spark Worker第一步统一安装 Spark 并配置环境变量在三台机器上执行以 Ubuntu 22.04 为例# 下载预编译版注意 Scala 版本 wget https://downloads.apache.org/spark/spark-3.3.2/spark-3.3.2-bin-hadoop3.tgz tar -xzf spark-3.3.2-bin-hadoop3.tgz sudo mv spark-3.3.2-bin-hadoop3 /opt/spark echo export SPARK_HOME/opt/spark ~/.bashrc echo export PATH$SPARK_HOME/bin:$PATH ~/.bashrc source ~/.bashrc第二步配置 master 节点的$SPARK_HOME/conf/spark-env.sh# 复制模板并编辑 cp $SPARK_HOME/conf/spark-env.sh.template $SPARK_HOME/conf/spark-env.sh nano $SPARK_HOME/conf/spark-env.sh填入以下内容关键参数已加注释#!/usr/bin/env bash # 必须指定 JAVA_HOME否则 start-master.sh 报错 export JAVA_HOME/usr/lib/jvm/java-11-openjdk-amd64 # Master 绑定地址不能写 0.0.0.0安全限制必须写本机 IP export SPARK_MASTER_HOST192.168.56.101 # Master Web UI 端口默认 8080若被占用可改 export SPARK_MASTER_WEBUI_PORT8080 # Master RPC 端口Spark Streaming 必须通过此端口注册 Receiver export SPARK_MASTER_PORT7077 # JVM 内存参数避免小内存机器 OOM export SPARK_DAEMON_MEMORY1g第三步配置 worker 节点的$SPARK_HOME/conf/spark-env.sh# 在 node1 和 node2 上执行 nano $SPARK_HOME/conf/spark-env.sh填入#!/usr/bin/env bash export JAVA_HOME/usr/lib/jvm/java-11-openjdk-amd64 # 指向 master 的 RPC 地址格式为 spark://host:port export SPARK_MASTERspark://192.168.56.101:7077 # Worker 内存分配建议至少 2gStreaming 场景需更多堆外内存 export SPARK_WORKER_MEMORY2g # Worker 核心数根据 CPU 核心数设置避免超卖 export SPARK_WORKER_CORES2 # Worker Web UI 端口每台机器必须唯一 export SPARK_WORKER_WEBUI_PORT8081 # node1 用 8081node2 改为 8082第四步启动集群并验证在 master 上执行$SPARK_HOME/sbin/start-master.sh # 查看日志确认是否绑定 7077 端口 tail -f $SPARK_HOME/logs/spark-*-org.apache.spark.deploy.master.Master-*.out在 node1 和 node2 上分别执行$SPARK_HOME/sbin/start-worker.sh spark://192.168.56.101:7077 # 查看日志确认是否注册成功 tail -f $SPARK_HOME/logs/spark-*-org.apache.spark.deploy.worker.Worker-*.out此时访问http://192.168.56.101:8080应看到 Web UI 显示 1 个 Alive WorkersTotal Cores4Total Memory4g。这是后续所有 Streaming 任务能运行的前提——如果 UI 里看不到 Workerspark-submit会 fallback 到 local 模式整个“分布式”就失效了。2.3 提交 Streaming 任务spark-submit的必填参数与常见失败原因进入系统源码目录src/main/python/streaming/执行提交命令spark-submit \ --master spark://192.168.56.101:7077 \ --deploy-mode client \ --driver-memory 2g \ --executor-memory 2g \ --executor-cores 2 \ --jars $SPARK_HOME/jars/spark-streaming-flume_2.12-3.3.2.jar,$SPARK_HOME/jars/spark-streaming-flume-sink_2.12-3.3.2.jar \ --py-files $SPARK_HOME/python/lib/pyspark.zip,$SPARK_HOME/python/lib/py4j-0.10.9.5-src.zip \ log_streaming_analyzer.py参数详解与血泪经验--master必须是spark://host:port不能是yarn或local--deploy-mode client必须用 client 模式因为 Streaming 任务需要 Driver 持续运行接收 Flume 数据cluster 模式下 Driver 退出即任务终止--jars显式指定 Flume 集成 jarSpark 3.3.2 对应的 artifactId 是spark-streaming-flume_2.12版本号必须与 Spark 一致否则NoClassDefFoundError--py-filesPySpark 依赖包路径必须准确py4j版本需与 Spark 打包版本一致3.3.2 对应py4j-0.10.9.5log_streaming_analyzer.py主程序内部StreamingContext初始化时指定了batchDuration10秒即每 10 秒处理一个 DStream Batch。若提交后报错Failed to connect to master 192.168.56.101:7077请立即检查① master 是否真的在运行jps应看到 Master 进程② 防火墙是否放行 7077 端口sudo ufw allow 7077③/etc/hosts中是否将192.168.56.101 master正确映射Spark 内部用主机名通信。3. Flume 实时采集从spooldir到exec再到avro为什么本系统强制使用 Avro Source Sink 架构3.1 三种采集方式对比为什么spooldir和exec不适合入侵检测场景Flume 提供多种 Source但并非都适用于安全分析Source 类型原理是否实时是否可靠是否支持断点续传是否适合本系统spooldir监控目录文件移动后读取❌ 秒级延迟轮询间隔✅ 文件移动原子性保证✅ 通过 .COMPLETED 标记否无法捕获正在写入的 access.logexectail -F命令输出✅ 准实时❌ 进程崩溃即丢失数据❌ 无状态记录否入侵行为常发生在日志滚动瞬间tail 可能漏掉最后一行avroAvro RPC 协议接收数据✅ 毫秒级✅ TCP ACK 保障✅ Flume ChannelFileChannel持久化✅ 唯一满足“不丢、不重、低延迟”的方案本系统采用Avro Source在 Spark 端 Avro Sink在 Flume 端的反向架构Flume Agent 不主动拉日志而是作为客户端将解析后的日志事件JSON 格式通过 Avro 协议 Push 到 Spark Streaming 的 AvroSource。这种设计规避了传统exec tail的进程稳定性问题也绕开了spooldir的文件移动延迟。3.2 Flume Agent 配置详解flume-conf.properties中的四个生死参数进入conf/flume-conf.properties核心配置如下# Agent 名称必须与启动命令一致 a1.sources r1 a1.sinks k1 a1.channels c1 # Source使用 exec tail -F但加了关键增强 a1.sources.r1.type exec a1.sources.r1.command tail -F /var/log/nginx/access.log a1.sources.r1.shell /bin/bash -c # 【关键】每行日志必须以 \n 结尾否则 Flume 会粘包 a1.sources.r1.restart true a1.sources.r1.restart Throttle 10000 # Channel必须用 FileChannel 保证可靠性 a1.channels.c1.type file a1.channels.c1.checkpointDir /var/flume/checkpoint a1.channels.c1.dataDirs /var/flume/data # 【关键】事务容量必须 source batch size否则丢数据 a1.channels.c1.transactionCapacity 1000 a1.channels.c1.capacity 1000000 # SinkAvro SinkPush 到 Spark 的 AvroSource a1.sinks.k1.type avro a1.sinks.k1.hostname 192.168.56.101 a1.sinks.k1.port 41414 # 【关键】批次大小必须与 Spark Streaming 的 batchDuration 匹配 a1.sinks.k1.batch-size 100 # 【关键】超时设置避免网络抖动导致阻塞 a1.sinks.k1.connect-timeout 20000 a1.sinks.k1.request-timeout 20000 # Bind source → channel → sink a1.sources.r1.channels c1 a1.sinks.k1.channel c1参数生死线解释transactionCapacity1000Channel 每次事务最多写入 1000 条事件。若batch-size100则每 10 次 Sink 操作才触发一次 Channel 事务降低 IO 压力但若设太小如 100高并发下 Channel 会成为瓶颈batch-size100Sink 每次向 Avro Server 发送 100 条日志。该值需与 Spark Streaming 的batchDuration10s对齐——假设 Nginx 每秒 50 条日志则 10s 产生 500 条batch-size100意味着 5 次网络请求比batch-size1500 次更高效connect-timeout20000连接超时 20 秒。若 Spark AvroSource 未启动Flume 会重试 20 秒后报错而非无限等待restarttrue确保tail -F进程崩溃后自动重启这是execSource 唯一的容错手段。3.3 Spark 端 AvroSource 启动FlumeUtils.createStream()的隐藏约束在log_streaming_analyzer.py中关键初始化代码为from pyspark.streaming.flume import FlumeUtils # 创建 StreamingContextbatchDuration10秒 ssc StreamingContext(sc, 10) # 【关键】AvroSource 绑定在 41414 端口必须与 flume-conf.properties 中 sink.port 一致 flumeStream FlumeUtils.createStream( ssc, hostname192.168.56.101, # Spark Driver 所在机器 port41414, # 必须与 Flume sink.port 完全相同 protoClassorg.apache.flume.source.avro.AvroFlumeEvent, # 固定值 transformlambda x: x # 默认返回原始 event后续需解析 body )注意hostname参数不是 Flume Agent 的地址而是Spark Driver 绑定的地址。因为 AvroSource 是 Spark 启动的 Netty ServerFlume Sink 是 Client所以hostname必须是 Spark Driver 所在机器即 master能被 Flume 访问到的 IP。若填localhostFlume 会尝试连127.0.0.1:41414而该端口只在 master 本机监听node1/node2 无法访问导致Connection refused。提示启动前务必在 master 上执行sudo lsof -i :41414确认端口空闲若被占用修改flume-conf.properties和 Python 代码中的 port 为 41415并同步更新。4. 入侵检测规则引擎从硬编码正则到可热加载 JSON 规则库以及 Spark 中读取 JSON 的正确姿势4.1 为什么不用 ML 模型基于规则的轻量级检测更适合毕业设计场景当前系统未集成孤立森林或 LSTM 等模型原因很务实①数据冷启动问题Nginx access.log 缺乏标签正常/攻击无法监督训练②特征工程成本高IP 地理位置、UA 设备指纹、Referer 可信度等需外部 API增加部署复杂度③可解释性刚需答辩时老师问“为什么判定这个 IP 是攻击者”展示正则r(?i)union\sselect比解释 embedding cosine similarity 更直观。因此系统采用四维规则组合高频扫描10s 内单 IP 请求 100 次异常状态码404 比率 70%UA 异常熵len(set([ua[:3] for ua in uas])) 2大量 UA 以curl/python-requests开头IOC 关键词SQLi/XSS/Path Traversal 三类共 23 个正则见rules/ioc_rules.json。4.2 Spark 中读取 JSON 规则spark.read.json()的陷阱与broadcast正确用法规则文件rules/ioc_rules.json格式如下[ {name: SQLi_UNION_SELECT, pattern: (?i)union\\sselect, severity: high}, {name: XSS_SCRIPT_TAG, pattern: script[^]*, severity: medium}, {name: PATH_TRAV_ASLASHDOT, pattern: \\.{2}/, severity: high} ]错误做法直接在 map 中读取文件# ❌ 千万不要这样写每个 Partition 都会打开文件IO 爆炸 def check_ioc(log_line): with open(rules/ioc_rules.json) as f: rules json.load(f) for rule in rules: if re.search(rule[pattern], log_line): return rule[name] return None正确做法Broadcast 预编译正则# ✅ 在 Driver 端一次性读取并广播 with open(rules/ioc_rules.json) as f: raw_rules json.load(f) # 预编译正则避免每个 record 重复 compile compiled_rules [ (rule[name], re.compile(rule[pattern]), rule[severity]) for rule in raw_rules ] # 广播到所有 Executor bc_rules sc.broadcast(compiled_rules) # 在 RDD map 中使用 def check_ioc_broadcast(log_line): rules bc_rules.value # 获取广播变量 for name, pattern, severity in rules: if pattern.search(log_line): return (name, severity, log_line[:100]) # 返回规则名、等级、日志片段 return None # 应用到 DStream alerts parsed_logs.map(check_ioc_broadcast).filter(lambda x: x is not None)为什么必须 broadcastsc.broadcast()将变量序列化后发送到每个 Executor 的内存中只传输一次若用map内部open()每个 Task每个 Partition 的每个 record都会触发磁盘 IO10 万条日志 10 万次文件打开必然 OOM预编译re.compile()避免正则引擎重复解析提升 3~5 倍匹配速度。4.3 滑动窗口聚合reduceByKeyAndWindow的窗口函数与触发时机入侵检测的核心是“行为链”单条日志无意义需看时间窗口内的模式。系统使用reduceByKeyAndWindow实现# 每条日志解析为 (ip, 1) 键值对 ip_counts parsed_logs.map(lambda log: (log[remote_addr], 1)) # 滑动窗口窗口长度 30 秒每 10 秒滑动一次 # reduceFunc: (acc, new) acc new 累加 # invReduceFunc: (acc, old) acc - old 减去离开窗口的老值需启用 state windowed_counts ip_counts.reduceByKeyAndWindow( reduceFunclambda x, y: x y, invReduceFunclambda x, y: x - y, # 启用增量计算必须提供 windowDuration30, # 窗口长度 30 秒 slideDuration10, # 每 10 秒计算一次 numPartitions4 # 避免单 partition 热点 ) # 过滤出高频 IP50 次/30s suspicious_ips windowed_counts.filter(lambda kv: kv[1] 50)关键点invReduceFunc是性能命脉若不提供Spark 每次窗口计算都重新 scan 全量数据O(n²) 复杂度提供后变为 O(n)仅更新增删部分numPartitions4默认 1 个 partition所有 IP hash 到同一 partition造成单核瓶颈设为 4 后负载均衡窗口时间单位是秒batchDuration10意味着每 10 秒生成一个 micro-batchwindowDuration30即跨 3 个 batch。5. Flask Web 服务与前端联动REST API 设计、ECharts 动态渲染、以及部署时的flask run --host0.0.0.0玄学5.1 Flask API 接口设计为什么/api/alerts/latest必须用app.route而非 WebSocket系统提供两个核心接口接口方法用途数据格式更新频率/api/alerts/latestGET获取最近 10 条告警JSON Array每 5 秒 AJAX 轮询/api/top_ipsGET获取当前 Top5 高频 IPJSON Object每 10 秒轮询为什么不选 WebSocketWebSocket 需维护长连接在 Flask 中需额外引入Flask-SocketIOeventlet增加部署复杂度毕设演示场景下5 秒轮询延迟完全可接受且前端 EChartssetOption()支持平滑过渡动画app.route更符合 RESTful 规范便于 Postman 测试和答辩现场 curl 验证。app.py中关键代码from flask import Flask, jsonify import redis # 使用 Redis 作中间缓存解耦 Spark 与 Flask app Flask(__name__) # Redis 连接池存储最新告警List和 Top IPSorted Set r redis.Redis(hostlocalhost, port6379, db0, decode_responsesTrue) app.route(/api/alerts/latest) def get_latest_alerts(): # LRANGE 获取最新 10 条Redis List 天然支持 LIFO alerts r.lrange(alerts, 0, 9) # 解析 JSON 字符串 return jsonify([json.loads(alert) for alert in alerts]) app.route(/api/top_ips) def get_top_ips(): # ZREVRANGE 获取分数最高的 5 个 IP分数 请求次数 ips r.zrevrange(top_ips, 0, 4, withscoresTrue) return jsonify({ip: int(score) for ip, score in ips})为什么用 RedisSpark Streaming 任务持续运行不能直接暴露DStream.foreachRDD()到 FlaskRedis 作为共享存储Spark 侧用r.lpush(alerts, json.dumps(alert))和r.zincrby(top_ips, 1, ip)更新Flask 侧只读无锁竞争性能稳定。5.2 前端 ECharts 渲染动态拓扑图的节点/边数据构造逻辑templates/index.html中ECharts 初始化代码// 初始化拓扑图 const chart echarts.init(document.getElementById(topology)); chart.setOption({ tooltip: {}, animation: false, series: [{ type: graph, layout: force, force: { repulsion: 1000 }, data: [], // 节点数组 links: [], // 边数组 categories: [ { name: Normal }, { name: Suspicious }, { name: Attacker } ] }] }); // 每 5 秒拉取新数据并更新 function updateChart() { fetch(/api/alerts/latest) .then(r r.json()) .then(alerts { const nodes []; const links []; const attackerIPs new Set(); // Step 1: 收集所有告警中的 attacker IP alerts.forEach(alert { if (alert.ip alert.severity high) { attackerIPs.add(alert.ip); } }); // Step 2: 构造节点attacker 为红色normal 为绿色 attackerIPs.forEach(ip { nodes.push({ name: ip, value: 10, category: 2, // Attacker symbolSize: 30, itemStyle: { color: #e74c3c } }); }); // Step 3: 构造边attacker → targettarget 来自 referer 或 uri alerts.forEach(alert { if (attackerIPs.has(alert.ip) alert.target) { links.push({ source: alert.ip, target: alert.target, lineStyle: { width: 2, curveness: 0.2 } }); } }); chart.setOption({ series: [{ data: nodes, links: links }] }); }); } setInterval(updateChart, 5000); updateChart();关键逻辑attackerIPs是 Set自动去重避免同一 IP 多次渲染links构造时alert.target来自日志中的request_uri如/admin/login.php或referer如http://evil.com/exploit.js形成“攻击者 → 受害资源”关系symbolSize: 30和itemStyle.color强化视觉区分答辩时老师一眼看出攻击链。5.3 Flask 部署避坑flask run --host0.0.0.0为何是玄学以及gunicorn的必要性现象本地开发时flask run正常部署到 master 服务器后浏览器访问http://192.168.56.101:5000显示空白curl http://localhost:5000却返回 HTML。原因Flask 默认绑定127.0.0.1:5000只接受本机回环请求。--host0.0.0.0强制监听所有网卡但存在两大风险①生产环境不安全flask run是 Werkzeug 开发服务器无并发能力高并发下直接挂②端口冲突若iptables或云平台安全组未放行 5000 端口外部无法访问。正确做法生产部署# 安装 gunicorn pip install gunicorn # 启动workers 数 CPU 核数 * 2 1 gunicorn -w 4 -b 0.0.0.0:5000 --timeout 120 app:app参数说明-w 4启动 4 个 worker 进程处理并发请求-b 0.0.0.0:5000绑定所有地址的 5000 端口--timeout 120请求超时 120 秒避免长连接阻塞app:app第一个app是文件名app.py第二个app是 Flask 实例名。注意启动前确保app.py中if __name__ __main__:块已删除或注释否则 gunicorn 会执行两次。6. 真实应急响应验证用ab模拟 CC 攻击、sqlmap触发 IOC、以及从日志到告警的端到端耗时测量技巧6.1 模拟攻击链三步复现“扫描 → 注入 → 告警”的完整闭环Step 1用 Apache Bench 模拟 CC 攻击高频请求在 client 机器非集群节点执行# 持续 60 秒每秒 100 个并发请求访问 /test.php不存在路径触发 404 ab -n 6000 -c 100 http://192.168.56.101/test.php此时 Flume 的tail -F会捕获大量 404 日志Spark Streaming 的windowed_counts将在 30 秒窗口内统计该 IP 请求量若 50 则进入suspicious_ips。Step 2用 sqlmap 触发 SQL 注入规则# 向存在漏洞的测试接口注入 sqlmap -u http://192.168.56.101/vuln?id1 --techniqueU --batch # 生成日志如GET /vuln?id1%20UNION%20SELECT%20NULL%2Cpassword%20FROM%20usersFlume 采集该行后Spark 的check_ioc_broadcast函数会匹配(?i)union\sselect生成告警事件并写入 Redis。Step 3验证告警是否到达前端打开浏览器http://192.168.56.101:5000观察右上角“最新告警”列表应在 15 秒内3 个 batch出现SQLi_UNION_SELECT条目拓扑图中应出现红色节点攻击者 IP指向/vuln的边redis-cli中执行LRANGE alerts 0 0应看到完整 JSON 告警。6.2 端到端耗时测量从日志写入磁盘到前端显示的 5 个时间戳要证明“实时性”必须量化各环节延迟。我们在关键节点插入时间戳环节时间戳位置测量方法典型耗时T1日志落盘Nginxaccess.log文件末尾tail -1 /var/log/nginx/access.log | awk {print $4}0ms写入即完成T2Flume 采集Flume Agent 的Log4jLogger输出grep SINK.k1 $FLUME_HOME/logs/flume.log | tail -1100~300ms取决于 batch-sizeT3Spark 接收FlumeUtils.createStream()的foreachRDD在foreachRDD内print(Received at:, time.time())200~500ms网络序列化T4规则匹配完成alerts.foreachRDD()内print(Alerted at:, time.time())同上300~800ms正则匹配Redis写入T5前端显示浏览器控制台console.log(new Date())在updateChart()开头添加5~10s5 秒轮询间隔结论从 T1 到 T4 的纯处理链路耗时约1~2 秒完全满足本文还有配套的精品资源点击获取