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

[wuxian12]

使用 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"}');

[limingxinleo]

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

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

[wuxian12]

好的
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]

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

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

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

[wuxian12]

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

[limingxinleo]

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

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

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

[wuxian12]

能大概的说下思路吗

[limingxinleo]

方案1：

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

方案2：

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

[wuxian12]

好的。我试试，谢谢

[limingxinleo]

#3026

[limingxinleo]

#3026

[wuxian12]

大佬，如果用这种方式，不断读数据的协程是阻塞读，会霸占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();
}

}`