基于MQTT协议实现从树莓派上获得的温度上传至数据库中

到这里就可以将前一博客中构建的框架放入项目代码中了，只需要修改之前项目中相关socket代码部分即可。

1.发布端：

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include "mosquitto.h"
#include 
#include "ds18b20.h"
#include 
#include 

#define BUF_SZIE  1024
#define ARRY_SIZE(x)    (sizeof(x)/sizeof(x[0]))
#define KEEP_ALIVE 60


int g_stop = 0;

static inline void print_usage(char *progname);
void sig_handler(int sig_num)
{
    if(sig_num == SIGUSR1)
        g_stop = 1;

}


int main (int argc, char **argv)
{

    char *progname = basename(argv[0]);
    int port;
    char *ip = NULL;
    char *hostname = NULL;
    char *topic = NULL;
    struct hostent *hostentp;
    struct mosquitto *mosq = NULL;
    int daemon_run = 0;
    int opt;
    int rv;
    float temperature = 0;
    char buffer[BUF_SZIE];
    time_t  t;
    struct tm *timep;
    int log_fd;
    int connect_flag = 1;
     struct option options[] = 
     {
         {"daemon",no_argument,NULL,'d'},
         {"topic", required_argument,NULL,'t'},
         {"hostname", required_argument,NULL,'n'},
         {"ip", required_argument, NULL, 'i'},
         {"port",required_argument,NULL,'p'},
         {"help",no_argument,NULL,'h'},
         {NULL,0,NULL,0}
     };


     while((opt = getopt_long(argc,argv,"dhp:t:i:n:",options,NULL)) != -1)
     {
        
         switch(opt)
         
         {

             case 't':
                 topic = optarg;
                 break;

             case 'i':
                 ip = optarg;
                 break;

             case 'n':
                 hostname = optarg;
                 break;


             case 'd':
                 daemon_run = 1;
                 break;
             case 'p':
                 port = atoi(optarg);
                 break;
             case 'h':
                 print_usage(argv[0]);
                 return 0;
             default:
                 break;
         }
     }

     if(!port)
     {
         print_usage(progname);
         return -1;
     }

     if(!hostname && !ip)
     {
        print_usage(progname);
        return -1;
     }
    
    
     if(hostname)
     {
         hostentp = gethostbyname(hostname);
         if(!hostentp)
         {
             printf("Failed to get host by name: %s\n",strerror(errno));
             return -2;
         }
        printf("hostname: %s\n",hostentp->h_name);
        ip = inet_ntoa(*(struct in_addr *)hostentp->h_addr);
        printf("address: %s\n",ip);

     }

    if(!topic)
    {
        topic = "temperature";
    }

     if(daemon_run)
     {
         printf("program %s running in backgrund\n", progname);
         if( (log_fd = open("client.log", O_CREAT|O_RDWR, 0666)) < 0)
         {
             printf("open() failed:%s\n", strerror(errno)) ;
             return -2;
         }

        dup2(log_fd, STDOUT_FILENO) ;
        dup2(log_fd, STDERR_FILENO) ;

         daemon(1,1);
     }
     
    
     signal(SIGUSR1,sig_handler);
     
     mosquitto_lib_init();
     mosq = mosquitto_new(NULL,true,NULL);

     if(!mosq)
     {
         printf("mosquitto_new() failed: %s\n",strerror(errno));
         mosquitto_lib_cleanup();
         return -1;
     }
     printf("Create mosquitto successfully!\n");
     if(mosquitto_connect(mosq,ip,port,KEEP_ALIVE) != MOSQ_ERR_SUCCESS)
     {

          printf("mosquitto_connect() failed: %s\n",strerror(errno));
          mosquitto_destroy(mosq);
          mosquitto_lib_cleanup();
          return -1;
     }

     printf("connect %s:%d successfully!\n",ip,port);
     if(mosquitto_loop_start(mosq) != MOSQ_ERR_SUCCESS)
     {

                printf("mosquitto_loop_start() failed: %s\n",strerror(errno));
                mosquitto_destroy(mosq);
                mosquitto_lib_cleanup();
     }

     while(!g_stop)
     {
        
         if(ds18b20_get_temperature(&temperature) < 0)
         {
             printf("ds18b20_get_temperature failed.\n");
             return -3;
         }

        time(&t);
        char * str = ctime(&t);//此字符串后有回车换行及‘/n’,所以要去掉
        char tempbuff[25];
        memset(tempbuff,0,25);
        memcpy(tempbuff,str,24);
        snprintf(buffer,ARRY_SIZE(buffer),"%s%f",tempbuff,temperature);
        printf("%s\n",buffer);
        

        if(mosquitto_publish(mosq,NULL,topic,strlen(buffer),buffer,0,0) != MOSQ_ERR_SUCCESS)
        {
            
                printf("mosquitto_publish() failed: %s\n",strerror(errno));
                mosquitto_destroy(mosq);
                mosquitto_lib_cleanup();
            
        
        }else
        {
            printf("Publish information of temperature Ok!\n") ;
        }
        sleep(30);
     }

    close(log_fd);
     return 0;
} 

void print_usage(char *progname)
{
    printf("%s usage:\n",progname);
    printf("-p(--port): sepcify server listen port.\n");
    printf("-h(--Help): print this help information.\n");
    printf("-d(--daemon): set program running on background.\n");
    printf("\nExample: %s -d -p 8889\n",progname);
}

2.订阅端：

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include "mosquitto.h"
#include 
#include 
#include 

#define MSG_MAX_SZIE 1024
#define ARRY_SIZE(x)    (sizeof(x)/sizeof(x[0]))
#define KEEP_ALIVE 60

static inline void print_usage(char *progname);
int dboperate(char *buffer);
int g_stop = 0;

void my_message_calllback(struct mosquitto *mosq,void *obj,const struct mosquitto_message *msg);
void sig_handler(int sig_num)
{
    if(sig_num == SIGUSR1)
        g_stop = 1;
            
}
     

int main (int argc, char **argv)
{
    char *progname = basename(argv[0]);
    char *ip = NULL;
    char *hostname = NULL;
    char *topic = NULL;
    struct hostent *hostentp;
    struct mosquitto *mosq = NULL;
    int connect_flag = 1;
    int log_fd;
    int port;
    int daemon_run = 0;
    int opt;
    char buffer[MSG_MAX_SZIE];
    //char *obj = buffer;


     struct option options[] = 
     {
         {"daemon",no_argument,NULL,'d'},   
         {"port",required_argument,NULL,'p'},
         {"topic", required_argument,NULL,'t'},
         {"hostname", required_argument,NULL,'n'},
         {"ip", required_argument, NULL, 'i'},
         {"help",no_argument,NULL,'h'},
         {NULL,0,NULL,0}
     };

     while((opt = getopt_long(argc,argv,"dhp:t:i:n:",options,NULL)) != -1)
     {
        
         switch(opt)
         
         {
             case 't':
                  topic = optarg;
                  break;
             case 'i':
                  ip = optarg;
                  break;
             case 'n':
                  hostname = optarg;
                  break;
             case 'd':
                 daemon_run = 1;
                 break;
             case 'p':
                 port = atoi(optarg);
                 break;
             case 'h':
                 print_usage(argv[0]);
                 return 0;
             default:
                 break;
         }
     }

     if(!port)
     {
         print_usage(progname);
         return -1;
     }

     if(!hostname && !ip)
     {
         print_usage(progname);
         return -1;
     }
    
    if(hostname)
    {
        hostentp = gethostbyname(hostname);
        if(!hostname)
        {
            printf("Failed to get host by name: %s\n",strerror(errno));
            return -2;
        }
        printf("hostname: %s\n",hostentp->h_name);

        ip = inet_ntoa(*(struct in_addr *)hostentp->h_addr);
        printf("address: %s\n",ip);
    }

    if(!topic)
    {
        topic = "temperature";
    }


     if(daemon_run)
     {
         printf("program %s running in background\n",progname);
         if((log_fd = open("server.log",O_CREAT|O_RDWR,0666)) < 0)
         {
             printf("open() server.log  failed: %s\n",strerror(errno));
             return -2;
         }

         dup2(log_fd,STDOUT_FILENO);
         dup2(log_fd,STDERR_FILENO);
         daemon(1,1);
     }
     

     signal(SIGUSR1,sig_handler);

    mosquitto_lib_init();
    mosq = mosquitto_new(NULL,true,(void *)buffer);//obj 用户指针的使用在这里
    if(!mosq)
    {
        printf("failed to new mosquitto: %s\n",strerror(errno));
        mosquitto_lib_cleanup();
        return -1;
    }

    printf("new mosq successfully!\n");
    mosquitto_message_callback_set(mosq,my_message_calllback);
    
    if(mosquitto_connect(mosq,ip,port,KEEP_ALIVE) != MOSQ_ERR_SUCCESS)
    {
        printf("mosquitto_connect() failed: %s\n",strerror(errno));
         mosquitto_destroy(mosq);
          mosquitto_lib_cleanup();
          return -2;
    }
    printf("mosq connect successfully!\n");
    if((mosquitto_subscribe(mosq,NULL,topic,2)) != MOSQ_ERR_SUCCESS)
    {
        printf("failed to subscribe from broker: %s\n",strerror(errno));
        mosquitto_destroy(mosq);
        mosquitto_lib_cleanup();
        return -3;
    }
    printf("mosq subscribe successfully!\n");
     while(!g_stop)
     {
         mosquitto_loop(mosq,-1,1);
     }



     mosquitto_destroy(mosq);
     mosquitto_lib_cleanup();
     return 0;
} 

void print_usage(char *progname)
{
    printf("%s usage:\n",progname);
    printf("-p(--port): sepcify server listen port.\n");
    printf("-h(--Help): print this help information.\n");
    printf("-d(--daemon): set program running on background.\n");
    printf("\nExample: %s -d -p 8889\n",progname);
}




int dboperate(char *buffer)
{
	        MYSQL mysql,*sock;	
		MYSQL_RES *res;
		MYSQL_FIELD *fd;
		MYSQL_ROW row;
        char qbuf[1024];

        char str1[25];//字符数组存字符串时 要多一个字节 来存放‘/0’
        memset(str1,0,25);
        memcpy(str1,buffer,24);
        char str2[1024];
        memset(str2,0,1024);
        memcpy(str2,buffer+24,1024-24);
        //printf("%d\n",222);
        printf("%s\n",str1);
        //printf("%d\n",224);
        printf("%s\n",str2);
        float t = atof(str2);

		mysql_init(&mysql);
        
	       	sock = mysql_real_connect(&mysql,"localhost","root","12345678","mydb",0,NULL,0);
		if(!sock)
		{
			fprintf(stderr,"Couldn't connect to database!: %s\n",mysql_error(&mysql));
			exit(1);
		}
        memset(qbuf,0,ARRY_SIZE(qbuf));
		sprintf(qbuf,"insert temperature(subtime,temperature) values('%s',%f)",str1,t);

		if(mysql_query(sock,qbuf))// when mysql_query() excute seccessfully it return 0;

		{
			fprintf(stderr,"Excute failed!: %s\n",mysql_error(sock));
			exit(1);
		}
        printf("write into database successfully!\n");
		return 0;
}


void my_message_calllback(struct mosquitto *mosq,void *obj,const struct mosquitto_message *msg)
{
    memset((char *)obj,0,MSG_MAX_SZIE);
    memcpy((char *)obj,(char *)msg->payload,MSG_MAX_SZIE);
    printf("%s\n",(char *)obj);
    dboperate((char *)obj);


}

再添加之前编写的从树莓派ds18b20传感器获取温度的代码即可，在这里我们采用一个单独的ds18b20.h进行调用。在写外部函数的时候，应该在源文件中写具体定义，而在对应的头文件中写声明，最后在执行文件中包含这个头文件就行了。这样的方法不仅达到了代码功能模块化的效果，也利于排查bug。

#ifndef _DS18_H_
#define _DS18_H_

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 

#define PATH "/sys/bus/w1/devices/"
#define BUF_SIZE 1024


extern int ds18b20_get_temperature(float *t);



#endif

#include "ds18b20.h"


int ds18b20_get_temperature(float *t)
{
      DIR             *dirp;
      struct dirent   *direntp;
      char            path[1024];
      int             fd;
      int             rv;
      char            buffer[BUF_SIZE];
      char            *result;

    if((dirp = opendir(PATH)) == NULL)
    {
		printf("Failed to open the destination directory: %s\n",strerror(errno));
		return -1;
	}

	while((direntp = readdir(dirp)) != NULL)
	{
		if(strstr(direntp->d_name,"28-") != NULL)
		{
			break;
		}
	}
	
	if(direntp == NULL)
	{
		printf("Failed to find the destination directory: 28-*** .\n");
		closedir(dirp);
        return -2;
	}
    
    memset(path,0,sizeof(path));
    strcat(path,PATH);
	strcat(path,direntp->d_name);
	strcat(path,"/w1_slave");
	closedir(dirp);

	if((fd = open(path,O_RDONLY)) < 0)
	{
		printf("Failed to open the destination file.\n");
		return -3;
	}

	memset(buffer,0,sizeof(buffer));
	while((rv = read(fd,buffer,sizeof(buffer))) > 0)
	{	
		
		if((result = strstr(buffer,"t=")) != NULL)
		{
			result = result + 2;
			break;
		}
	}

	if(rv <= 0)
	{
		printf("Failed to read the temperature.\n");
		close(fd);
        return -4;
	}
	
	*t = atof(result)/1000;
    close(fd);
	return 0;
}


#if 0
int main()
{   
    float t;
    ds18b20_get_temperature(&t);
    printf("%f\n",t);
}
#endif 

makefile：

all:
	@echo ""
	@echo "Start compiling..."
	@echo ""
	gcc -o mqttpub mqttpub.c ds18b20.c -lmosquitto
	gcc -o mqttsub mqttsub.c -I/usr/include/mysql -L/usr/lib/mysql -lmosquitto -lmysqlclient
	@echo "end"

clean:
	rm mqttpub mqttsub

runsub:
	./mqttsub -p 1883 -n lastbreath.club

runpub:
	./mqttpub -p 1883 -n lastbreath.club

运行截图：

查看数据库：