分享一个 workerman实现UDP同端口接收和主动发送数据的实践经验

workerman作为一个基于PHP的实时通信框架，开发高效、业务稳定，深受大家喜欢。支持TCP服务器端、客户端，UDP通信。

我认为workerman长期以来缺乏某一项能力：如何将workerman同时作为同一端口的UDP服务端和客户端，这应该是一个编程环境基本的功能，但是官方教程里面却没有。即：监听某个特定端口(如5060)，能否既正常使用 onMessage 接收数据，又能使用监听的这个端口作为源端口，使用sendto主动向特定的某个IP、端口主动发送数据？而不是等等别人连接以后才能回复数据？

这个应用场景很多呀，比如SIP网关（GB28181通信）、比如开发某些特定的UDP应用。

长期以来有人在workerman的官网上提问这种场景要如何解决，但是维护方并没有给出直接的答复。
在AI的帮助下我实现了这个功能特意分享出来，供大家参考。

workerman 5.2.3     swoole 6.1.4    PHP 8.4.14

接下来直接上代码

<?php
chdir(dirname($_SERVER['SCRIPT_FILENAME']));

include_once __DIR__ . '/vendor/autoload.php';
include_once ("./Lib_global.php");

use Workerman\Worker;
use Workerman\Timer;

// 轻量的 UDP 连接上下文，模拟 Workerman 的 UdpConnection
class SimpleUdpConnection {
    public $socket;
    private $remoteIp;
    private $remotePort;

    public function __construct($socket, $ip, $port) {
        $this->socket     = $socket;
        $this->remoteIp   = $ip;
        $this->remotePort = $port;
    }

    public function getRemoteIp()   { return $this->remoteIp; }
    public function getRemotePort() { return $this->remotePort; }

    // 回复给来源方（源端口仍是 5060）
    public function send($data) {
        return @stream_socket_sendto( $this->socket, $data,  0, "{$this->remoteIp}:{$this->remotePort}");
    }
}

$worker = new Worker();
$worker->count = 2;
$processName = "sip_udp_server";
$worker->name = $processName;
$worker->reusePort = true;

Worker::$pidFile    = ROOT_PATH."var/{$processName}.pid";
Worker::$logFile    = ROOT_PATH."var/{$processName}_logFile.log";
Worker::$stdoutFile = ROOT_PATH."var/{$processName}_stdout.log";

$udp_socket = "";

function app_log($log , $fileName = ""){
    global $udp_socket , $processName;
    $name = $processName;
    if(strlen($fileName) > 0 ){
        $name = $processName."_".$fileName;
    }
    logToFile($log , "./Log/".date("Ymd")."_{$name}.txt");
}

function logToFile($message , $logName_s =""){
    $msg = date("Y-m-d H:i:s")." ".$message."\n";
    WriteLog($msg , $logName_s);
}

function WriteLog($msg , $logName_s =""){
    global $config , $worker;

    if($logName_s != ""){
        $logName = $logName_s;
    }else{
        if( isset($worker) ){
            $str_worker_id = "_".$worker->id;
        }else{
            $str_worker_id = "";
        }

        if( file_exists($config['logPath']) == false ){
            mkdir($config['logPath']);
        }

        $logName = $config['logPath'].date("Y-m-d")."_workerDaemon{$str_worker_id}.txt";
    }

    if( file_exists($logName) == false ){
        //file_put_contents( $logName , date("Y-m-d") . " 创建文件\n");
    }

    file_put_contents( $logName , $msg , FILE_APPEND );
}

/**
 * 拼接 "IP:port"，IPv6 自动加方括号
 */
function format_udp_address($ip, $port) {
    if (strpos($ip, ':') !== false) {
        return "[{$ip}]:{$port}";
    }
    return "{$ip}:{$port}";
}

/**
 * 解析 stream_socket_recvfrom 返回的来源地址
 * IPv4: 192.168.1.5:5060
 * IPv6: [fe80::1]:5060
 */
function parse_udp_address($address) {
    if ($address === '') {
        return ['', 0];
    }
    if ($address[0] === '[') {
        $pos = strpos($address, ']');
        if ($pos === false) {
            return ['', 0];
        }
        return [substr($address, 1, $pos - 1), (int) substr($address, $pos + 2)];
    }
    $pos = strrpos($address, ':');
    if ($pos === false) {
        return ['', 0];
    }
    return [substr($address, 0, $pos), (int) substr($address, $pos + 1)];
}

/**
 * 主动发送 UDP 数据（源端口为 $udp_socket 绑定的端口）
 * 支持 IPv4/IPv6、单播/广播/组播
 */
function udp_server_sendto($targetIp, $targetPort, $data) {
    global $udp_socket;

    if (empty($udp_socket)) {
        app_log("udp_server_sendto: udp_socket 未初始化");
        return false;
    }

    if (!filter_var($targetIp, FILTER_VALIDATE_IP)) {
        app_log("udp_server_sendto: 非法目标 IP -> {$targetIp}");
        return false;
    }
    $targetPort = (int)$targetPort;
    if ($targetPort < 1 || $targetPort > 65535) {
        app_log("udp_server_sendto: 非法目标端口 -> {$targetPort}");
        return false;
    }
    if (!is_string($data) || $data === '') {
        app_log("udp_server_sendto: 空数据");
        return false;
    }

    //if (strlen($data) > 8192) {
    //    app_log("udp_server_sendto: 数据超业务上限 " . strlen($data) . " 字节");
    //   return false;
    //}

    $address = format_udp_address($targetIp, $targetPort);
    $sent = @stream_socket_sendto($udp_socket, $data, 0, $address);

    if ($sent === false) {
        $err = error_get_last();
        app_log("udp_server_sendto: 发送失败 -> {$address} : " . ($err['message'] ?? 'unknown'));
        return false;
    }

    return $sent;
}

$worker->onWorkerStart = function () use ($worker) {
    global $processName, $udp_socket;

    $context = stream_context_create([
        'socket' => [
            'so_reuseaddr' => true,
            'so_broadcast' => true,   // 关键：允许发广播
        ],
    ]);

    //双栈监听（IPv4 + IPv6）
    $udp_socket = stream_socket_server(
        'udp://[::]:5060',
        $errno,
        $errstr,
        STREAM_SERVER_BIND,
        $context
    );

    //接收组播数据
    //$raw = socket_import_stream($udp_socket);
    //if ($raw !== false) {
        // IPv4 加入组播组
        //$mreq = inet_pton('239.255.0.1') . inet_pton('0.0.0.0'); // group + interface
        //@socket_set_option($raw, IPPROTO_IP, IP_ADD_MEMBERSHIP, $mreq);

        // IPv6 加入组播组（对应 IPPROTO_IPV6 / IPV6_JOIN_GROUP）
        // $mreq6 = inet_pton('ff02::1') . pack('V', 0);
        // @socket_set_option($raw, IPPROTO_IPV6, IPV6_JOIN_GROUP, $mreq6);
    //}

    //组播发送
    // 可选：调整 IPv4 组播 TTL 和出口网卡
    //$raw = socket_import_stream($udp_socket);
    //if ($raw !== false) {
        // TTL 调为 4（默认 1，仅本子网）
        //@socket_set_option($raw, IPPROTO_IP, IP_MULTICAST_TTL, 4);
        // 指定出口网卡（按需）
        // @socket_set_option($raw, IPPROTO_IP, IP_MULTICAST_IF, inet_pton('192.168.1.10'));
    //}

    if (!$udp_socket) {
        throw new \Exception("stream_socket_server 失败: {$errstr} ({$errno})");
    }

    stream_set_blocking($udp_socket, false);

    $worker::$globalEvent->onReadable($udp_socket, function () use ($udp_socket, $worker) {
        $address = '';
        $buf = stream_socket_recvfrom($udp_socket, 65535, 0, $address);

        if ($buf === false || $buf === '') {
            return;
        }

        // 业务上限保护，按需调整
        //if (strlen($buf) > 4096) {
        //    app_log("丢弃超大包: " . strlen($buf) . " 字节，来自 {$address}");
        //    return;
        //}

        // 统一解析来源（兼容 IPv4 / IPv6）
        list($fromIp, $fromPort) = parse_udp_address($address);

        if ($fromIp === '') {
            app_log("无法解析来源地址: {$address}");
            return;
        }

        if (isset($worker->onMessage) && is_callable($worker->onMessage)) {
            $connection = new SimpleUdpConnection($udp_socket, $fromIp, $fromPort);
            call_user_func($worker->onMessage, $connection, $buf);
        }
    });

    // 主动发送示例
    Timer::add(1, function () {
        udp_server_sendto('192.168.1.7', 5060, "Proactive Heartbeat from 5060\n");
    });
};

// ✅ 这里定义 onMessage，和普通 Workerman Worker 写法一致
$worker->onMessage = function (SimpleUdpConnection $connection, $data) {
    $fromIp   = $connection->getRemoteIp();
    $fromPort = $connection->getRemotePort();

    echo "收到来自 {$fromIp}:{$fromPort} 的数据: {$data}\n";

    udp_server_sendto($fromIp, $fromPort, "data got ，reply\n");

    // 业务处理……
    // 需要回复时：
    // $connection->send("reply from 5060");
};

Worker::runAll();