PHP使用MySQL实现消息队列

Posted 2020-10-20 寻觅beyond

消息队列常用在流量削峰（秒杀场景），异步通信等地方。

大体的结构如下：

 　　类似于消费者和生产者的关系，首先生产者在消息队列未满的时候，才将生产的产品放进消息队列中；消费者在消息队列不为空的时候，才从消息队列中取出产品进行消费。出队的那个步骤常用的方法是一直轮询和定时操作。

这里举一个外卖送餐的案例：

　　有个生意很好的饭店，好到什么程度呢？一分钟有500人下单，这样的话，店家掌柜肯定处理不过来，于是，就先暂时不通知用户是够接单，先把所有的订单先存着，只告诉他们正在处理中，但是呢，还有一个问题，就是有一些是其他饭店专门来搞事的（眼红了），所以就要查看订单是否合法。

　　上面的情景可以这样实现：用户下单之后，后台会创建一个随机的order_id对应该订单；并且该订单的初始状态（status）为待处理（0）；当商家查看该订单情况时，将status改为1，表示正在处理；当商家确认这个订单可以接受时，就将该订单的status改为2，表示成功接单。

　　首先在数据库中创建一个order_list订单表，表结构如下：

mysql> desc order_list;
+----------+------------+------+-----+----------+-------+
| Field    | Type       | Null | Key | Default  | Extra |
+----------+------------+------+-----+----------+-------+
| order_id | int(11)    | NO   | PRI | NULL     |       |
| mobile   | int(8)     | NO   |     | 88888888 |       |
| status   | tinyint(1) | YES  |     | 0        |       |
+----------+------------+------+-----+----------+-------+
3 rows in set (0.05 sec)

　　有两个php程序，分别是producer.php（用户、生产者）；     consumer.php（商家、消费者）　　

　　一旦有用户下单，就往order_list中添加一条数据，这里方便测试，于是执行下面的PHP程序，表示用户方（producer.php），每隔2秒下一次单；

<?php 
	$pdo=new PDO("mysql:host=localhost;dbname=test","root","root");
	$stmt=$pdo->prepare("insert into order_list (order_id,mobile,status) values (?,?,?)");
	while(1){
		$order_id=rand(10000,99999);
		$mobile=rand(11111111,99999999);
		$stmt->execute(array($order_id,$mobile,0));
		echo date("Y-m-d H:i:s",time())."添加了一条订单，订单号为{$order_id}，手机号为{$mobile}\\n";
		sleep(2);
	}
 ?>

　　用户提交订单后，剩下的事情就交给商家了，商家（consumer.php）生意太忙了，每4秒才能处理一个订单：

<?php 
	$pdo=new PDO("mysql:host=localhost;dbname=test","root","root");
	$stmt=$pdo->prepare("update order_list set status=? where status=? limit 1");
	$stmt_select=$pdo->prepare("select order_id from order_list where status=1");//正在处理的订单号
	while(1){
		$init=0;//初始status
		$lock=1;//标记为正在处理
		$success=2;//成功接单
		//为了保证数据的一致性，处理订单之前，要先锁定一个订单，将其status由0改为1，然后才可处理
		//处理完毕后，然后再将status从1改为2

		$stmt->execute(array($lock,$init));//锁定要处理的订单
		$stmt_select->execute();
		$result=$stmt_select->fetch(PDO::FETCH_ASSOC);//查询正在处理的订单号
		$order_id=$result[\'order_id\'];
		echo date("Y-m-d H:i:s")."准备处理订单，订单号为{$order_id}\\n";
		sleep(3);//处理3秒
		$stmt->execute(array($success,$lock));
		echo date("Y-m-d H:i:s")."订单处理完成，订单号为{$order_id}\\n";
		sleep(1);//休息1秒
	}
 ?>

　　这样就使用MySQL实现了一个简单的消息队列，可以看一下如何使用Redis实现消息队列，与这个方法类似。需要注意的是，上面的代码其实有很大的问题，比如，如果涉及到并发，数据库中同时操作一条数据怎么办，这个时候，考虑的问题就多了。