webman与influxdb数据库的如何实现长连接？

lvhejin 2023-07-19

问题描述

有人实现webman与influxdb数据库的长连接吗？官方示例提供的示例如果频繁写入会造成TCP连接数过多

为此你搜索到了哪些方案及不适用的原因

https://github.com/influxdata/influxdb-client-php

 1051  1 1

1个回答

智佳思远 2023-07-20

app/functions.php 里加一个函数

function influxdb() {
    static $client;
    if (!$client) {
        $client = new Client(....);
    }
    return $client
}

调用的时候就 influxdb()->xxx()

lvhejin 2023-07-21

app/functions.php

<?php
/**
 * Here is your custom functions.
 */

use InfluxDB2\Client;
use InfluxDB2\Model\WritePrecision;
use InfluxDB2\Point;
use InfluxDB2\WriteType as WriteType;

function influxdb() {

    $token = 'xxxxxxxxx';
    $org = 'test';
    $bucket = 'dbname';

    static $influxClient;
    if (!$influxClient) {
        $influxClient = new Client([
            "url" => "http://127.0.0.1:8086",
            "token" => $token,
            "bucket" => $bucket,
            "org" => $org,
            "precision" => WritePrecision::S
        ]);
    }
    return $influxClient;
}

queue/redis/Dynamic.php

<?php

namespace app\queue\redis;

use Webman\RedisQueue\Consumer;
use InfluxDB2\Client;
use InfluxDB2\Model\WritePrecision;
use InfluxDB2\Point;
use InfluxDB2\WriteType as WriteType;

class Dynamic implements Consumer
{
    // 要消费的队列名
    public $queue = 'send-dynamic';
    // 连接名，对应 plugin/webman/redis-queue/redis.php 里的连接`
    public $connection = 'default';

    public $token = 'iv16aGoyk6NsVPRpZT4vdtMUTjNoax-ijJ6kazqXa9bwBbShpZEFx2T7mqU-Ep1IDiYmB-k0LcDDTFaqRrye2A==';
    public $org = 'jtkj';
    public $bucket = 'bridgedb';

    // 消费
    public function consume($dataJson)
    {

        $deviceId = $dataJson['deviceId'];
        $dataCount = $dataJson['count'];
        $dataTime = $dataJson['time'];
        $dataData = $dataJson['data'];

        try {
                //echo $dataJson['deviceId'] . "\r\n";

                $writeApi = influxdb()->createWriteApi(["writeType" => WriteType::BATCHING, 'batchSize' => $dataCount, 'maxRetries' => 1, 'maxRetryDelay' => 10000, 'maxRetryTime' => 10000]);

                for ($i = 0; $i < $dataCount; $i++) {
                    $point = Point::measurement(strval($deviceId))
                        ->addTag('channel', strval($dataData[$i]['channel']))
                        ->addField('originalValue1', (double)$dataData[$i]['originalValue1'][0])
                        ->addField('calculatedValue1', (double)$dataData[$i]['originalValue1'][0])
                        ->time(strtotime($dataTime));
                    $writeApi->write($point);
                }
                $writeApi->close();
            } catch (\Error $e) {
                print "\n\n $e \n\n";
            influxdb()->close();
            }
        influxdb()->close();
    }
}

这样写还是不行，还是一样报这个警告

[WARNING] - The retryable error occurred during writing of data. Reason: '[0] Failed to connect to 192.168.100.60 port 8086: Address already in use'. Retry in: 9.685s.