1.项目介绍

  本项目基于物联量平台远程的视频监控项目，通过MQTT协议实现两个设备间的数据上报与订阅。通过这个项目来演示，两个MQTT设备如何互相订阅，进行消息流转。在阿里云服务器上创建2个设备，分为为设备A和设备B；设备A负责采集本地摄像头画面上传，设备B负责接收设备A上传的数据然后解析显示出来。在阿里云服务器上需要配置云产品流转，让设备A的数据上传后自动发送给设备B。这样就完成了视频画面数据的流转。不过因为阿里云的最大数据限制，每次最大发送10240字节的数据。

1.1 硬件平台

操作系统： Ubuntu18.04
硬件设备： 电脑自带或USB免驱摄像头(V4L2框架)
服务器： 阿里云物联网平台(基于MQTT协议)
图像渲染： GTK2.0

1.2 开发流程

  设备A： 获取摄像头的数据–>缩放成240*320–>编码成JPEG格式—>base64-编码–>组合成MQTT报文–>发布到服务器。
  设备B： 订阅设备A上传的数据–>base64解码–>解码JPEG数据格式–>GTK图像渲染。

2.MQTT协议介绍

  MQTT(消息队列遥测传输)是ISO 标准(ISO/IEC PRF 20922)下基于发布/订阅范式的消息协议。它工作在 TCP/IP协议族上，是为硬件性能低下的远程设备以及网络状况糟糕的情况下而设计的发布/订阅型消息协议，为此，它需要一个消息中间件 。
  MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。由IBM在1999年发布。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。如：机器与机器（M2M）通信和物联网（IoT）。其在通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。
  MQTT最大优点在于，用极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。

2.1 MQTT特点

  本协议运行在 TCP/IP，或其它提供了有序、可靠、双向连接的网络连接上。MQTT属于应用层协议，它有以下特点：

• 使用发布/订阅消息模式，提供了一对多的消息分发和应用之间的解耦。
• 消息传输不需要知道负载内容。
• 提供三种等级的服务质量： .
• QS0：“最多一次”，尽操作环境所能提供的最大努力分发消息。消息可能会丢失。
  例如，这个等级可用于环境传感器数据，单次的数据丢失没关系，因为不久之后会再次发送。
• QS1：“至少一次”，保证消息可以到达，但是可能会重复。
• QS2：“仅一次”，保证消息只到达一次。例如，这个等级可用在一个计费系统中，这里如果消息重复或丢失会导致不正确的收费。很小的传输消耗和协议数据交换，最大限度减少网络流量。

  在MQTT连接建立时，客户端需要通过TCP连接到MQTT服务器，并进行握手协商，包括协议版本、客户端标识符、遗嘱消息、 QoS级别等信息，以确保双方能够正确地交换数据。一旦握手成功，客户端和服务器之间就建立了一个持久化的TCP连接，可以随时进行消息传输。
  由于TCP协议本身已经提供了一定程度的可靠性保证，因此MQTT协议只需要在TCP的基础上实现发布/订阅机制、 QoS级别控制、保留消息等特性即可，从而使得它成为一种轻量级且高效的物联网通信协议。

2.2 MQTT协议数据量限制

  MQTT协议本身没有限制数据包的大小，但是它需要遵循底层传输协议（TCP/IP）的限制和约束。在实际应用中，MQTT协议能够传输的有效数据量是受多种因素影响的，如网络带宽、QoS级别、MQTT消息头部信息等。一般来说，在默认情况下，MQTT协议对于单个消息的有效载荷有一个限制，即不超过256MB。这个限制主要由MQTT协议的消息长度字段决定，该字段的最大值是4字节，因此最大能表示2^32-1个字节的消息长度，即约为4GB。然而，在实际应用中，由于网络带宽和设备性能等方面的限制，很难实现传输如此巨大的消息。
  另外，需要注意的是，如果使用较高级别的QoS，如“至少一次”或“恰好一次”，则MQTT协议会对每条消息进行确认和重传，这可能会导致更多的网络流量和延迟。因此，在选择QoS级别时，需要根据应用场景和网络环境的实际情况进行优化和调整，以充分利用MQTT协议的特点和优势。

3.阿里云物联网平台搭建

3.1 搭建阿里云物联网平台

1.登录阿里云物联网平台

2.创建产品

3.功能定义，添加自定义功能，发布上线。

4.添加设备

3.2 设备登录

  MQTT连接阿里云需要三个参数：客户端id、用户名、密码。此信息可直接在设备中获取。

3.3 消息订阅与消息发布

  Topic是消息发布（Pub）者和订阅（Sub）者之间的传输中介。设备可通过Topic实现消息的发送和接收，从而实现服务端与设备端的通信。为方便海量设备基于Topic进行通信，简化授权操作，物联网平台定义了产品Topic类和设备Topic。

3.4 数据上报格式

  MQTT上报数据格式为JSON格式。内容格式如下：

发布主题："/sys/{产品ID}/{设备名}/thing/service/property/set"
数据内容格式：{"method":"thing.event.property.post","params":{"image":"hello,world"}}

  设备端订阅消息内容：

订阅主题："/sys/{产品ID}/{设备名}/thing/event/property/post"

4.Linux下socket编程连接阿里云物联网平台

#define SERVER_IP "asfdda.iot-as-mqtt.cn-shanghai.aliyuncs.com"//服务器IP
#define SERVER_PORT 1883 //端口号
#define ClientID "aasfsaXABf.Imasfas|securemode=2,signmethod=hmacsha256,timestamp=1678323607797|"
#define Username "ImsfeA&a1sadf8XABf"
#define Password "15566ab496e81da728a3792ebe532fd4a3f4026a2b831df5af24da06"//密文 
#define SET_TOPIC  "/sys/a14dXABf/ImagfA/thing/service/property/set"  //订阅  
#define POST_TOPIC "/sys/a14sdf8XABf/ImdfeA/thing/event/property/post"  //发布  
int main()
{pthread_t id;signal(SIGPIPE,SIG_IGN);/*忽略SIGPIPE信号*/signal(SIGALRM,signal_func);/*闹钟信号*/sockfd=socket(AF_INET,SOCK_STREAM,0);if(sockfd==-1){printf("网络套接字打开失败\n");return 0;}/*设置发送缓冲区大小*/int nSendBuf=40*1024;//设置为 20Kif(setsockopt(sockfd,SOL_SOCKET,SO_SNDBUF,(const char*)&nSendBuf,sizeof(int))){printf("setsockopt(SO_SNDBUF) 设置错误!\n");return 0;}/*域名解析*/struct hostent *hostent;while(1){hostent=gethostbyname(SERVER_IP);if(hostent==NULL){printf("域名解析失败\n");sleep(1);}else break;}printf("主机名:%s\n",hostent->h_name);printf("协议类型:%s\n",(hostent->h_addrtype == AF_INET)?"AF_INET":"AF_INET6");printf("IP地址长度:%d\n",hostent->h_length);char *ip;for(int i=0;hostent->h_addr_list[i];i++){ip=inet_ntoa(*(struct in_addr *)hostent->h_addr_list[i]);printf("ip=%s\n",ip);}	/*连接服务器*/struct sockaddr_in addr;addr.sin_family=AF_INET;//IPV4addr.sin_port=htons(SERVER_PORT);/*端口号*/addr.sin_addr.s_addr=inet_addr(ip);//服务器IPif(connect(sockfd, (struct sockaddr *)&addr,sizeof(struct sockaddr_in))==0){printf("服务器连接成功\n");while(1){MQTT_Init();/*登录服务器*/if(MQTT_Connect(ClientID,Username,Password)==0){break;}sleep(1);printf("服务器连接中....\n");}printf("连接成功\r\n");//订阅物联网平台数据stat=MQTT_SubscribeTopic(SET_TOPIC,1,1);if(stat){close(sockfd);printf("订阅失败\r\n");  exit(0);}printf("订阅成功\r\n");/*创建线程*/pthread_create(&id, NULL,pth_work_func,NULL);pthread_detach(id);//设置分离属性alarm(3);//闹钟函数，时间到达会产生SIGALRM信号int a=0;while(1){  sprintf(mqtt_message,"{\"method\":\"thing.event.property.post\",\"params\":{\"image\":\"阿里云物联网平台测试\"}}");MQTT_PublishData(POST_TOPIC,mqtt_message,0);//发布数据}}
}

5.云产品流转

  云产品流转文档：阿里云云产品流转

5.1 什么是云产品流转

  设备基于Topic与物联网平台进行通信时，您可以在数据流转中，编写SQL对Topic中的数据进行处理，并配置转发规则将处理后的数据转发到其他设备Topic或阿里云其他服务。

5.2云产品流转配置

1.创建解析器

2.关联数据源

4.编写解析器脚本

  解析器说明文档：解析器说明文档


  格式示例：

//通过payload函数，获取设备上报的消息内容，并按照JSON格式转换。 
var data = payload("json"); //直接流转物模型上报数据。 
writeIotTopic(1000, topic, data);

  topic如下：


  编辑好后发布即可，至此，阿里物联网平台配置完成。

6.代码实现

6.1 设备A发送方

1.USB摄像头应用编程

  采用Linux下V4L2框架初始化USB摄像头，采集图像数据。

/*
摄像头初始化
返回值：成功返回摄像头描述符，失败返回负数
*/
int Video_Init(struct CAMERA *camera)
{int video_fd;int i=0;/*1.打开设备节点*/video_fd=open(VIDEO_DEV,O_RDWR);if(video_fd==-1)return -1;/*2.设置摄像头格式*/struct v4l2_format format;memset(&format,0,sizeof(format));format.type=V4L2_BUF_TYPE_VIDEO_CAPTURE;//视频捕获格式format.fmt.pix.width=320;format.fmt.pix.height=240;format.fmt.pix.pixelformat=V4L2_PIX_FMT_YUYV;//图像数据格式yuyvif(ioctl(video_fd,VIDIOC_S_FMT,&format))return -2;printf("图像尺寸:%d * %d\n",format.fmt.pix.width,format.fmt.pix.height);camera->image_w=format.fmt.pix.width;camera->image_h=format.fmt.pix.height;/*3.向内核请求缓冲区*/struct v4l2_requestbuffers reqbuf;memset(&reqbuf,0,sizeof(reqbuf));reqbuf.count=4;/*缓冲区个数*/reqbuf.type=V4L2_BUF_TYPE_VIDEO_CAPTURE;//视频捕获格式reqbuf.memory=V4L2_MEMORY_MMAP;/*内存映射*/if(ioctl(video_fd,VIDIOC_REQBUFS,&reqbuf))return -3;printf("缓冲区个数:%d\n",reqbuf.count);/*4.将缓冲区映射到进程空间*/struct v4l2_buffer quebuff;for(i=0;imemset(&quebuff,0,sizeof(quebuff));quebuff.index=i;//缓冲区数组下标quebuff.type=V4L2_BUF_TYPE_VIDEO_CAPTURE;//视频捕获格式quebuff.memory=V4L2_MEMORY_MMAP;/*内存映射*/if(ioctl(video_fd,VIDIOC_QUERYBUF,&quebuff))return -4;camera->mamp_buff[i]=mmap(NULL,quebuff.length,PROT_READ|PROT_WRITE,MAP_SHARED,video_fd,quebuff.m.offset);printf("buff[%d]=%p\n",i,camera->mamp_buff[i]);camera->mmap_size=quebuff.length;}/*5.将缓冲区添加到采集队列*/for(i=0;imemset(&quebuff,0,sizeof(quebuff));quebuff.index=i;//缓冲区数组下标quebuff.type=V4L2_BUF_TYPE_VIDEO_CAPTURE;//视频捕获格式quebuff.memory=V4L2_MEMORY_MMAP;/*内存映射*/if(ioctl(video_fd,VIDIOC_QBUF,&quebuff))return -5;}/*6.开启摄像头*/int type=V4L2_BUF_TYPE_VIDEO_CAPTURE;//视频捕获格式if(ioctl(video_fd,VIDIOC_STREAMON,&type))return -6;return video_fd;
}

2.图片编码处理

  实时采集图像数据，将图片数据编码为jpg图像格式，再进base64格式编码。
  Base64编码是一种将二进制数据转换为ASCII字符的方法，它使用64个字符来表示任意序列的二进制数据。Base64编码后的数据长度会比原始二进制数据略长，但可以方便地被转换为文本格式并在网络上进行传输。

3.base64格式编码

#include 
static const char * base64char = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
/*
函数功能：将图片编码为base64格式
形参：bindata 源图片数据base64  编码后的数据binlength --源文件大小
返回值：返回编码后的base64数据
*/
char * base64_encode( const unsigned char * bindata, char * base64, int binlength )
{int i, j;unsigned char current;for ( i = 0, j = 0 ; i < binlength ; i += 3 ){current = (bindata[i] >> 2) ;current &= (unsigned char)0x3F;base64[j++] = base64char[(int)current];current = ( (unsigned char)(bindata[i] << 4 ) ) & ( (unsigned char)0x30 ) ;if ( i + 1 >= binlength ){base64[j++] = base64char[(int)current];base64[j++] = '=';base64[j++] = '=';break;}current |= ( (unsigned char)(bindata[i+1] >> 4) ) & ( (unsigned char) 0x0F );base64[j++] = base64char[(int)current];current = ( (unsigned char)(bindata[i+1] << 2) ) & ( (unsigned char)0x3C ) ;if ( i + 2 >= binlength ){base64[j++] = base64char[(int)current];base64[j++] = '=';break;}current |= ( (unsigned char)(bindata[i+2] >> 6) ) & ( (unsigned char) 0x03 );base64[j++] = base64char[(int)current];current = ( (unsigned char)bindata[i+2] ) & ( (unsigned char)0x3F ) ;base64[j++] = base64char[(int)current];}base64[j] = '\0';return base64;
}

4.base64格式解码

/*
函数功能：base64格式数据解码
形参：base64 base64格式数据bindata  保存解码成功的图像数据
返回值：成功返回解码的图像大小
*/
static const char * base64char = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
int base64_decode( const char * base64, unsigned char * bindata )
{int i, j;unsigned char k;unsigned char temp[4];for ( i = 0, j = 0; base64[i] != '\0' ; i += 4 ){memset( temp, 0xFF, sizeof(temp) );for ( k = 0 ; k < 64 ; k ++ ){if ( base64char[k] == base64[i] )temp[0]= k;}for ( k = 0 ; k < 64 ; k ++ ){if ( base64char[k] == base64[i+1] )temp[1]= k;}for ( k = 0 ; k < 64 ; k ++ ){if ( base64char[k] == base64[i+2] )temp[2]= k;}for ( k = 0 ; k < 64 ; k ++ ){if ( base64char[k] == base64[i+3] )temp[3]= k;}bindata[j++] = ((unsigned char)(((unsigned char)(temp[0] << 2))&0xFC)) |((unsigned char)((unsigned char)(temp[1]>>4)&0x03));if ( base64[i+2] == '=' )break;bindata[j++] = ((unsigned char)(((unsigned char)(temp[1] << 4))&0xF0)) |((unsigned char)((unsigned char)(temp[2]>>2)&0x0F));if ( base64[i+3] == '=' )break;bindata[j++] = ((unsigned char)(((unsigned char)(temp[2] << 6))&0xF0)) |((unsigned char)(temp[3]&0x3F));}return j;
}

5.数据上报

  Linux下socket网络编程，连接阿里云服务器，接入阿里云物联网平台，通过MQTT协议实时上报数据。

6.2 设备B订阅方

1.数据获取

  Linux下socket网络编程，连接阿里云服务器，接入阿里云物联网平台，订阅设备A端发送的消息。

2.数据解析

  物联网平台下发消息格式为JSON格式，进行消息数据解析，提取图像数据，将图像数据进行base64格式解码得到JPG图片数据。

3.JPG图片解析

  解析JPG图片获取到RGB颜色数据。

//显示JPEG 编译时加-ljpeg
int LCD_ShowJPEG(unsigned char *jpg_buffer,int size,struct ImageDecodingInfo *image_rgb)
{struct jpeg_decompress_struct cinfo; //存放图像的数据struct jpeg_error_mgr jerr; //存放错误信息unsigned char  *buffer;unsigned int i,j;unsigned int  color;static int written;/*init jpeg压缩对象错误处理程序*/cinfo.err = jpeg_std_error(&jerr); //初始化标准错误，用来存放错误信息jpeg_create_decompress(&cinfo);    //创建解压缩结构信息jpeg_mem_src(&cinfo, (unsigned char *)jpg_buffer, size);//jpeg_stdio_src(&cinfo, infile);/*读jpeg头*/jpeg_read_header(&cinfo, TRUE);/*开始解压*/jpeg_start_decompress(&cinfo);#if 0printf("JPEG图片高度: %d\n",cinfo.output_height);printf("JPEG图片宽度: %d\n",cinfo.output_width);printf("JPEG图片颜色位数（字节单位）: %d\n",cinfo.output_components);#endifimage_rgb->Height=cinfo.output_height;image_rgb->Width=cinfo.output_width;unsigned char *rgb_data=image_rgb->rgb;/*为一条扫描线上的像素点分配存储空间，一行一行的解码*/int row_stride = cinfo.output_width * cinfo.output_components;buffer = (unsigned char *)malloc(row_stride);  //将图片内容显示到framebuffer上,cinfo.output_scanline表示当前行的位置，读取数据是会自动增加i=0;while(cinfo.output_scanline < cinfo.output_height){//读取一行的数据    jpeg_read_scanlines(&cinfo,&buffer,1);memcpy(rgb_data + i * cinfo.output_width * 3, buffer, row_stride);i++;}    /*完成解压,摧毁解压对象*/jpeg_finish_decompress(&cinfo); //结束解压jpeg_destroy_decompress(&cinfo); //释放结构体占用的空间/*释放内存缓冲区*/free(buffer);return 0;      
}

4.GTK窗口渲染

  创建GTK窗口，将原始图片进行缩放，实时渲染图像数据。

/*****************************BMP图片放大缩小************************
**image_rgb --图像结构体信息
**int lcd_width,int lcd_hight --屏幕大小
**返回值:0 -- 成功； 其它值 -- 失败
*********************************************************************/
int ZoomInandOut(struct ImageDecodingInfo *image_rgb,int lcd_width,int lcd_hight)
{//printf("源图片宽:%d\n",image_rgb->Width);//printf("源图片高:%d\n",image_rgb->Height);u32 w=image_rgb->Width;u32 h=image_rgb->Height;u8 *src_rgb=image_rgb->rgb;//源图片RGB值unsigned long oneline_byte=w*3;//一行字节数float zoom_count=0;/*按比例缩放*/zoom_count=(lcd_width/(w*1.0)) > (lcd_hight/(h*1.0)) ? (lcd_hight/(h*1.0)):(lcd_width/(w*1.0));int new_w,new_h;new_w=zoom_count*w;//新图片宽new_h=zoom_count*h;//新图片高//printf("新图片宽:%d\n",new_w);//printf("新图片高:%d\n", new_h);//printf("缩放比例:%.0f%%\n",(new_w*1.0/w)*100);unsigned long new_oneline_byte=new_w*3;unsigned char *newbmp_buff=(unsigned char *)malloc(new_h*new_oneline_byte);//动态分配新图片RGB颜色数据缓冲区if(newbmp_buff==NULL){printf("[%s line %d]动态分配空间失败\n",__FUNCTION__,__LINE__);return -1;}memset(newbmp_buff, 0, new_h*new_oneline_byte);/************************图像处理算法(双线性插值)*******************************/int i,j;for(i=0;ifor(j=0;jdouble d_original_img_h = i*h / (double)new_h;double d_original_img_w = j*w / (double)new_w;int i_original_img_h = d_original_img_h;int i_original_img_w = d_original_img_w;double distance_to_a_x = d_original_img_w - i_original_img_w;//在原图像中与a点的水平距离 double distance_to_a_y = d_original_img_h - i_original_img_h;//在原图像中与a点的垂直距离 int original_point_a = i_original_img_h*oneline_byte + i_original_img_w * 3;//数组位置偏移量，对应于图像的各像素点RGB的起点,相当于点A    int original_point_b = i_original_img_h*oneline_byte + (i_original_img_w + 1) * 3;//数组位置偏移量，对应于图像的各像素点RGB的起点,相当于点B  int original_point_c = (i_original_img_h + 1)*oneline_byte + i_original_img_w * 3;//数组位置偏移量，对应于图像的各像素点RGB的起点,相当于点C   int original_point_d = (i_original_img_h + 1)*oneline_byte + (i_original_img_w + 1) * 3;//数组位置偏移量，对应于图像的各像素点RGB的起点,相当于点D   if (i_original_img_h +1== new_h - 1){original_point_c = original_point_a;original_point_d = original_point_b;}if (i_original_img_w +1== new_w - 1){original_point_b = original_point_a;original_point_d = original_point_c;}int pixel_point = i*new_oneline_byte + j*3;//映射尺度变换图像数组位置偏移量  newbmp_buff[pixel_point] =src_rgb[original_point_a] * (1 - distance_to_a_x)*(1 - distance_to_a_y) +src_rgb[original_point_b] * distance_to_a_x*(1 - distance_to_a_y) +src_rgb[original_point_c] * distance_to_a_y*(1 - distance_to_a_x) +src_rgb[original_point_c] * distance_to_a_y*distance_to_a_x;newbmp_buff[pixel_point + 1] =src_rgb[original_point_a + 1] * (1 - distance_to_a_x)*(1 - distance_to_a_y) +src_rgb[original_point_b + 1] * distance_to_a_x*(1 - distance_to_a_y) +src_rgb[original_point_c + 1] * distance_to_a_y*(1 - distance_to_a_x) +src_rgb[original_point_c + 1] * distance_to_a_y*distance_to_a_x;newbmp_buff[pixel_point + 2] =src_rgb[original_point_a + 2] * (1 - distance_to_a_x)*(1 - distance_to_a_y) +src_rgb[original_point_b + 2] * distance_to_a_x*(1 - distance_to_a_y) +src_rgb[original_point_c + 2] * distance_to_a_y*(1 - distance_to_a_x) +src_rgb[original_point_c + 2] * distance_to_a_y*distance_to_a_x;}}memcpy(image_rgb->rgb,newbmp_buff,new_h*new_oneline_byte);//新图像RGB数据image_rgb->Width=new_w;//新图像宽image_rgb->Height=new_h;//新图像高free(newbmp_buff);return 0;
}

6.3 项目效果