PHP边缘计算与物联网数据采集

PHP边缘计算与物联网数据采集

物联网场景中，边缘设备需要采集传感器数据并进行初步处理。PHP可以运行在边缘设备上，MQTT协议是物联网通信的标准协议。今天说说PHP在物联网数据采集中的应用。

PHP通过MQTT协议与物联网设备通信。可以使用php-mqtt库或直接通过socket连接MQTT代理。

```php
class MqttClient
{
private $socket;
private string $host;
private int $port;
private string $clientId;
private array $callbacks = [];

public function __construct(string $host = 'localhost', int $port = 1883, string $clientId = '')
{
$this->host = $host;
$this->port = $port;
$this->clientId = $clientId ?: 'php_client_' . bin2hex(random_bytes(4));
}

public function connect(): bool
{
$this->socket = @fsockopen($this->host, $this->port, $errno, $errstr, 5);
if (!$this->socket) {
throw new \RuntimeException("连接失败: {$errstr}");
}

// MQTT CONNECT报文
$variableHeader = pack('n', strlen($this->clientId)) . $this->clientId;
$fixedHeader = chr(0x10);
$remainingLength = strlen($variableHeader);
$fixedHeader .= chr($remainingLength);

fwrite($this->socket, $fixedHeader . $variableHeader);
$response = fread($this->socket, 4);

if (strlen($response) < 4) return false;
return true;
}

public function subscribe(string $topic, callable $callback): void
{
$this->callbacks[$topic] = $callback;

// MQTT SUBSCRIBE报文
$packetId = 1;
$topicLength = strlen($topic);
$payload = pack('n', $packetId) . pack('n', $topicLength) . $topic . chr(0);

$fixedHeader = chr(0x82);
$remainingLength = strlen($payload);
$fixedHeader .= chr($remainingLength);

fwrite($this->socket, $fixedHeader . $payload);
}

public function publish(string $topic, string $message): void
{
$topicLength = strlen($topic);
$payload = pack('n', $topicLength) . $topic . $message;

$fixedHeader = chr(0x30);
$remainingLength = strlen($payload);
$fixedHeader .= chr($remainingLength);

fwrite($this->socket, $fixedHeader . $payload);
}

public function loop(): void
{
while (!feof($this->socket)) {
$byte = fread($this->socket, 1);
if ($byte === false || $byte === '') break;

$type = ord($byte) >> 4;

if ($type === 3) { // PUBLISH
$remaining = ord(fread($this->socket, 1));
$topicLength = unpack('n', fread($this->socket, 2))[1];
$topic = fread($this->socket, $topicLength);
$message = fread($this->socket, $remaining - 2 - $topicLength);

if (isset($this->callbacks[$topic])) {
($this->callbacks[$topic])($topic, $message);
}
}
}
}

public function disconnect(): void
{
if ($this->socket) {
fwrite($this->socket, chr(0xE0) . chr(0x00));
fclose($this->socket);
}
}
}

// MQTT示例
$client = new MqttClient('mqtt.eclipseprojects.io', 1883);

try {
$client->connect();
$client->subscribe('sensors/temperature', function ($topic, $message) {
echo "收到数据 - {$topic}: {$message}\n";
});

// 模拟传感器数据发布
for ($i = 0; $i < 5; $i++) {
$temperature = 20 + rand(0, 100) / 10;
$humidity = 50 + rand(0, 100) / 10;
$client->publish('sensors/temperature', json_encode([
'value' => $temperature,
'unit' => 'celsius',
'time' => time(),
]));
$client->publish('sensors/humidity', json_encode([
'value' => $humidity,
'unit' => 'percent',
'time' => time(),
]));
sleep(1);
}

$client->disconnect();
} catch (\Exception $e) {
echo "错误: {$e->getMessage()}\n";
}
?>
>

传感器数据采集和存储：

```php
class SensorDataCollector
{
private PDO $pdo;
private array $buffer = [];
private int $batchSize = 10;

public function __construct(PDO $pdo)
{
$this->pdo = $pdo;
$this->initTable();
}

private function initTable(): void
{
$this->pdo->exec("
CREATE TABLE IF NOT EXISTS sensor_data (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
device_id VARCHAR(50) NOT NULL,
sensor_type VARCHAR(50) NOT NULL,
value DECIMAL(10, 2) NOT NULL,
unit VARCHAR(20),
recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_device_time (device_id, recorded_at),
INDEX idx_type_time (sensor_type, recorded_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
");
}

public function record(string $deviceId, string $type, float $value, string $unit = ''): void
{
$this->buffer[] = compact('deviceId', 'type', 'value', 'unit');

if (count($this->buffer) >= $this->batchSize) {
$this->flush();
}
}

public function flush(): void
{
if (empty($this->buffer)) return;

$this->pdo->beginTransaction();
$stmt = $this->pdo->prepare("
INSERT INTO sensor_data (device_id, sensor_type, value, unit, recorded_at)
VALUES (?, ?, ?, ?, NOW())
");

foreach ($this->buffer as $data) {
$stmt->execute([$data['deviceId'], $data['type'], $data['value'], $data['unit']]);
}

$this->pdo->commit();
$this->buffer = [];
}

public function queryHistory(string $deviceId, string $type, int $minutes = 60): array
{
$stmt = $this->pdo->prepare("
SELECT * FROM sensor_data
WHERE device_id = ? AND sensor_type = ?
AND recorded_at > DATE_SUB(NOW(), INTERVAL ? MINUTE)
ORDER BY recorded_at ASC
");
$stmt->execute([$deviceId, $type, $minutes]);
return $stmt->fetchAll();
}

public function getLatest(string $deviceId): array
{
$stmt = $this->pdo->prepare("
SELECT sensor_type, value, unit, recorded_at
FROM sensor_data
WHERE device_id = ?
ORDER BY recorded_at DESC
LIMIT 10
");
$stmt->execute([$deviceId]);
return $stmt->fetchAll();
}

public function getStats(string $deviceId, string $type, int $hours = 24): array
{
$stmt = $this->pdo->prepare("
SELECT
MIN(value) as min_value,
MAX(value) as max_value,
AVG(value) as avg_value,
COUNT(*) as sample_count
FROM sensor_data
WHERE device_id = ? AND sensor_type = ?
AND recorded_at > DATE_SUB(NOW(), INTERVAL ? HOUR)
");
$stmt->execute([$deviceId, $type, $hours]);
return $stmt->fetch(PDO::FETCH_ASSOC);
}

public function __destruct()
{
$this->flush();
}
}

$pdo = new PDO('mysql:host=localhost;dbname=iot', 'root', '');
$collector = new SensorDataCollector($pdo);

$collector->record('device_001', 'temperature', 23.5, 'celsius');
$collector->record('device_001', 'humidity', 65.2, 'percent');
$collector->record('device_001', 'temperature', 23.8, 'celsius');

$stats = $collector->getStats('device_001', 'temperature', 24);
echo "温度统计: " . json_encode($stats, JSON_UNESCAPED_UNICODE) . "\n";
?>

PHP在物联网场景中的优势是部署简单、开发快速。通过MQTT协议采集传感器数据，存入时序数据库，可以构建简单的物联网数据采集系统。对于大规模的物联网平台，建议使用专门的物联网平台或时序数据库来实现。