Stomp隊(duì)列

Stomp是簡(jiǎn)單(流)文本定向消息協(xié)議，它提供了一個(gè)可互操作的連接格式，允許STOMP客戶端與任意STOMP消息代理（Broker）進(jìn)行交互。workerman/stomp實(shí)現(xiàn)了Stomp客戶端，主要用于 RabbitMQ、Apollo、ActiveMQ 等消息隊(duì)列場(chǎng)景。

安裝

composer require webman/stomp

配置

配置文件在 config/plugin/webman/stomp 下

投遞消息

<?php
namespace app\controller;

use support\Request;
use Webman\Stomp\Client;

class Index
{
    public function queue(Request $request)
    {
        // 隊(duì)列
        $queue = 'examples';
        // 數(shù)據(jù)（傳遞數(shù)組時(shí)需要自行序列化，比如使用json_encode，serialize等）
        $data = json_encode(['to' => 'tom@gmail.com', 'content' => 'hello']);
        // 執(zhí)行投遞
        Client::send($queue, $data);

        return response('redis queue test');
    }

}

為了兼容其它項(xiàng)目，Stomp組件沒有提供自動(dòng)序列化反序列化功能，如果投遞的是數(shù)組數(shù)據(jù)，需要自行序列化，消費(fèi)的時(shí)候自行反序列化

消費(fèi)消息

新建 app/queue/stomp/MyMailSend.php (類名任意，符合psr4規(guī)范即可)。

<?php
namespace app\queue\stomp;

use Workerman\Stomp\AckResolver;
use Webman\Stomp\Consumer;

class MyMailSend implements Consumer
{
    // 隊(duì)列名
    public $queue = 'examples';

    // 連接名，對(duì)應(yīng) stomp.php 里的連接`
    public $connection = 'default';

    // 值為 client 時(shí)需要調(diào)用$ack_resolver->ack()告訴服務(wù)端已經(jīng)成功消費(fèi)
    // 值為 auto   時(shí)無(wú)需調(diào)用$ack_resolver->ack()
    public $ack = 'auto';

    // 消費(fèi)
    public function consume($data, AckResolver $ack_resolver = null)
    {
        // 如果是數(shù)據(jù)是數(shù)組，需要自行反序列化
        var_export(json_decode($data, true)); // 輸出 ['to' => 'tom@gmail.com', 'content' => 'hello']
        // 告訴服務(wù)端，已經(jīng)成功消費(fèi)
        $ack_resolver->ack(); // ack為 auto時(shí)可以省略此調(diào)用
    }
}

rabbitmq開啟stomp協(xié)議

rabbitmq默認(rèn)沒有開啟stomp協(xié)議，需要執(zhí)行以下命令開啟

rabbitmq-plugins enable rabbitmq_stomp

開啟后stomp的端口默認(rèn)為61613。