swoole4应用之协程并发和多进程实践

背景

某租车平台，在用户请求搜索车辆时，需要先去请求租车公司的接口，拿到对应的套餐报价回来，然后再进行统一的根据具体规则进行业务加装，例如加装一些优惠券，评分和芝麻标示等，然后返回结果给前端。

分析

• 上诉过程，由于考虑请求的供应商是大量的，假设一次二十个供应商，然后每个供应商有1千的报价返回，就有2万报价数据返回。然后在自身业务加装的时候，也不单单只有一两个业务，可能数十个。所以==每一次用户请求直接curl所有的供应商接口，拿到数据后遍厉加装，入库，返回。这样的方法是行不通的，不仅仅整个接口有可能被供应商的API超时拖垮，还可能打满整个机器的fpm进程，从而拖垮整站。==
• 目前我们应用的方案是：
• 上面方案先撇开队列的相关的东西，就看数据入库后再读取这里分析如何优化：
  1. 每次读取都是一次性读取一次搜索产生的所有套餐
  2. 所有业务逻辑加装都是串行，在一个进程处理。

     优化

• 协程或者多线程并发读取数据
• 多进程或者线程加装独立的业务逻辑 若用多线程处理，需换语言，如Java，但需学习新语言，这样后期维护难，成本高。若直接全部用php的多进程，一个进程切换和销毁销耗大，且封装不好。所有决定用swoole扩展，用协程的并发请求去拉起数据，然后用进程池去加装套餐业务。进程间的数据共享通过共享内存解决。大致的流程demo代码如下（例子中协程拉取数据部分是用http方式，实际应为直接读取mongo）：

<?php


use Swoole\Process;

Co::set(['hook_flags' => SWOOLE_HOOK_ALL]);//使用php一些底层函数一键协程化，产生io调度，不包括CURL

//协程并发，分页获取车型列表数据
$carList = [];
Co\run(function () use (&$carList) {

	$wg = new \Swoole\Coroutine\WaitGroup();
	$carList = $result = [];

	$wg->add();
	//启动第一个协程
	go(function () use ($wg, &$result) {
		$result[0] = getCarList(0, 1);
		echo 'co[1] end' . PHP_EOL;
		$wg->done();
	});

	$wg->add();
	go(function () use ($wg, &$result) {
		$result[1] = getCarList(2, 3);
		echo 'co[2] end' . PHP_EOL;
		$wg->done();
	});

	$wg->add();
	go(function () use ($wg, &$result) {
		$result[2] = getCarList(4, 5);
		echo 'co[3] end' . PHP_EOL;
		$wg->done();
	});

	//挂起当前协程，等待所有任务完成后恢复组装数据
	$wg->wait();
	foreach ($result as $item) {
		$carList = array_merge($carList, $item);
	}
});

//多进程处理获取后的车型列表数据
$table = new Swoole\Table(1024);
$table->column('clid', Swoole\Table::TYPE_STRING, 16);
$table->column('id', Swoole\Table::TYPE_STRING, 16);
$table->column('sesame', Swoole\Table::TYPE_INT, 8);
$table->column('easy', Swoole\Table::TYPE_INT, 8);
$table->column('coupon', Swoole\Table::TYPE_INT, 8);
$table->create();//申请内存

$startTime = time();

for ($n = 1; $n <= 3; $n++) {
	$process = new Process(function () use ($n, $table, $carList) {
		switch ($n) {
			case 1:
				handleSesame($carList, $table);
				break;
			case 2:
				handleEasy($carList, $table);
				break;
			case 3:
				handleCoupon($carList, $table);
				break;
		}
	});
	$process->start();
}

for ($n = 3; $n--;) {
	$status = Process::wait(true);
	echo "Recycled #{$status['pid']}, code={$status['code']}, signal={$status['signal']}" . PHP_EOL;
}
echo "used time " . (time() - $startTime) . 'S' . PHP_EOL;
foreach ($carList as $carDetail) {
	var_dump($table->get(getKey($carDetail)));
	$table->del(getKey($carDetail));
}


exit;

//处理芝麻业务
function handleSesame($carList = [], $table = null)
{
	echo 'Child #' . getmypid() . " start to handle Sesame" . PHP_EOL;
	foreach ($carList as $carDetail) {
		$carDetail['sesame'] = 1;
		$table->set(getKey($carDetail), $carDetail);
	}
}

//处理闪租业务
function handleEasy($carList = [], $table = null)
{
	echo 'Child #' . getmypid() . " start to handle Easy" . PHP_EOL;
	foreach ($carList as $carDetail) {
		//sleep(1);
		$carDetail['easy'] = 1;
		$table->set(getKey($carDetail), $carDetail);
	}
}

//处理优惠券业务
function handleCoupon($carList = [], $table = null)
{
	echo 'Child #' . getmypid() . " start to handle Coupon" . PHP_EOL;
	foreach ($carList as $carDetail) {
		$carDetail['coupon'] = 1;
		$table->set(getKey($carDetail), $carDetail);
	}
}

//唯一key
function getKey($carDetail = [])
{
	return "{$carDetail['clid']}-{$carDetail['id']}";
}

function getCarList($start = 0, $end = 0)
{
	if ($end <= $start) {
		return null;
	}
	$data = file_get_contents("http://127.0.0.1:8089/getCarList.php?start={$start}&end=$end");
	$data = json_decode($data, true);
	return $data['data'];
}

打赏