php_async_worker/transfer1b.php

<?php
define('CONFIG_FILE', __DIR__ . '/config.php');
define('LOG_NAME', 'transfer1b');
define('DGRAM_MAXSIZE', 1000); //理论上限65507，内网建议值1472，公网建议值548

include __DIR__ . '/functions.php';
include __DIR__ . '/class.async.php';

//加载配置
$cfg = [];
load_config();

//检查参数
$addr = "{$cfg['udp_host']}:{$cfg['udp_port']}";
$quit = $restart = false;

//保障只运行一个进程
$lockFile = run_single_instance($cfg['udp_port']);

//设置信号处理函数
pcntl_signal(SIGTERM, 'sig_handle', false);
pcntl_signal(SIGUSR1, 'sig_handle', false);
pcntl_signal(SIGUSR2, 'sig_handle', false);

//创建UDP服务
$s = stream_socket_server("udp://{$addr}", $errno, $error, STREAM_SERVER_BIND);
if (!$s) {
	debug_log("Transfer@{$addr} stream_socket_server() call failed: #{$errno},{$error}", 'ERROR', LOG_NAME);
	exit(1);
}

//接收UDP请求
try {
	$async = new Async($cfg['redis_host'], $cfg['redis_port'], $cfg['worker_num']);
	debug_log("Transfer@{$addr} started, pid = " . getmypid(), 'DEBUG', LOG_NAME);
	while (1) {
		$buffer = '';
		declare (ticks = 1) {
			$buffer = stream_socket_recvfrom($s, DGRAM_MAXSIZE, 0, $from);
		}
		if (empty($buffer)) {
			if ($quit) { //退出
				debug_log("Transfer@{$addr} quiting", 'WARNING', LOG_NAME);
				fclose($s);
				unlink($lockFile);
				exit(0);
			} elseif ($restart) { //重启
				fclose($s);
				unlink($lockFile);
				debug_log("Transfer@{$addr} restarting", 'WARNING', LOG_NAME);
				pcntl_exec($cfg['php_bin_path'], $argv);
				debug_log("Transfer@{$addr} restart failed", 'ERROR', LOG_NAME);
				exit(1);
			} else { //定时器
				$async->ping();
			}
		} else {
			$data = json_decode($buffer, true);
			if (!empty($data) && is_array($data)) {
				$async->pushJob($data);
			}
		}
	}
} catch (RedisException $e) {
	debug_log("Transfer@{$addr} catched an RedisException:" . $e->getMessage(), 'EXCEPTION', LOG_NAME);
}
fclose($s);
unlink($lockFile);

/**
 * 信号处理函数
 * @param int $signo 信号
 */
function sig_handle($signo)
{
	global $quit, $restart;
	switch ($signo)
	{
		//退出
		case SIGTERM:
			$quit = true;
			break;
		//重启
		case SIGUSR1:
			$restart = true;
			break;
		//定时器
		case SIGUSR2:
			break;
	}
}