用guzzle和phpunit测试基于mongo应用的聊天服务的性能

PointTest.php

<?php

use GuzzleHttp\Psr7\Request;
use GuzzleHttp\Client;
use GuzzleHttp\Pool;
use Monolog\Logger;
use Monolog\Handler\StreamHandler;

class PointTest extends PHPUnit_Framework_TestCase
{
    public function setUp()
    {
        $this->url = 'http://dev.chatapi.here.cn/chat/point';
        $this->clientInfo = [
            
        ];
        // 建立mongo链接
        $dsn = 'mongodb://192.168.100.186,192.168.100.133:20301/?replicaSet=dmmongo';
        // $dsn = 'mongodb://192.168.100.133:20301';
        $mongoClient = new MongoDB\Client($dsn);
        $this->pointMsg = $mongoClient->chat->jiyu_point_msg; // mongo中的点聊表

        $this->log = new Logger('PointTest');
        $this->log->pushHandler(new StreamHandler('/tmp/pointtest.log', Logger::INFO));

        $this->uid = 2634258;
        $this->ruid = 2634203;
        $this->tkey = '2634203_2634258';
        $this->allRequestNum = 30;
        $this->maxConcurrent = 1;
        $this->id = '';
        $this->counter = 0; // 响应计数器
        $this->wait = 300000; // 单位us
        $this->prevId = ''; // 上一个拉消息操作返回的id
        $this->nextId = ''; // 下一个拉消息操作返回的id

        // 清除数据
        $this->pointMsg->deleteMany(['tkey' => $this->tkey]);
    }

    // 清空测试数据
    public function tearDown()
    {
        // 清除数据
        // $this->pointMsg->deleteMany(['tkey' => $this->tkey]);
    }

    // 测试发布点聊
    public function testStore()
    {
        $url = $this->url.'/store';
        $msg = '测试聊天'.'-'.date('Y-m-d');
        $verifystr = md5($msg);
        $data = [
            'client_info' => $this->clientInfo,
            'apptoken' => 'dmaitoken01',
            'uid' => 2634258,
            'ruid' => 2634203,
            'data' => $msg,
            'verifystr' => $verifystr,
            'data_type' => 1,
        ]; // 发送给聊天服务的数据
        $request = new Request('POST', $url, [], json_encode($data));
        $client = new Client;
        $response = $client->send($request);
        $s = (string)$response->getBody();

        $this->assertJson($s);
        $arr = json_decode($s, true);
        $this->assertEquals($arr['success'], 1); // success字段为1
        $this->assertEquals(strlen($arr['_id']), 24);

        return $arr['_id'];
    }

    /**
     * 测试mongo中有写入的那一条记录
     *
     * @depends testStore
     */
    public function testPointMsg()
    {
        $arr = func_get_args();
        $id = new MongoDB\BSON\ObjectID($arr[0]);
        $result = $this->pointMsg->find(['_id' => $id]);
        $this->assertNotEmpty($result);
        foreach ($result as $row) {
            $this->assertNotEmpty($row);
            $this->assertNotEmpty($row['_id']);
            $this->assertEquals(strlen($row['_id']), 24); // id 24位
            $this->assertNotEmpty($row['suid']);
            $this->assertNotEmpty($row['ruid']);
            $this->assertNotEmpty($row['tkey']);
        }
    }

    // 测试异步请求
    public function testAsyncGet()
    {
        $url = $this->url.'/store';
        $uid = $this->uid;
        $ruid = $this->ruid;
        $client = new Client;

        $requests = function ($total) use ($client, $url, $uid, $ruid) {
            for ($i = 0; $i < $total; $i++) {
                $msg = '测试数据-'.$i;
                $verifystr = md5($msg);
                $data = [
                    'uid' => $uid,
                    'ruid' => $ruid,
                    'data' => $msg,
                    'verifystr' => $verifystr,
                    'data_type' => 1,
                ];
                yield new Request('POST', $url, [], json_encode($data));
            }
        };
        $success = function ($response, $index) {
            usleep($this->wait);
            $res = json_decode((string)$response->getBody(), true);

            if ($this->counter == 0) { // 第一次发消息结束之后第一次进去拉取最新消息的id
                $req2 = [
                    'uid' => $this->ruid,
                    'ruid' => $this->uid,
                    'limit' => 1,
                ];
                $res2 = $this->getMsg($req2);
                $data2 = $res2['data'];
                $this->prevId = $this->nextId = $this->id = $data2[0]['_id'];
                $this->assertEquals(strlen($this->id), 24);
                $this->log->addInfo('第一次拉取的id', ['id' => $this->id]);
                $this->log->addInfo('拉取到了消息', ['req' => $req2, 'resp' => $res2, 'index' => $index]);
            }
            if ($this->counter > 0) {
                $this->log->addInfo('后续的拉取', ['id' => $this->id, 'prevId' => $this->prevId, 'nextId' => $this->nextId]);
                $this->assertEquals(strlen($this->id), 24); // 确保第一条消息拉取成功
                $req2 = [
                    'eid' => $this->id,
                    'uid' => $this->ruid,
                    'ruid' => $this->uid,
                    'limit' => 20,
                ];
                $res2 = $this->getMsg($req2);
                $data2 = $res2['data'];

                if (empty($data2)) {
                    // $this->findById($res['_id'], $res);
                    $this->log->addError('未拉取到消息', ['req' => $req2, 'resp' => $res2, 'index' => $index]);
                    // 看看通过eid去mongo中查能否查到
                    // $this->findByEid();
                } else {
                    $this->prevId = $this->nextId;
                    $this->nextId = $data2[0]['_id'];
                    $this->id = min($this->prevId, $this->nextId);

                    // $this->findByEid();
                    $this->log->addInfo('拉取到了消息', ['req' => $req2, 'resp' => $res2, 'index' => $index]);
                }
            }
            $this->counter += 1;
        };
        $fail = function ($reason, $index) {
            echo '请求失败', PHP_EOL;
        };
        $pool = new Pool($client, $requests($this->allRequestNum), [
            'concurrency' => $this->maxConcurrent,
            'fulfilled' => $success->bindTo($this),
            'rejected' => $fail,
        ]);

        $promise = $pool->promise();
        $promise->wait();
    }

    private function getMsg($data)
    {
        $url = $this->url.'/show';
        $request = new Request('POST', $url, [], json_encode($data));
        $client = new Client;

        $response = $client->send($request);
        $res = json_decode((string)$response->getBody(), true);
        return $res;
    }

    // mongo插入之后会不会立刻排序????
    private function findByEid()
    {
        $mongoId = new MongoDB\BSON\ObjectID($this->id);
        $filter = ['tkey' => $this->tkey, '_id' => ['$gt' => $mongoId]];
        $mongoRes = $this->pointMsg->findOne($filter);
        if (empty($mongoRes)) {
            $this->log->addError('通过eid去mongo中查不到结果', $filter);
        } else {
            $this->log->addError('通过eid去mongo中查到结果', ['filter' => $filter, 'data' => $mongoRes->data]);
        }
    }

    private function findById($id, $res)
    {
        $mongoId = new MongoDB\BSON\ObjectID($id);
        $mongoRes = $this->pointMsg->findOne(['_id' => $mongoId]); // 去mongo中查看最新发的消息是否写入了
        if (empty($mongoRes)) {
            $this->log->addError('mongo中还未写入', ['id' => (string)$mongoId, 'resp' => $res]);
            sleep(1);
            $mongoRes = $this->pointMsg->findOne(['_id' => $mongoId]);
            if (empty($mongoRes)) {
                $this->log->addError('休眠1s之后mongo中还未写入', ['id' => (string)$mongoId, 'resp' => $res]);
                sleep(3);
                $mongoRes = $this->pointMsg->findOne(['_id' => $mongoId]);
                if (empty($mongoRes)) {
                    $this->log->addError('再次休眠3s之后mongo中还未写入', ['id' => (string)$mongoId, 'resp' => $res]);
                } else {
                    $this->log->addInfo('再次休眠3s之后mongo中已写入', ['id' => (string)$mongoId, 'resp' => $res, 'data' => $mongoRes->data]);
                }
            } else {
                $this->log->addInfo('休眠1s之后mongo中已写入', ['id' => (string)$mongoId, 'resp' => $res, 'data' => $mongoRes->data]);
            }
        } else {
            $this->log->addInfo('mongo中已写入', ['id' => (string)$mongoId, 'resp' => $res, 'data' => $mongoRes->data]);
        }
    }
}