基于MQTT通信开发的失物招领小程序

发布于:2024-05-05 ⋅ 阅读:(27) ⋅ 点赞:(0)

项目架构设计

这个项目采用前后端分离的方式,重新设计了两条链路来支撑程序的信息获取和传递

  1. 前端的小程序页面再启动页面渲染时,直接通过DBAPI从后端数据库获取信息,直接渲染在小程序中
  2. 项目中给DBAPI的定位是快速从后端获取信息,但是无法修改数据库,DBAPI和数据库同时及都部署在后端服务器,可以实现信息的快速交互,其中没有网络通信也保证信息安全
  3. 当小程序用户发布失物,或者是提交寻得失物,删除物品或者是用户注册等需要修改数据库的场景时,他们的信息会通过MQTT通信的方式经过小程序专用的socket连接发送给服务器
  4. 然后MQTT服务器会收到这些消息,然后分发给订阅了对应主题消息的客户端
  5. 我们开发的客户端订阅了所有小程序的消息,这个客户端会基于小程序发送的消息主题,选择对应函数,然后基于消息内容进一步去操作后端数据库,实现数据库的存储和修改

数据库设计

设计了以下两个表来存储项目的业务数据

失物表存储了丢失物品相关的信息:失物ID-id;丢失用户名-username;当前状态-type;丢失地点-area;照片-photo

用户表存储了用户相关的数据:学生ID-studentid;用户名-username;手机号-phonenumber

MQTT通信客户端

MQTT服务器所谓一个独立的中枢服务器,依赖于EMQX部署。而基于MQTT通信和数据库做交互,需要编写一个新的通信客户端。

客户端基于配置文件config.json运行,需先填写相关配置信息,客户端会读取并运行

	{  
	    "database server":{  
	        "host":"",  
	        "port":3306,  
	        "user":"",  
	        "password":"",  
	        "database":""  
	    },  
	    "mqtt server":{  
	        "host":"",  
	        "port":1883,  
	        "other_topic":[  
	              
	        ]  
	    }  
	}  

客户端要具备连接MQTT服务器和操作数据库的功能,为此做了以下设计:

1. 程序会识别系统和输出当前平台的运行状况,然后尝试连接到MySQL数据库。如果连接失败,程序会抛出错误并停止运行。

2. 然后,程序创建了一个MQTT客户端,尝试连接到MQTT代理服务器。如果连接失败,程序会记录错误并停止运行。

3. 定义了一个消息处理函数,该函数会在接收到MQTT消息时被调用。这个函数根据消息的主题进行不同的处理,包括插入数据到数据库、更新数据库中的数据、打印日志等。

4. 客户端程序将订阅一些MQTT主题,包括"lost"、"find"、"signup"、"delete"、"exit"、"error"。当这些主题有新的消息时,消息处理函数会被调用。这里设计了高并发处理,当由多个消息时会自动生成子线程去处理,同时对数据库的操作加锁,防止重复操作数据库

5. 当接程序收到中断信号时,程序会断开MQTT连接并退出。

6. 程序带有详细的日志功能,除了在终端会实时输出信息之外还会保存日志文件,方便后续追踪错误

7. 通用性也是我们开发这个客户端的目标,程序交叉编译后的程序可以运行在不同的操作系统和处理器架构上,利于部署到物联网设备

处理消息和SQL语句相关代码

import (
	"database/sql"
	"log"
	"strconv"
	"strings"

	mqtt "github.com/eclipse/paho.mqtt.golang"
)

// 处理 "lost" 主题的消息
func handleLostTopic(payload string, db *sql.DB) {
	parts := strings.Split(payload, ",")
	if len(parts) == 4 {
		user := parts[0]
		datatype := "lost"
		name := parts[1]
		area := parts[2]
		photo := parts[3]

		dbMutex.Lock() // 加锁
		// 往数据库中存储失物信息
		_, err := db.Exec(
			"INSERT INTO sutff (username, type, name, area, photo) VALUES (?, ?, ?, ?, ?)",
			user,
			datatype,
			name,
			area,
			photo,
		)
		dbMutex.Unlock() // 解锁
		if err != nil {
			log.Fatal(err)
		}
		log.Printf(
			"Lost item add to database {User: %s, Name: %s, Area: %s}\n",
			user,
			name,
			area,
		)
	}
}

// 处理 "delete" 主题的消息
func handleDeleteTopic(payload string, db *sql.DB) {
    id, err := strconv.Atoi(payload)
    if err != nil {
        log.Fatal(err)
    }

    // 删除数据库中指定物品
    dbMutex.Lock()
    _, err = db.Exec(
        "DELETE FROM sutff WHERE id = ?", 
        id,
    )
    dbMutex.Unlock()
    if err != nil {
        log.Fatal(err)
    }
    log.Printf("Lost item with ID:%d has been deleted\n", id)
}

// 处理 "find" 主题的消息
func handleFindTopic(payload string, db *sql.DB) {
	id, err := strconv.Atoi(payload)
	if err != nil {
		log.Fatal(err)
	}

	// 更新数据库中的数据
	dbMutex.Lock()
	_, err = db.Exec(
		"UPDATE sutff SET type = ? WHERE id = ?",
		"find",
		id,
	)
	dbMutex.Unlock()
	if err != nil {
		log.Fatal(err)
	}
	log.Printf("Lost item with ID:%d has been marked as found\n", id)
}

// 处理 "signup" 主题的消息
func handleSignupTopic(payload string, db *sql.DB) {
    parts := strings.Split(payload, ",")
    if len(parts) == 3 {
        studentid := parts[0]
        username := parts[1]
        phonenumber := parts[2]
        dbMutex.Lock()

        // 查询数据库中是否已经存在具有相同 studentid 的用户
        var existingUser string
        err := db.QueryRow("SELECT studentid FROM user WHERE studentid = ?", studentid).Scan(&existingUser)

        // 如果查询结果返回了一个或多个记录,那么我们就不需要再插入新的用户
        if err == nil {
            dbMutex.Unlock()
            log.Printf("User with studentid %s already exists\n", studentid)
            return
        }

        // 往数据库中存储新用户信息
        _, err = db.Exec(
            "INSERT INTO user (studentid, username, phonenumber) VALUES (?, ?, ?)",
            studentid,
            username,
            phonenumber,
        )
        dbMutex.Unlock()
        if err != nil {
            log.Fatal(err)
        }
        log.Printf(
            "New user add to database {User: %s, Name: %s, Phone: %s}\n",
            studentid,
            username,
            phonenumber,
        )
    }
}

// 处理接收到的消息
func handleMessage(client mqtt.Client, msg mqtt.Message, db *sql.DB) {
	getmsg := string(msg.Payload())
	if msg.Topic() == "lost" {
		lastComma := strings.LastIndex(getmsg, ",")
		if lastComma != -1 {
			getmsg = getmsg[:lastComma] + ",BASE64(photo)"
		}
	}
	log.Printf("Recevie topic[%s]  message: %s\n", msg.Topic(), getmsg)
	payload := string(msg.Payload())

	// 根据主题处理消息
	switch msg.Topic() {
	case "lost":
		handleLostTopic(payload, db)
	case "delete":
		handleDeleteTopic(payload, db)
	case "find":
		handleFindTopic(payload, db)
	case "signup":
		handleSignupTopic(payload, db)
	case "exit":
		log.Println("remot-eclient log out safely.")
	case "error":
		log.Println("remot-eclient lost connection, please try again later.")
	}
	// 可以基于此位置进一步开发,处理更多的主题
}

构建MQTT通信相关代码

import (
	"database/sql"
	"strconv"
	"strings"
	"sync"
	"time"

	"log"

	mqtt "github.com/eclipse/paho.mqtt.golang"
)

// 项目必须的主题
var extraTopics = []string{
	"lost",
	"delete",
	"find",
	"signup",
	"exit",
	"error",
}


// 创建MQTT客户端
func createMqttClient(config Config) mqtt.Client {
	opts := mqtt.NewClientOptions()
	broker := strings.Join(
		[]string{
			config.MqttServer.Host,
			strconv.Itoa(config.MqttServer.Port),
		},
		":",
	)
	opts.AddBroker(broker)
	timestamp := time.Now().Unix()
	clientID := strings.Join(
		[]string{
			"receiveclient_",
			strconv.FormatInt(timestamp, 10),
		},
		"",
	)
	opts.SetClientID(clientID)
	client := mqtt.NewClient(opts)
	log.Printf("Created MQTT client with ID: %s\n", clientID)
	// 建立连接
	token := client.Connect()
	if token.Wait() && token.Error() != nil {
		log.Fatal(token.Error())
	}
	// 打印连接到服务器的 IP 和端口
	log.Printf("Connected to MQTT broker at: %s\n", broker)
	return client
}

// 订阅主题
func subscribeTopics(client mqtt.Client, config Config, db *sql.DB) {
	messageHandler := func(client mqtt.Client, msg mqtt.Message) {
		// 启动一个新的goroutine来处理这个消息
		go handleMessage(client, msg, db)
	}
	// 创建一个新的切片,包含config.MqttServer.Topic和extraTopics的所有元素
	allTopics := append([]string{}, config.MqttServer.Topic...)
	allTopics = append(allTopics, extraTopics...)

	var wg sync.WaitGroup
	for _, topic := range allTopics {
		wg.Add(1)
		go func(t string) {
			defer wg.Done()
			token := client.Subscribe(t, 0, messageHandler)
			if token.Wait() && token.Error() != nil {
				log.Fatal(token.Error())
			}
			log.Printf("Subscribe: %s\n", t)
		}(topic)
	}
	wg.Wait()
}

失物招领小程序

总共为这个失物招领小程序设计了五个页面:

主页:

是小程序的入口,也是所有功能页面的入口

用户页:

显示用户信息,所有用户都需要在这个页面先登录,校验用户名,学号和手机号。如果没有账号的用户也可以在这里注册登录,登录这里采用了DBAPI来校验数据库的信息,直接在同一个节点做数据库查找和比对,效率高

丢失页:

用户可以在这个页面上传和管理他们的失物,这里只有检测到用户登录了之后才会从后端获取该用户的失物记录,页面启动时渲染的数据来自DBAPI的直接获取,若用户上传失物,失物信息通过MQTT客户端传递,在MQTT服务器接收后MQTT客户端才能收到,然后客户端进一步操作修改数据库,保证数据安全。

寻得页:

找到了失物的用户可以在这个页面更新信息,这里和丢失页面一样,初始的数据渲染来自后端DBAPI的直接调,而当用户寻得物品,他的的消息通过MQTT服务器传递,由我们开发的MQTT客户端接收后,进一步操作修改数据库信息,保证信息安全

总结页:

程序运行历史记录总结,信息直接用DBAPI获取

项目运行

小程序直接从源码运行依赖于微信小程序的开发这工具,直接把wxapp文件夹导入开发者工具即可运行。MQTT客户端已经编译好了exe版本,亦可以交叉编译其他的版本,相关命令:

	go mod init client  
	go mod tidy  
	GOOS={$YOUR_SYSTEM} GOARCH={$YOUR_CPU} go build -o {$EXE_FILE_NAME} -ldflags '-w -s' ./*.go  
	./{$EXE_FILE_NAME}  

GitHub地址

JJLi0427/MQTT_LostFind_WXAPP: Bulid MQTT connection with all arch plantform through our client. Build a lost and found weapp for BJTU use this client. (github.com)


网站公告

今日签到

点亮在社区的每一天
去签到