1、安装及启动：

• 参考：https://my.oschina.net/u/3523423/blog/1618130

2、业务逻辑：

• 每接受50条消息，插入一次数据库
• 若半小时内还未接受到50条消息，将已有的数据先插入

3、代码实现：

• 安装依赖：

composer require php-amqplib/php-amqplib

• 发送消息：

    private function getConn()
    {
        return new AMQPStreamConnection(
            C('RABBITMQ.host'),
            C('RABBITMQ.port'),
            C('RABBITMQ.login'),
            C('RABBITMQ.password'),
            C('RABBITMQ.vhost')
        );
    }

    /**
     * 此方法向rabbitmq服务发送模拟数据
     */
    public function send()
    {
        $connection = $this->getConn();
        /** @var AMQPChannel $channel */
        $channel = $connection->channel();

        $channel->exchange_declare(self::EXCHANGE, self::EXCHANGE_TYPE, false, true, false);
        $body = '{"mailno":"3102478466' . rand(100, 999) . '","callTime":"2019-06-03 20:00:51","ringTime":"10","returnAnswerTime":' . rand(0, 50) . ',"businessType":"delivery"}';
        $msg = new AMQPMessage($body);
        $channel->basic_publish($msg, self::EXCHANGE, self::ROUTING_KEY);

        $this->shutdown($channel, $connection);
    }

• 接受数据 ：

循环50次消息消费
利用超时异常解决半小时无消息进来的业务

    /**
     * @throws ErrorException
     */
    public function receive()
    {
        $connection = $this->getConn();
        /** @var AMQPChannel $channel */
        $channel = $connection->channel();

        $channel->exchange_declare(self::EXCHANGE, self::EXCHANGE_TYPE, false, true, false);
        $channel->queue_declare(self::QUEUE, false, true, false, false);
        $channel->queue_bind(self::QUEUE, self::EXCHANGE, self::ROUTING_KEY);

        $iSql = '';
        $iSqlPre = "INSERT ……";
        $expirePeriod = 30 * 60;

        // 每次处理50条数据
        for ($i = 0; $i < self::PROCESS_NUM_PT; $i++) {
            $channel->basic_consume(
                self::QUEUE,
                self::TAG . uniqid() . (string)$i,
                false,
                false,
                false,
                false,
                function ($msg) use ($iSqlPre, $i, &$iSql) {
                    $this->process_message($msg, $iSqlPre, $i + 1, $iSql);
                }
            );
        }

        try {
            // 30分钟内未接到新的消息则提前插入已存在的数据
            while ($channel->is_consuming()) {
                $channel->wait(null, false, $expirePeriod);
            }
        } catch (AMQPTimeoutException $e) {
            Log::write(
                sprintf(
                    "%s: %s, start inserting data due to no message in 30 minutes.",
                    self::TAG,
                    $e->getMessage()
                )
            );
            if (!empty($iSql)) {
                $iSql = sprintf($iSqlPre, rtrim($iSql, ','));
                ……
            }
            $this->shutdown($channel, $connection);
            $this->receive();
        }

        $this->shutdown($channel, $connection);
    }

    /**
     * @param AMQPMessage $message
     * @param string $iSqlPre
     * @param int $num
     * @param string $iSql
     */
    private function process_message($message, $iSqlPre, $num, &$iSql)
    {
        if (empty($message->body)) {
            return;
        }

        /** @var AMQPChannel $channel */
        $channel = $message->delivery_info['channel'];

        // 1表示未更新，2表示已更新及已acknowledge
        $state = 1;

        try {
            $data = json_decode($message->body, true);

            if ($data['businessType'] != 'delivery') {
                return;
            }

            // 更新
            ……

            // 插入
            $iSql .= 

            ……

            $channel->basic_ack($message->delivery_info['delivery_tag']);

            $state = 2;

            if ($num >= self::PROCESS_NUM_PT) {
                $iSql = sprintf($iSqlPre, rtrim($iSql, ','));
                ……
                $iSql = '';
            }
        } catch (Exception $e) {
            if ($state == 1) {
                $channel->basic_nack($message->delivery_info['delivery_tag'], true, true);
            }
            if ($num >= self::PROCESS_NUM_PT && (!empty($iSql))) {
                $this->redis->lPush(sprintf('%s:%s', self::TAG, 'Exp'), $iSql);
                $iSql = '';
            }
            Log::write(
                sprintf(
                    "%s: %s, body of the message: %s",
                    self::TAG,
                    $e->getMessage(),
                    $message->body
                )
            );
        }
    }

• 循环函数嵌套（递归）次数限制

报错如下：
PHP Fatal error: Maximum function nesting level of '500' reached, aborting! in /mnt/d/…… on line 185
PHP Stack trace:……

解决方法：

namespace ……下加入下行代码(top观察了一会，没啥异常，暂不知道是否有隐患)：

ini_set('xdebug.max_nesting_level', -1);

或者不使用receive自身嵌套，使用handle代替，每分钟执行一次handle，crontab -e 中加入：

* * * * * nohup php /……/index.php Module/Controller/handle > /dev/null &

    /**
     * @throws ErrorException
     */
    public function handle()
    {
        if (!$this->redis->exists(self::EXPIRE_KEY)) {
            $this->redis->set(self::EXPIRE_KEY,1);
            $this->receive();
        }
    }

    /**
     * @throws ErrorException
     */
    public function receive()
    {
        ……
        } catch (AMQPTimeoutException $e) {
            $this->redis->del(self::EXPIRE_KEY);
            Log::write(
                sprintf(
                    "%s: %s, start inserting data due to no message in 30 minutes.",
                    self::TAG,
                    $e->getMessage()
                )
            );
            if (!empty($iSql)) {
                $iSql = sprintf($iSqlPre, rtrim($iSql, ','));
                ……
            }
            $this->shutdown($channel, $connection);
        }

        $this->shutdown($channel, $connection);
    }

keyword: rabbitmq 定量处理消息 指定时间内 无消息推送