阿里云生活物联网架构师专题 ⑨基于TP6 基于AMPQ协议接收阿里云物联网设备的各种状态数据，实现m2m流转；

Posted 2022-02-04 半颗心脏

---

• 本系列博客学习由非官方人员 半颗心脏 潜心所力所写，仅仅做个人技术交流分享，不做任何商业用途。如有不对之处，请留言，本人及时更改。

1、esp32接入阿里云物联网平台，实现天猫精灵语音控制；
2、esp8266直连接入阿里云物联网平台，实现天猫精灵找队友零配网功能和语音控制；
3、esp32 sdk 直连接入天猫精灵IOT开放平台，实现天猫精灵找队友零配网功能和语音控制；
4、如何在天猫精灵IOT开放平台二次开发智能设备的 H5控制面板；
5、分享可商用的ESP8266 SDK连接阿里云物联网生活平台的在线远程升级OTA笔记。
6、ESP8266接入阿里生活飞燕平台国际版，实现亚马逊Alexa Echo音响语音控制。
7、阿里云物联网平台的网关-子设备理论协议说明，支持Zigbee/ble等没上云能力的设备；；
8、基于在 SpringBoot Java 私有云上接收阿里云物联网平台设备的各种状态数据，实现m2m流转；
9、基于TP6 基于AMPQ协议接收阿里云物联网设备的各种状态数据，实现m2m流转；；

文章目录

• 寄言
• 一、前言
• • 什么是服务器订阅？
  • • 适用场景
• 二、集成步骤
• 三、常见问题
• • 另外，不要把我的博客作为学习标准，我的只是笔记，难有疏忽之处，如果有，请指出来，也欢迎留言哈！

寄言

---

       我写过很多物联网控制的博文系列，包括微信公众号、微信小程序控制硬件，私有云对接天猫精灵服务器，抑或是硬件端 esp8266/esp32 等系列博文，这是一个一个专题是写下我们如何在阿里云物联网上全栈开发我们的应用专题，让我们无须企业账号也可以体验设备-云端-App” 的过程；让我们变得更强，一个人担任一个公司的全部职责，全栈开发物联网攻城狮前进；

       我会带领大家轻轻松松地把自己的设备接入天猫精灵，告别 “单机时代”，走进语音控制物联网时代。 有疑问请留言区留言，或者加群大伙们讨论；写总结，写博文不容易，望大家多多体谅！

• 自带资料：
  • git 分布式管理软件的基本使用；
  • 硬件开发：乐鑫 esp8266、esp32模块一个；具备 c 语言基础 ，不需要很熟练；
  • 移动端开发：android 端具备 java 、vue.js 开发语言，AndroidStudio 环境；
  • 服务器端开发：php 开发，熟悉 服务器运行、部署等原理操作；
    

---

一、前言

去年介绍了如何在 SpringBoot 框架上面实现流转阿里云物联网设备信息到私有服务器。今天给大家带来在 TP6 PHP上实现此功能。

什么是服务器订阅？

    私有云可以直接订阅产品下多种类型的消息：设备上报消息、设备状态变化通知、设备生命周期变更、网关发现子设备上报、设备拓扑关系变更等。配置服务端订阅后，物联网平台会将产品下所有设备的已订阅类型的消息转发至您的服务器。

适用场景

• 1、服务端订阅适用于单纯的接收设备数据的场景，并且适用于高并发场景。
• 2、服务端接收产品下全部设备的订阅数据。
• 3、如果您有多个服务器消费同一个产品的订阅消息，消息会随机转发至某个服务器。
• 4、服务端订阅与规则引擎数据流转的使用场景和能力对比，请参见数据流转方案对比。

    今天使用的是 TP6 架构语言，这个也是我擅长开发的架构，所以这篇我给大家介绍下如何集成AMQP高级消息队列协议，配置AMQP服务端订阅后，物联网平台会将产品下所有已订阅类型的消息，通过AMQP通道推送至私有云。

---

    AMQP服务端订阅消息流转流程图：

二、集成步骤

①、首先安装 TP6 , 这里使用 git 克隆最新 TP6 源码；

// 克隆
git clone https://gitee.com/top-think/think
// 安装依赖
cd think && composer install

②、安装阿里云AMPQ的运行环境依赖

// 安装 stomp
composer require stomp-php/stomp-php

②、在 TP6 上实现自定义命令。

php think make:command Aliyun_stomp

这时候会自动生成 app\\command\\Aliyun_stomp.php文件，在其修改内容后如下：

<?php
declare (strict_types=1);
namespace app\\command;
use think\\console\\Command;
use think\\console\\Input;
use think\\console\\input\\Argument;
use think\\console\\input\\Option;
use think\\console\\Output;
use think\\facade\\Config;
use Stomp\\Client;
use Stomp\\Network\\Observer\\Exception\\HeartbeatException;
use Stomp\\Network\\Observer\\ServerAliveObserver;
use Stomp\\StatefulStomp;

class Aliyun_stomp extends Command

    protected function configure()
    
        // 指令配置
        $this->setName('app\\command\\aliyun_stomp')
            ->setDescription('阿里云物联网AMPQ订阅');
    

    protected function execute(Input $input, Output $output)
    
        // 指令输出
        $output->writeln('start');
        //参数说明，请参见AMQP客户端接入说明文档。
        $accessKey = Config::get('app.config.accessKey');
        $accessSecret = Config::get('app.config.accessSecret');
        $consumerGroupId = "DEFAULT_GROUP";
        //iotInstanceId：实例ID。
        $iotInstanceId = Config::get('app.config.iotInstanceId');
        //随意填写
        $clientId = "12346989";
        $timeStamp = round(microtime(true) * 1000);
        //签名方法：支持hmacmd5，hmacsha1和hmacsha256。
        $signMethod = "hmacsha1";
        //userName组装方法，请参见AMQP客户端接入说明文档。
        //若使用二进制传输，则userName需要添加encode=base64参数，服务端会将消息体base64编码后再推送。具体添加方法请参见下一章节“二进制消息体说明”。
        $userName = $clientId . "|authMode=aksign"
            . ",signMethod=" . $signMethod
            . ",timestamp=" . $timeStamp
            . ",authId=" . $accessKey
            . ",iotInstanceId=" . $iotInstanceId
            . ",consumerGroupId=" . $consumerGroupId
            . "|";
        $signContent = "authId=" . $accessKey . "&timestamp=" . $timeStamp;
        //计算签名，password组装方法，请参见AMQP客户端接入说明文档。
        $password = base64_encode(hash_hmac("sha1", $signContent, $accessSecret, $raw_output = TRUE));
        //接入域名，请参见AMQP客户端接入说明文档。下方 123456 替换为你的阿里云账号id， cn-shanghai 替换为你的地区代码  若是PHP开发，端口号是  61614
        $client = new Client('ssl://'.Config::get('app.config.accountId').'.iot-amqp.cn-shanghai.aliyuncs.com:61614');
        $sslContext = ['ssl' => ['verify_peer' => true, 'verify_peer_name' => false],];
        $client->getConnection()->setContext($sslContext);
        //服务端心跳监听。
        $observer = new ServerAliveObserver();
        $client->getConnection()->getObservers()->addObserver($observer);
        //心跳设置，需要云端每50s发送一次心跳包。
        $client->setHeartbeat(0, 5000);
        $client->setLogin($userName, $password);
        try 
            $client->connect();
         catch (StompException $e) 
            echo "failed to connect to server, msg:" . $e->getMessage(), PHP_EOL;
        
        //无异常时继续执行。
        $stomp = new StatefulStomp($client);
        $stomp->subscribe('/topic/#');
        $output->writeln('connect success .');
        while (true) 
            try 
                // 检查连接状态
                if (!$client->isConnected()) 
                    echo "connection not exists, will reconnect after 10s.", PHP_EOL;
                    sleep(10);
                    $client->connect();
                    $stomp->subscribe('/topic/#');
                    echo "connect success.", PHP_EOL;
                
                $msg = $stomp->read();
                if (($msg) != "") 
                    // 处理消息业务逻辑。
                    $output->writeln("Read Topic:" . $msg->getHeaders()["topic"]);
                    $output->writeln("Read PayLoad:" . $msg->getBody());
                
             catch (HeartbeatException $e) 
                echo 'The server failed to send us heartbeats within the defined interval.', PHP_EOL;
                $stomp->getClient()->disconnect();
             catch (Exception $e) 
                echo 'process message occurs error: ' . $e->getMessage(), PHP_EOL;
                $stomp->getClient()->disconnect();
            
        
    

其次，在app\\config\\console修改指令定义，指向具体的文件：

<?php
return [
    // 指令定义
    'commands' => [
    	'start_ampq' => 'app\\command\\Aliyun_stomp',
    ],
];

其次，代码部分结束， 切换到命令行工具，定位到我们的项目根目录(think文件所在的目录)，执行：

php think start_ampq

如设备端上报信息，控制台就会打印出来了。

三、常见问题

• Q1：为什么无法执行自定义指令？

答：首先要保证你在app\\config\\console修改指令定义，指向具体的文件是正确的。可以通过 php think list获取支持指令列表。

• Q2：配置参数从哪里获取？

答：参考 https://help.aliyun.com/document_detail/30594.html

• Q3：如何守护进程执行此自定义命令，永久执行？

答：使用 nohup php think start_ampq &

若要停止，先查到其进程 ps -ef | grep php

这时候，开始抹杀掉 kill -9 5088即可。

另外，不要把我的博客作为学习标准，我的只是笔记，难有疏忽之处，如果有，请指出来，也欢迎留言哈！

• 玩转无线物联网带你飞、免费加千人群讨论，群里众多国内ESP开发者，找到你一份归属，免费白嫖。
• QQ讨论一群，点击加群：434878850
• QQ讨论二群，点击加群：623325168
• 个人微信公众号：徐宏blog ， 不定时推送干货文章，不推送任何广告。
• 个人邮箱：[e-mail] 24小时在线，有发必回复！
• esp8266源代码学习汇总（持续更新，欢迎star）：https://github.com/xuhongv/StudyInEsp8266
• esp32源代码学习汇总（持续更新，欢迎star）：https://github.com/xuhongv/StudyInEsp32
• 关注下面微信公众号二维码，干货多多，第一时间推送！