Hyperf对接风控

Hyperf 风控系统完整案例 PHP 生态没有专用风控库，最佳实践是自建规则引擎 + Redis 滑动窗口 + 异步事件，这也是金融/电商行业的标准做法。 --- 架构设计 请求进入 ↓ RiskMiddleware（同步，<5ms 决策） ↓ RiskEngine → 并行执行多条规则 ↓ 决策：PASS / REVIEW / BLOCK ↓ 异步事件 → 记录日志 / 触发人工审核 --- 完整代码1. 安装依赖composerrequire hyperf/redis hyperf/async-queue hyperf/event ---2. 风险规则接口 // app/Risk/Contracts/RuleInterface.php namespace App\Risk\Contracts;use App\Risk\RiskContext;use App\Risk\RiskResult;interface RuleInterface{publicfunctionname(): string;publicfunctionevaluate(RiskContext$ctx): RiskResult;}---3. 风险上下文&结果 // app/Risk/RiskContext.php namespace App\Risk;class RiskContext{publicfunction__construct(publicreadonlystring$userId, publicreadonlystring$ip, publicreadonlystring$action, // login / pay / withdraw publicreadonlyfloat$amount, publicreadonlyarray$extra=[],){}}// app/Risk/RiskResult.php namespace App\Risk;class RiskResult{const PASS='pass';const REVIEW='review';const BLOCK='block';publicfunction__construct(publicreadonlystring$decision, publicreadonlyint$score, //0-100 publicreadonlystring$reason='', publicreadonlystring$rule='',){}public staticfunctionpass(string$rule): self{returnnew self(self::PASS,0,'',$rule);}public staticfunctionblock(string$rule, string$reason, int$score=100): self{returnnew self(self::BLOCK,$score,$reason,$rule);}public staticfunctionreview(string$rule, string$reason, int$score=50): self{returnnew self(self::REVIEW,$score,$reason,$rule);}}---4. 具体规则实现 // app/Risk/Rules/IpFrequencyRule.php namespace App\Risk\Rules;use App\Risk\Contracts\RuleInterface;use App\Risk\RiskContext;use App\Risk\RiskResult;use Hyperf\Redis\Redis;class IpFrequencyRule implements RuleInterface{publicfunction__construct(private Redis$redis){}publicfunctionname(): string{return'ip_frequency';}publicfunctionevaluate(RiskContext$ctx): RiskResult{$key="risk:ip:{$ctx->ip}:{$ctx->action}";$count=$this->redis->incr($key);if($count===1){$this->redis->expire($key,60);//60秒窗口}if($count>20){returnRiskResult::block($this->name(),"IP {$ctx->ip} 60秒内请求 {$count} 次",90);}if($count>10){returnRiskResult::review($this->name(),"IP频率偏高: {$count}/min",40);}returnRiskResult::pass($this->name());}}// app/Risk/Rules/UserVelocityRule.php namespace App\Risk\Rules;use App\Risk\Contracts\RuleInterface;use App\Risk\RiskContext;use App\Risk\RiskResult;use Hyperf\Redis\Redis;class UserVelocityRule implements RuleInterface{publicfunction__construct(private Redis$redis){}publicfunctionname(): string{return'user_velocity';}publicfunctionevaluate(RiskContext$ctx): RiskResult{if($ctx->action!=='pay'&&$ctx->action!=='withdraw'){returnRiskResult::pass($this->name());}// 滑动窗口：记录最近1小时的交易金额$key="risk:user_amount:{$ctx->userId}";$now=time();$window=3600;$this->redis->zAdd($key,$now,$now.':'.$ctx->amount);$this->redis->zRemRangeByScore($key,0,$now-$window);$this->redis->expire($key,$window);$members=$this->redis->zRange($key,0, -1);$total=array_sum(array_map(fn($m)=>(float)explode(':',$m)[1],$members));if($total>50000){returnRiskResult::block($this->name(),"1小时累计金额 {$total} 超限",95);}if($total>20000){returnRiskResult::review($this->name(),"1小时累计金额 {$total} 偏高",60);}returnRiskResult::pass($this->name());}}// app/Risk/Rules/BlacklistRule.php namespace App\Risk\Rules;use App\Risk\Contracts\RuleInterface;use App\Risk\RiskContext;use App\Risk\RiskResult;use Hyperf\Redis\Redis;class BlacklistRule implements RuleInterface{publicfunction__construct(private Redis$redis){}publicfunctionname(): string{return'blacklist';}publicfunctionevaluate(RiskContext$ctx): RiskResult{if($this->redis->sIsMember('risk:blacklist:user',$ctx->userId)){returnRiskResult::block($this->name(),'用户在黑名单中',100);}if($this->redis->sIsMember('risk:blacklist:ip',$ctx->ip)){returnRiskResult::block($this->name(),'IP在黑名单中',100);}returnRiskResult::pass($this->name());}}---5. 风控引擎（并行执行规则） // app/Risk/RiskEngine.php namespace App\Risk;use App\Risk\Contracts\RuleInterface;use Swoole\Coroutine;class RiskEngine{/** @var RuleInterface[]*/ private array$rules=[];publicfunctionaddRule(RuleInterface$rule): self{$this->rules[]=$rule;return$this;}publicfunctionevaluate(RiskContext$ctx): RiskDecision{$results=[];// 协程并行执行所有规则$wg=new Coroutine\WaitGroup();$channel=new Coroutine\Channel(count($this->rules));foreach($this->rules as$rule){$wg->add();Coroutine::create(function()use($rule,$ctx,$channel,$wg){try{$channel->push($rule->evaluate($ctx));}finally{$wg->done();}});}$wg->wait();$channel->close();while(($result=$channel->pop(0))!==false){$results[]=$result;}return$this->aggregate($results);}privatefunctionaggregate(array$results): RiskDecision{$totalScore=0;$decision=RiskResult::PASS;$reasons=[];foreach($resultsas$result){$totalScore+=$result->score;if($result->decision===RiskResult::BLOCK){$decision=RiskResult::BLOCK;$reasons[]=$result->reason;}elseif($result->decision===RiskResult::REVIEW&&$decision!==RiskResult::BLOCK){$decision=RiskResult::REVIEW;$reasons[]=$result->reason;}}returnnew RiskDecision($decision, min($totalScore,100),$reasons);}}// app/Risk/RiskDecision.php namespace App\Risk;class RiskDecision{publicfunction__construct(publicreadonlystring$decision, publicreadonlyint$score, publicreadonlyarray$reasons,){}publicfunctionisBlocked(): bool{return$this->decision===RiskResult::BLOCK;}publicfunctionneedsReview(): bool{return$this->decision===RiskResult::REVIEW;}}---6. 中间件接入 // app/Middleware/RiskMiddleware.php namespace App\Middleware;use App\Risk\RiskContext;use App\Risk\RiskEngine;use App\Risk\Rules\BlacklistRule;use App\Risk\Rules\IpFrequencyRule;use App\Risk\Rules\UserVelocityRule;use Hyperf\HttpServer\Contract\RequestInterface;use Psr\Http\Message\ResponseInterface;use Psr\Http\Message\ServerRequestInterface;use Psr\Http\Server\MiddlewareInterface;use Psr\Http\Server\RequestHandlerInterface;class RiskMiddleware implements MiddlewareInterface{publicfunction__construct(private RequestInterface$request, private BlacklistRule$blacklistRule, private IpFrequencyRule$ipFrequencyRule, private UserVelocityRule$userVelocityRule,){}publicfunctionprocess(ServerRequestInterface$request, RequestHandlerInterface$handler): ResponseInterface{$ctx=new RiskContext(userId:(string)$this->request->input('user_id',''), ip:$this->request->getServerParams()['remote_addr']??'', action:$this->resolveAction(), amount:(float)$this->request->input('amount',0),);$engine=(new RiskEngine())->addRule($this->blacklistRule)// 黑名单优先 ->addRule($this->ipFrequencyRule)->addRule($this->userVelocityRule);$decision=$engine->evaluate($ctx);if($decision->isBlocked()){returnresponse()->json(['code'=>403,'message'=>'请求被风控拦截','reasons'=>$decision->reasons,],403);}// 注入决策供后续使用$request=$request->withAttribute('risk_decision',$decision);return$handler->handle($request);}privatefunctionresolveAction(): string{$path=$this->request->getUri()->getPath();returnmatch(true){str_contains($path,'/pay')=>'pay', str_contains($path,'/withdraw')=>'withdraw', str_contains($path,'/login')=>'login', default=>'default',};}}---7. 注册中间件 // config/autoload/middlewares.phpreturn['http'=>[App\Middleware\RiskMiddleware::class,],];--- 扩展方向 ┌────────────────┬──────────────────────────────────────────────┐ │ 需求 │ 方案 │ ├────────────────┼──────────────────────────────────────────────┤ │ 规则动态配置 │ 规则存 MySQL/Redis，运行时热加载 │ ├────────────────┼──────────────────────────────────────────────┤ │ 机器学习评分 │ 调用 Python 微服务（gRPC），结果注入 score │ ├────────────────┼──────────────────────────────────────────────┤ │ 人工审核队列 │ hyperf/async-queue 推送待审核事件 │ ├────────────────┼──────────────────────────────────────────────┤ │ 规则 A/B 测试 │ 按 userIdhash分流不同规则集 │ ├────────────────┼──────────────────────────────────────────────┤ │ 接入第三方风控 │ 同步调 SEON / 同盾 API，结果合并到 aggregate │ └────────────────┴──────────────────────────────────────────────┘ --- 核心思路：黑名单秒拒 → 频率规则 → 金额规则，三层并行跑完合并分数，整个链路控制在 5ms 以内。