MQTT 物联网服务

import paho.mqtt.client as mqtt

# MQTT连接配置
broker = "mqtt.apifm.com"
port = 1883
username = "your_username"
password = "your_password"
topic = "your/topic"

# 连接成功回调
def on_connect(client, userdata, flags, rc):
    print(f"连接结果: {rc}")
    client.subscribe(topic)

# 接收消息回调
def on_message(client, userdata, msg):
    print(f"收到消息: {msg.topic} - {msg.payload.decode()}")

# 创建客户端
client = mqtt.Client()
client.username_pw_set(username, password)
client.on_connect = on_connect
client.on_message = on_message

# 连接并发送消息
client.connect(broker, port, 60)
client.loop_start()
client.publish(topic, "Hello from Python!")

# 保持连接
client.loop_forever()

const mqtt = require('mqtt');

// MQTT连接配置
const options = {
  host: 'mqtt.apifm.com',
  port: 1883,
  username: 'your_username',
  password: 'your_password',
  clientId: 'mqtt_client_' + Math.random().toString(16).substr(2, 8)
};

// 连接MQTT服务器
const client = mqtt.connect(options);

client.on('connect', () => {
  console.log('已连接到MQTT服务器');
  
  // 订阅主题
  client.subscribe('your/topic', (err) => {
    if (!err) {
      // 发送消息
      client.publish('your/topic', 'Hello from Node.js!');
    }
  });
});

client.on('message', (topic, message) => {
  console.log(`收到消息: ${topic} - ${message.toString()}`);
});

client.on('error', (err) => {
  console.error('连接错误:', err);
});

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class MqttExample {
    public static void main(String[] args) {
        String broker = "tcp://mqtt.apifm.com:1883";
        String clientId = "JavaClient_" + System.currentTimeMillis();
        String topic = "your/topic";
        String username = "your_username";
        String password = "your_password";
        
        try {
            MqttClient client = new MqttClient(broker, clientId, new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setUserName(username);
            options.setPassword(password.toCharArray());
            options.setCleanSession(true);
            
            // 设置回调
            client.setCallback(new MqttCallback() {
                public void messageArrived(String topic, MqttMessage message) {
                    System.out.println("收到消息: " + topic + " - " + new String(message.getPayload()));
                }
                
                public void connectionLost(Throwable cause) {
                    System.out.println("连接丢失: " + cause.getMessage());
                }
                
                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.println("消息发送完成");
                }
            });
            
            // 连接
            client.connect(options);
            System.out.println("已连接到MQTT服务器");
            
            // 订阅
            client.subscribe(topic);
            
            // 发送消息
            MqttMessage message = new MqttMessage("Hello from Java!".getBytes());
            message.setQos(1);
            client.publish(topic, message);
            
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}

<?php
require 'vendor/autoload.php';

use PhpMqtt\Client\MqttClient;
use PhpMqtt\Client\ConnectionSettings;

$server   = 'mqtt.apifm.com';
$port     = 1883;
$clientId = 'php-mqtt-client-' . uniqid();
$username = 'your_username';
$password = 'your_password';
$topic    = 'your/topic';

// 连接设置
$connectionSettings = (new ConnectionSettings)
    ->setUsername($username)
    ->setPassword($password)
    ->setKeepAliveInterval(60);

// 创建MQTT客户端
$mqtt = new MqttClient($server, $port, $clientId);

try {
    $mqtt->connect($connectionSettings, true);
    echo "已连接到MQTT服务器\n";
    
    // 发送消息
    $mqtt->publish($topic, 'Hello from PHP!', 0);
    echo "消息已发送\n";
    
    // 订阅消息
    $mqtt->subscribe($topic, function ($topic, $message) {
        echo "收到消息: {$topic} - {$message}\n";
    }, 0);
    
    $mqtt->loop(true);
    $mqtt->disconnect();
    
} catch (Exception $e) {
    echo "错误: " . $e->getMessage() . "\n";
}
?>

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTClient.h"

#define ADDRESS     "tcp://mqtt.apifm.com:1883"
#define CLIENTID    "CClient"
#define TOPIC       "your/topic"
#define USERNAME    "your_username"
#define PASSWORD    "your_password"
#define QOS         1
#define TIMEOUT     10000L

int messageArrived(void *context, char *topicName, int topicLen, MQTTClient_message *message) {
    printf("收到消息: %s - %.*s\n", topicName, message->payloadlen, (char*)message->payload);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void connectionLost(void *context, char *cause) {
    printf("连接丢失: %s\n", cause);
}

int main(int argc, char* argv[]) {
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    int rc;

    MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    conn_opts.username = USERNAME;
    conn_opts.password = PASSWORD;

    MQTTClient_setCallbacks(client, NULL, connectionLost, messageArrived, NULL);

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
        printf("连接失败，错误码: %d\n", rc);
        exit(EXIT_FAILURE);
    }

    printf("已连接到MQTT服务器\n");

    MQTTClient_subscribe(client, TOPIC, QOS);

    pubmsg.payload = "Hello from C!";
    pubmsg.payloadlen = strlen(pubmsg.payload);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    
    MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);
    MQTTClient_waitForCompletion(client, token, TIMEOUT);
    printf("消息已发送\n");

    // 保持连接，接收消息
    while(1) {
        MQTTClient_yield();
    }

    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    return rc;
}

#include <WiFi.h>
#include <PubSubClient.h>

// WiFi配置
const char* ssid = "your_wifi_ssid";
const char* password = "your_wifi_password";

// MQTT配置
const char* mqtt_server = "mqtt.apifm.com";
const int mqtt_port = 1883;
const char* mqtt_username = "your_username";
const char* mqtt_password = "your_password";
const char* mqtt_topic = "your/topic";

WiFiClient espClient;
PubSubClient client(espClient);

void setup_wifi() {
  delay(10);
  Serial.println();
  Serial.print("连接WiFi: ");
  Serial.println(ssid);

  WiFi.begin(ssid, password);

  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.print(".");
  }

  Serial.println("");
  Serial.println("WiFi已连接");
  Serial.println("IP地址: ");
  Serial.println(WiFi.localIP());
}

void callback(char* topic, byte* payload, unsigned int length) {
  Serial.print("收到消息 [");
  Serial.print(topic);
  Serial.print("] ");
  for (int i = 0; i < length; i++) {
    Serial.print((char)payload[i]);
  }
  Serial.println();
}

void reconnect() {
  while (!client.connected()) {
    Serial.print("连接MQTT服务器...");
    String clientId = "ESP32Client-";
    clientId += String(random(0xffff), HEX);
    
    if (client.connect(clientId.c_str(), mqtt_username, mqtt_password)) {
      Serial.println("已连接");
      client.subscribe(mqtt_topic);
    } else {
      Serial.print("连接失败, rc=");
      Serial.print(client.state());
      Serial.println(" 5秒后重试");
      delay(5000);
    }
  }
}

void setup() {
  Serial.begin(115200);
  setup_wifi();
  client.setServer(mqtt_server, mqtt_port);
  client.setCallback(callback);
}

void loop() {
  if (!client.connected()) {
    reconnect();
  }
  client.loop();

  // 每10秒发送一次消息
  static unsigned long lastMsg = 0;
  unsigned long now = millis();
  if (now - lastMsg > 10000) {
    lastMsg = now;
    String msg = "Hello from Arduino! Uptime: " + String(now/1000) + "s";
    Serial.print("发送消息: ");
    Serial.println(msg);
    client.publish(mqtt_topic, msg.c_str());
  }
}

package main

import (
    "fmt"
    "time"
    mqtt "github.com/eclipse/paho.mqtt.golang"
)

func main() {
    // MQTT配置
    broker := "tcp://mqtt.apifm.com:1883"
    clientID := "go-mqtt-client"
    username := "your_username"
    password := "your_password"
    topic := "your/topic"

    // 消息回调
    var messagePubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
        fmt.Printf("收到消息: %s - %s\n", msg.Topic(), msg.Payload())
    }

    // 连接回调
    var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {
        fmt.Println("已连接到MQTT服务器")
    }

    // 断开连接回调
    var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
        fmt.Printf("连接丢失: %v\n", err)
    }

    // 设置选项
    opts := mqtt.NewClientOptions()
    opts.AddBroker(broker)
    opts.SetClientID(clientID)
    opts.SetUsername(username)
    opts.SetPassword(password)
    opts.SetDefaultPublishHandler(messagePubHandler)
    opts.OnConnect = connectHandler
    opts.OnConnectionLost = connectLostHandler

    // 创建并连接
    client := mqtt.NewClient(opts)
    if token := client.Connect(); token.Wait() && token.Error() != nil {
        panic(token.Error())
    }

    // 订阅主题
    if token := client.Subscribe(topic, 1, nil); token.Wait() && token.Error() != nil {
        fmt.Println(token.Error())
        return
    }

    // 发送消息
    text := "Hello from Go!"
    token := client.Publish(topic, 0, false, text)
    token.Wait()
    fmt.Printf("消息已发送: %s\n", text)

    // 保持运行
    time.Sleep(60 * time.Second)
    
    client.Disconnect(250)
}