PHP 实现 Redis 高并发解决方案
PHP 实现 Redis 高并发解决方案
使用 PHP 实现 Redis 在缓存加速、分布式锁和队列场景中的应用。
首先确保已安装 PHP Redis 扩展
一、缓存加速实现
1. 基本缓存操作
1<?php
2$redis = new Redis();
3$redis->connect('127.0.0.1', 6379);
4
5// 设置缓存
6function setCache($key, $value, $expire = 3600) {
7 global $redis;
8 $serialized = serialize($value);
9 return $redis->setex($key, $expire, $serialized);
10}
11
12// 获取缓存
13function getCache($key) {
14 global $redis;
15 $serialized = $redis->get($key);
16 return $serialized ? unserialize($serialized) : false;
17}
18
19// 删除缓存
20function deleteCache($key) {
21 global $redis;
22 return $redis->del($key);
23}
24
25// 示例：用户数据缓存
26function getUser($userId) {
27 $cacheKey = "user:{$userId}";
28 $user = getCache($cacheKey);
29
30 if ($user === false) {
31 // 模拟数据库查询
32 $user = [
33 'id' => $userId,
34 'name' => 'User ' . $userId,
35 'email' => "user{$userId}@example.com"
36 ];
37 // 写入缓存，有效期1小时
38 setCache($cacheKey, $user, 3600);
39 }
40
41 return $user;
42}
43
44// 使用示例
45$user = getUser(123);
46print_r($user);
2. 防止缓存穿透
1function getProduct($productId) {
2 global $redis;
3 $cacheKey = "product:{$productId}";
4 $product = getCache($cacheKey);
5
6 if ($product === false) {
7 // 使用互斥锁防止缓存击穿
8 $lockKey = "lock:product:{$productId}";
9 $locked = $redis->set($lockKey, 1, ['nx', 'ex' => 10]);
10
11 if ($locked) {
12 try {
13 // 模拟数据库查询
14 $product = [
15 'id' => $productId,
16 'name' => 'Product ' . $productId,
17 'price' => rand(100, 1000)
18 ];
19
20 if (empty($product)) {
21 // 缓存空对象防止穿透，有效期5分钟
22 setCache($cacheKey, [], 300);
23 } else {
24 setCache($cacheKey, $product, 3600);
25 }
26 } finally {
27 $redis->del($lockKey);
28 }
29 } else {
30 // 等待其他进程完成缓存设置
31 usleep(500000); // 等待500ms
32 return getProduct($productId); // 重试
33 }
34 }
35
36 return $product ?: null;
37}
二、 分布式锁实现
1. 基本分布式锁
1class RedisLock {
2 private $redis;
3 private $lockKey;
4 private $identifier;
5 private $lockTimeout;
6
7 public function __construct($redis, $lockKey, $lockTimeout = 10) {
8 $this->redis = $redis;
9 $this->lockKey = $lockKey;
10 $this->lockTimeout = $lockTimeout;
11 $this->identifier = uniqid();
12 }
13
14 public function acquire($waitTimeout = 5) {
15 $end = microtime(true) + $waitTimeout;
16
17 while (microtime(true) < $end) {
18 if ($this->redis->set(
19 $this->lockKey,
20 $this->identifier,
21 ['nx', 'ex' => $this->lockTimeout]
22 )) {
23 return true;
24 }
25
26 usleep(10000); // 等待10ms
27 }
28
29 return false;
30 }
31
32 public function release() {
33 $script = '
34 if redis.call("get", KEYS[1]) == ARGV[1] then
35 return redis.call("del", KEYS[1])
36 else
37 return 0
38 end
39 ';
40
41 return $this->redis->eval($script, [$this->lockKey, $this->identifier], 1);
42 }
43
44 public function __destruct() {
45 $this->release();
46 }
47}
48
49// 使用示例
50$lock = new RedisLock($redis, 'lock:order:123');
51if ($lock->acquire()) {
52 try {
53 // 执行需要加锁的操作
54 echo "Lock acquired, doing critical section...\n";
55 sleep(2);
56 } finally {
57 $lock->release();
58 }
59} else {
60 echo "Failed to acquire lock\n";
61}
2. 可重入锁实现
1class ReentrantRedisLock extends RedisLock {
2 private $heldCount = 0;
3
4 public function acquire($waitTimeout = 5) {
5 // 检查是否已经持有锁
6 if ($this->heldCount > 0) {
7 $this->heldCount++;
8 return true;
9 }
10
11 $acquired = parent::acquire($waitTimeout);
12 if ($acquired) {
13 $this->heldCount = 1;
14 }
15 return $acquired;
16 }
17
18 public function release() {
19 if ($this->heldCount > 1) {
20 $this->heldCount--;
21 return true;
22 }
23
24 $released = parent::release();
25 if ($released) {
26 $this->heldCount = 0;
27 }
28 return $released;
29 }
30}
三、队列实现
1. 简单队列
1// 生产者
2function enqueue($queueName, $data) {
3 global $redis;
4 return $redis->lPush($queueName, json_encode($data));
5}
6
7// 消费者
8function dequeue($queueName, $timeout = 30) {
9 global $redis;
10 $result = $redis->brPop($queueName, $timeout);
11 return $result ? json_decode($result[1], true) : null;
12}
13
14// 示例
15enqueue('email_queue', [
16 'to' => 'user@example.com',
17 'subject' => 'Welcome',
18 'body' => 'Thank you for registering'
19]);
20
21$task = dequeue('email_queue');
22if ($task) {
23 // 处理任务
24 echo "Sending email to: {$task['to']}\n";
25}
2. 延迟队列
1function enqueueDelayed($queueName, $data, $delaySeconds) {
2 global $redis;
3 $score = time() + $delaySeconds;
4 return $redis->zAdd($queueName, $score, json_encode($data));
5}
6
7function processDelayedQueue($queueName) {
8 global $redis;
9 $now = time();
10 $items = $redis->zRangeByScore($queueName, 0, $now);
11
12 if (!empty($items)) {
13 // 使用事务移除已处理项
14 $redis->multi();
15 foreach ($items as $item) {
16 $redis->zRem($queueName, $item);
17 }
18 $redis->exec();
19
20 return array_map(function($item) {
21 return json_decode($item, true);
22 }, $items);
23 }
24
25 return [];
26}
27
28// 示例
29enqueueDelayed('reminder_queue', ['user_id' => 123, 'message' => 'Pay your bill'], 60);
30
31$tasks = processDelayedQueue('reminder_queue');
32foreach ($tasks as $task) {
33 echo "Sending reminder to user {$task['user_id']}: {$task['message']}\n";
34}
3. Pub/Sub 模式
1// 发布者
2function publishMessage($channel, $message) {
3 global $redis;
4 return $redis->publish($channel, json_encode($message));
5}
6
7// 订阅者
8function subscribe($channels, $callback) {
9 global $redis;
10 $pubsub = $redis->pSubscribe($channels);
11
12 try {
13 foreach ($pubsub as $message) {
14 if ($message->kind === 'message') {
15 $data = json_decode($message->payload, true);
16 call_user_func($callback, $message->channel, $data);
17 }
18 }
19 } catch (Exception $e) {
20 $redis->pUnsubscribe();
21 throw $e;
22 }
23}
24
25// 示例使用
26// 在一个进程中
27publishMessage('notifications', ['user_id' => 123, 'text' => 'New message']);
28
29// 在另一个进程中
30subscribe(['notifications'], function($channel, $message) {
31 echo "Received on {$channel}: ";
32 print_r($message);
33});
四、综合应用示例 - 秒杀系统
1class SeckillService {
2 private $redis;
3 private $productKey;
4
5 public function __construct($redis, $productId) {
6 $this->redis = $redis;
7 $this->productKey = "seckill:product:{$productId}";
8 }
9
10 public function initInventory($inventory) {
11 // 设置商品库存
12 $this->redis->set($this->productKey, $inventory);
13 }
14
15 public function seckill($userId) {
16 // 使用分布式锁
17 $lock = new RedisLock($this->redis, "lock:{$this->productKey}");
18
19 if (!$lock->acquire(1)) {
20 return ['success' => false, 'message' => '系统繁忙'];
21 }
22
23 try {
24 // 检查库存
25 $inventory = $this->redis->get($this->productKey);
26 if ($inventory <= 0) {
27 return ['success' => false, 'message' => '已售罄'];
28 }
29
30 // 扣减库存
31 $this->redis->decr($this->productKey);
32
33 // 记录订单 (实际应用中应该入队列异步处理)
34 $orderId = uniqid();
35 $this->redis->hSet("seckill:orders", $orderId, json_encode([
36 'user_id' => $userId,
37 'product_key' => $this->productKey,
38 'created_at' => time()
39 ]));
40
41 return ['success' => true, 'order_id' => $orderId];
42 } finally {
43 $lock->release();
44 }
45 }
46}
47
48// 使用示例
49$redis = new Redis();
50$redis->connect('127.0.0.1', 6379);
51
52$seckill = new SeckillService($redis, 1001);
53$seckill->initInventory(100); // 初始化100件库存
54
55// 模拟并发请求
56for ($i = 0; $i < 5; $i++) {
57 $result = $seckill->seckill(rand(1000, 9999));
58 print_r($result);
59}
五、最佳实践建议
1. 连接管理：使用连接池或持久连接
1$redis = new Redis();
2$redis->pconnect('127.0.0.1', 6379);
2. 错误处理：添加重试机制
1function redisRetry($callback, $maxRetries = 3) {
2 $retries = 0;
3 while ($retries < $maxRetries) {
4 try {
5 return $callback();
6 } catch (RedisException $e) {
7 $retries++;
8 if ($retries >= $maxRetries) {
9 throw $e;
10 }
11 usleep(100000 * $retries); // 指数退避
12 }
13 }
14}
3. 性能优化：使用管道(pipeline)
1$pipe = $redis->pipeline();
2$pipe->set('key1', 'value1');
3$pipe->set('key2', 'value2');
4$pipe->incr('counter');
5$pipe->expire('key1', 60);
6$results = $pipe->exec();
4. Lua脚本：保证原子性
1$script = '
2 local current = redis.call("GET", KEYS[1])
3 if current == ARGV[1] then
4 return redis.call("INCRBY", KEYS[1], ARGV[2])
5 else
6 return nil
7 end
8';
9$result = $redis->eval($script, ['counter', 'expected_value', 'increment'], 1);
发表评论