如何在一个进程中，使多个协程共用一个连接 #3024

wuxian12 · 2020-12-25T09:31:23Z

wuxian12
Dec 25, 2020

使用 mqtt 协程客户端包 simps/mqtt 时，会出现下述报错

PHP Fatal error:  Uncaught Swoole\Error: Socket#47 has already been bound to another coroutine#2, reading of the same socket in coroutine#3 at the same time is not allowed in /www/wwwroot/Y3/cloud-chat/vendor/simps/mqtt/src/Client.php:205

以下是使用源码

在进程启动时，启动mqtt

public function listen(): array
    {
        // 返回一个该监听器要监听的事件数组，可以同时监听多个事件
        return [
            MainWorkerStart::class,
            OtherWorkerStart::class,
        ];
    }

    /**
     * @param UserRegistered $event
     */
    public function process(object $event)
    {
        //co(function () {
            $mqtt = ApplicationContext::getContainer()->get(MQTTPublish::class);
            $mqtt->reconnect();
        //});
        
    }

通过http请求给主题发送数据

class IndexController extends AbstractController
{

    /**
     * @Inject
     * @var ChatLogDao
     */
    protected $ldao;


    public function index()
    {
      
        $mqtt = $this->container->get(MQTTPublish::class);
        //return $mqtt->request('user.login',['mobile'=>$this->request->input('mobile'),'proxy'=>$mqtt->getProxy()]);
        $mqtt->sendBinary('test/php', '{"test":"php"}');

Answered by limingxinleo

Dec 27, 2020

#3026

View full answer

limingxinleo · 2020-12-25T09:39:48Z

limingxinleo
Dec 25, 2020
Maintainer

提个建议，代码啥的，提交 Issue 的时候，都好好整理一下。。。

这看着没头没尾的，显得很乱

2 replies

wuxian12 Dec 25, 2020
Author

好的
mqtt的类是对第三方包再次封装了下 simps/mqtt

`<?php

declare (strict_types = 1);

namespace App\MQTT\Publish;

use App\Constants\ErrorCode;
use App\Constants\MqttCode;
use App\Service\Dispatch\MqttDispatchService;
use App\Utils\ApiUtils;
use App\Utils\ServerUtil;
use Hyperf\Utils\ApplicationContext;
use Hyperf\Utils\Codec\Json;
use Simps\MQTT\Client;
use Swoole\Server;
use Swoole\Coroutine;
use App\MQTT\Repositories\MemoryRepository;

abstract class MQTTAbstractPublish
{

protected static $client = null;

//发送主题前缀
public $sendTop = 'sub';

//监听主题前缀
public $subTop = 'pub';

//mqtt客户端名字
public $mqttClient = 'php-mqtt-';

//配置
public $config;

/** @var MemoryRepository */
protected static $repository;

protected $inChan;
protected $outChan;

const QOS_AT_MOST_ONCE  = 0;
const QOS_AT_LEAST_ONCE = 1;
const QOS_EXACTLY_ONCE  = 2;

//创建socket 并连接
public function __construct()
{
    
    if (self::$repository == null) {
        self::$repository = new MemoryRepository();
        
    }
    
}

//socket连接销毁
public function __destruct()
{
    $clientId = $this->config['client_id'];
    //下线通知
    foreach (config('account') as $v) {
        $this->send($v['cate'] . $this->sendTop, 'admin.imAlive', ['serverId' => $clientId, 'code' => 400, 'connected' => false, 'errMsg' => 'Abnormal Closed']);
    }
}

//重连
public function reconnect()
{
    if (self::$client == null) {
        $this->config = $config = $this->getConfig();
        $swooleConfig = $this->getSwConfig();
        self::$client = new Client($config, $swooleConfig, SWOOLE_SOCK_TCP | SWOOLE_SSL);
        while (!self::$client->connect()) {
            if(config('mqtt.auto_connect') > 0){
                //自动重连
                Coroutine::sleep(intval(config('mqtt.auto_connect')));
                self::$client->connect();
            }
        }

        $clientId = $this->config['client_id'];
    
        //发送上线广播
        foreach (config('account') as $v) {
            $this->send($v['cate'] . $this->sendTop, 'admin.imAlive', ['serverId' => $clientId, 'code' => 200, 'connected' => true, 'errMsg' => $clientId . ' is connected']);
        }
        
        co(function () {
            $this->loop();
        });
        
    }
    
   
    
}

//生成配置
public function getConfig()
{
    return [
        'host' => config('mqtt.host'),
        'port' => config('mqtt.port'),
        'time_out' => config('mqtt.timeout'),
        'user_name' => config('mqtt.username'),
        'password' => config('mqtt.password'),
        'client_id' => Client::genClientID(),
        'keep_alive' => config('mqtt.keeplive'),
    ];
}

//生成swoole配置
public function getSwConfig()
{
    return [
        'open_mqtt_protocol' => true,
        'package_max_length' => 2 * 1024 * 1024,
        'ssl_allow_self_signed' => true,
        'ssl_verify_peer' => false,
        'ssl_cafile' => config('mqtt.certfile'), 
        //'ssl_key_file' => '', 
        //'ssl_cert_file' => '',
    ];
}

//添加监听事件
public function addSubscribe($topic, callable $callback): void
{
    self::$repository->addNewTopicSubscription($topic, $callback);
    self::$client->subscribe($topic, $callback);

}

//取消监听事件
public function unsetSubscribe($topic): void
{
    self::$repository->removeTopicSubscription($topic);
    self::$client->unSubscribe($topic);

}

//开启监听
public function loop(): void
{
    $timeSincePing = time();
    while (true) {
        $buffer = self::$client->recv();
        var_dump($buffer);
        if ($buffer) {
            //空回复
            if($buffer !== true){
                $timeSincePing = time();
            }else if(is_array($buffer)){
                $this->deliverMessage($buffer['topic'] ?? '', $buffer['message'] ?? '');
            }
            
        }
        if ($timeSincePing < (time() - config('mqtt.keeplive'))) {
            $buffer = self::$client->ping();
            if ($buffer) {
                echo 'send ping success' . PHP_EOL;
                $timeSincePing = time();
            } else {
                self::$client->connect();
                Coroutine::sleep(5);
            }
        }
    }
    
}

/**
 * Delivers a published message to subscribed callbacks.
 *
 * @param string $topic
 * @param string $message
 * @return void
 */
protected function deliverMessage(string $topic, string $message): void
{
    $subscribers = self::$repository->getTopicSubscriptionsMatchingTopic($topic);
    foreach ($subscribers as $subscriber) {
        try {
            call_user_func($subscriber->getCallback(), $topic, $message);
        } catch (\Throwable $t) {
            ServerUtil::info('mqtt处理回调消息错误' . $topic, Json::decode($message));
        }
    }
}

/**
 * 发送二进制
 * @param string $pubTopic
 * @param string $data
 * @return void
 */
public function sendBinary(string $pubTopic, string $data)
{
    self::$client->publish($pubTopic, $data, self::QOS_EXACTLY_ONCE);
    
}

/**
 * 发布消息
 * @param string $pubTopic
 * @param string $api
 * @param string $reply_id
 * @param string $subTopic
 * @param mixed $callback
 * @param array $data
 * @param string $type
 * @return void
 */
public function send(string $pubTopic, string $api, array $data = [], string $replyId = '', string $subTopic = '', $callback = null, $raw = '', string $type = 'json')
{
    //先监听
    if ($subTopic) {
        $this->addSubscribe($subTopic, $callback);
    }
    $data = $this->getData($api, $data, $replyId, $raw, $type);
    ServerUtil::info('mqtt推送消息' . $pubTopic, Json::decode($data));
    
    self::$client->publish($pubTopic, $data, self::QOS_EXACTLY_ONCE);
    

}

//组装数据
public function getData($api, $data, $replyId = '', $raw = '', $type = 'json' )
{
    $replyId = empty($replyId) ? ServerUtil::uniqueNum() : $replyId;
    $sj = ServerUtil::getMillisecond();
    $param = [
        'msg_id' => ServerUtil::orderNo(),
        'app_data' => [
            'api' => $api,
            'type' => $type,
            'raw' => $raw, //保留字段
            'data' => !empty($data) ? $data : new \StdClass(),
        ],
        'control' => [
            'id' => $replyId,
            'request_time' => strval($sj),
        ],
    ];
    return Json::encode($param, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
}

/**
 * //发送请求 此接口废弃
 * @param string $pubTopic  发布主题
 * @param string $api  接口
 * @param string $subTopic  监听主题
 * @param mixed $callback  监听逻辑处理
 * @param array $data 发送数据
 * @return array
 */
public function reply(string $pubTopic, string $api, array $data = [], string $subTopic = '', $callback = null, $replyId = ''): array
{
    $replyId = empty($replyId) ? ServerUtil::uniqueNum() : $replyId;
    $container = ApplicationContext::getContainer();
    $this->redis->set('exsis_' . $replyId, 1, ['EX' => 60]); //设置同步返回结果标识
    $this->send($pubTopic, $api, $data, $replyId, $subTopic, $callback);

    $response = $this->redis->bRPop($replyId, 60);
    $this->redis->del($replyId, 'exsis_' . $replyId);
    // Pop timeout
    if (empty($response)) {
        return $this->overTime($api, $replyId);
    }

    return Json::decode($response[1]);

}
//超时处理
public function overTime($api, $replyId)
{
    return [
        'errCode' => 502,
        'msg' => '终端网络超时',
        'data' => [],
    ];
}


/**
 * 获取主题
 * @param string $name
 * @param array $clientId
 * @param array $mobile
 * @return mixed
 */
public function getTopic(string $clientId, string $name = '', $mobile = '', $accountType = 1)
{
    $cate = config('account')[$accountType]['cate'];
    if (empty($mobile)) {
        return $cate . $clientId . '/' . $name;
    } else {
        return $cate . $clientId . '/' . $mobile . '/' . $name;
    }

}


/**
 * 发送请求
 * @param string $api 接口名字
 * @param array $data  请求参数
 * @param string $mobile  手机号
 * @param string $replyId  请求id 
 * @param string $clientId 客户端id
 * @param integer $accountType  账号类型
 * @param mixed $raw  请求额外数据，终端还原样返回
 * @return mixed
 */
public function request(string $api, array $data = [], string $mobile = '', string $replyId = '', string $clientId = '',  $accountType = 1, $raw = '')
{
    //手机号码为空，需要先拿终端id
    if (empty($mobile)) { //登录需要监听开启回调
        $clientId = empty($clientId) ? make(AccountClient::class)->getClientId($accountType) : $clientId;
        $pubTopic = $this->getTopic($clientId, $this->sendTop, '', $accountType);
        $subTopic = $this->getTopic($clientId, $this->subTop, $data['mobile'], $accountType);
        $data['account'] = $data['mobile'];
        //处理监听
        $this->handMonitor($data['mobile'], $clientId, $accountType);
        unset($data['mobile']);
    
        //session登陆仅有一次回调
        $this->send($pubTopic, $api, $data, $replyId, $subTopic, [$this, 'handLogic'],$raw);
        

    } else {
        $clientId = empty($clientId) ? make(AccountClient::class)->checkClientId($mobile, $accountType) : $clientId;
        //重新启动因work进程死掉的监听
        $this->restartSub($mobile, $clientId, $accountType);
        $pubTopic = $this->getTopic($clientId, $this->sendTop, $mobile, $accountType);
        
        $this->send($pubTopic, $api, $data, $replyId,'',null,$raw);
        

    }

}

/**
 * 获取代理
 * @return mixed
 */
public function getProxy()
{
    return $this->proxyIp;

}

public function handLogic(string $topic, string $message)
{
    try {
        $msg = Json::decode($message);
        //$keyval = $this->redis->get('exsis_' . $msg['control']['id']);
        $topic_arr = \explode('/', $topic);
        if (isset($msg['app_data']['data']['channel'])) {
            //进入图片加载
            if ($msg['app_data']['api'] == "user.updateAvatar") {
                return make(MqttDispatchService::class)->handUpdateAvatar($msg['app_data']['data']['channel'], $msg['control']['id']);
            }
            $chunk_id = empty($msg['app_data']['data']['chunkId']) ? 1 : $msg['app_data']['data']['chunkId']; //分块传输
            return $this->handUpload($msg['app_data']['data']['channel'], $msg['control']['id'], $chunk_id, $topic_arr[0]);

        }
        //  else if ($keyval == 1) { //http请求需要同步返回
        //     //监听维护账号状态
        //     make(config('account.cate.' . $topic_arr[0]))->handleAccountStatus($msg['app_data']['data']['code'], '', $topic_arr[2]);
        //     return $this->handAsyn($msg);
        // } 
        else {
            return $this->handSync($msg, $topic_arr);
        }
    } catch (\Throwable $t) {
        ServerUtil::info('mqtt推送错误' . $t->getMessage(), ['line' => $t->getLine(), 'file' => $t->getFile()]);

    }
}

//异常处理消息
abstract public function handSync(array $msg, array $topic_arr);

//处理监听
abstract public function handMonitor(string $mobile, string $clientId,$accountType);

//判断服务死掉，监听重新开启
abstract public function restartSub(string $mobile, string $clientId,$accountType);

public function handAsyn(array $msg)
{
    if ($msg['app_data']['data']['code'] == MqttCode::SUCCESS) {
        //组装数据
        $api = $msg['app_data']['data'];
        unset($api['code'], $api['errMsg']);
    } else {
        $api = [];
    }
    $this->redis->lPush($msg['control']['id'], Json::encode(ApiUtils::mqttToApi($msg['app_data']['data']['code'] ?? 200, $msg['app_data']['data']['errMsg'] ?? '操作成功', '', $api)));
    $this->redis->expire($msg['control']['id'], 60);
    return;
}

}`

limingxinleo Dec 25, 2020
Maintainer

我仔细看了你代码的实现，让我深吸了一口气。。。

我本以为就是一个 Channel 的问题，但你的封装直接让原本简单的问题，复杂程度上升了一个量级

这个代码，我觉得。。你还是重写吧

wuxian12 · 2020-12-25T10:11:23Z

wuxian12
Dec 25, 2020
Author

现在不知道怎么写，能帮我把reconnect这个方法重写下吗，其它的发消息，监听消息我自己写，关键是怎么启动mqtt死循环不断的读数据

1 reply

limingxinleo Dec 25, 2020
Maintainer

你还是跟你组员一起，重新写吧。。

我给你把这块都整好，估计没一两个小时，是做不完的。。

我没那么多充裕的时间来做

wuxian12 · 2020-12-25T10:18:28Z

wuxian12
Dec 25, 2020
Author

能大概的说下思路吗

1 reply

limingxinleo Dec 25, 2020
Maintainer

方案1：

Channel 配合 MQTTClient ，使用 pop push 隔离协程调用

方案2：

自定义进程里实例化 MQTTClient 配合 NSQ 等消息队列，实现同时只会执行一个请求，去掉循环逻辑，请求到了之后判断连接状态

wuxian12 · 2020-12-25T10:24:55Z

wuxian12
Dec 25, 2020
Author

好的。我试试，谢谢

1 reply

limingxinleo Dec 25, 2020
Maintainer

#3026

limingxinleo · 2020-12-27T05:01:54Z

limingxinleo
Dec 27, 2020
Maintainer

#3026

1 reply

wuxian12 Dec 28, 2020
Author

大佬，如果用这种方式，不断读数据的协程是阻塞读，会霸占chan，导致发送数据要等待
`<?php

declare (strict_types = 1);

namespace App\MQTT\Publish;

use App\Constants\ErrorCode;
use App\Constants\MqttCode;
use App\Service\Dispatch\MqttDispatchService;
use App\Utils\ApiUtils;
use App\Utils\ServerUtil;
use Hyperf\Utils\Codec\Json;
use Swoole\Server;
use Simps\MQTT\Client;
use Swoole\Coroutine;
use Swoole\Coroutine\Channel;

abstract class MQTTAbstractPublish
{

protected static $client = null;

protected static $chan = null;

//发送主题前缀
public $sendTop = 'sub';

//监听主题前缀
public $subTop = 'pub';

//mqtt客户端名字
public $mqttClient = 'php-mqtt-';


//创建socket 并连接
public function __construct()
{
    if (self::$chan == null) {
        self::$chan = new Channel();
    }
}

//socket连接销毁
public function __destruct()
{
    $clientId = self::$client->getClientId();
    //下线通知
    foreach (config('account') as $v) {
        $this->send($v['cate'] . $this->sendTop, 'admin.imAlive', ['serverId' => $clientId, 'code' => 400, 'connected' => false, 'errMsg' => 'Abnormal Closed']);
    }
}

public function loop()
{
    while (true) {
        $closure = self::$chan->pop();
        if (! $closure) {
            break;
        }
        $closure->call($this);
    }
}

public function connect()
{
    $cont = new Channel();
    self::$chan->push(function () use ($cont) {
        $config = [
            'host' => '192.168.0.51',
            'port' => 8883,
            'time_out' => 5,
            'user_name' => '',
            'password' => '',
            'client_id' => Client::genClientID(),
            'keep_alive' => 20,
        ];

        $swooleConfig = [
            'open_mqtt_protocol' => true,
            'package_max_length' => 2 * 1024 * 1024,
            'ssl_allow_self_signed' => true,
            'ssl_verify_peer' => false,
            'ssl_cafile' => '/www/wwwroot/Y3/cloud-chat/cert/cacert.pem', // https://test.mosquitto.org/ssl/mosquitto.org.crt
            
        ];

        var_dump('连接------------');
        self::$client = new Client($config, $swooleConfig, SWOOLE_SOCK_TCP | SWOOLE_SSL);
        self::$client->connect();
        
        $cont->push(true);
    });
    $cont->pop();
}

public function subscribe($topics)
{
    $cont = new Channel();
    self::$chan->push(function () use ($cont, $topics) {
        self::$client->subscribe($topics);
        $cont->push(true);
        var_dump('订阅------------');
    });
    $cont->pop();
}


public function subscribeAndSend($subtopics,$topic, $data)
{
    $cont = new Channel();
    self::$chan->push(function () use ($cont, $subtopics, $topic, $data) {
        self::$client->subscribe($subtopics);
        $response = self::$client->publish($topic, $data, 0);
        $cont->push(true);
        var_dump('订阅发送数据------------');
    });
    $cont->pop();
}


public function send($topic, $data)
{
    $cont = new Channel();
    self::$chan->push(function () use ($cont, $topic, $data) {
        $response = self::$client->publish($topic, $data, 0);
        $cont->push(true);
        var_dump('发送数据-----------');
    });
    $cont->pop();
}

public function recv($timeSincePing)
{
    $cont = new Channel();
    self::$chan->push(function () use ($cont, $timeSincePing) {
        $buffer = self::$client->recv();
        var_dump($buffer,'收到信息===========');
        if ($buffer && $buffer !== true) {
            $timeSincePing = time();
        }
        if ($timeSincePing < (time() - 10)) {
            $buffer = self::$client->ping();
            if ($buffer) {
                echo 'send ping success' . PHP_EOL;
                $timeSincePing = time();
            } else {
                self::$client->close();
            }
        }
        $cont->push(1);
    });
    return $cont->pop();
}

}`

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Hyperf Community

如何在一个进程中，使多个协程共用一个连接 #3024

{{title}}

{{editor}}'s edit

{{editor}}'s edit

Replies: 5 comments 6 replies

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

Select a reply

Hyperf Community

如何在一个进程中，使多个协程共用一个连接 #3024

wuxian12 Dec 25, 2020

Replies: 5 comments · 6 replies

limingxinleo Dec 25, 2020 Maintainer

wuxian12 Dec 25, 2020 Author

limingxinleo Dec 25, 2020 Maintainer

wuxian12 Dec 25, 2020 Author

limingxinleo Dec 25, 2020 Maintainer

wuxian12 Dec 25, 2020 Author

limingxinleo Dec 25, 2020 Maintainer

wuxian12 Dec 25, 2020 Author

limingxinleo Dec 25, 2020 Maintainer

limingxinleo Dec 27, 2020 Maintainer

wuxian12 Dec 28, 2020 Author

wuxian12
Dec 25, 2020

Replies: 5 comments 6 replies

limingxinleo
Dec 25, 2020
Maintainer

wuxian12 Dec 25, 2020
Author

limingxinleo Dec 25, 2020
Maintainer

wuxian12
Dec 25, 2020
Author

limingxinleo Dec 25, 2020
Maintainer

wuxian12
Dec 25, 2020
Author

limingxinleo Dec 25, 2020
Maintainer

wuxian12
Dec 25, 2020
Author

limingxinleo Dec 25, 2020
Maintainer

limingxinleo
Dec 27, 2020
Maintainer

wuxian12 Dec 28, 2020
Author