php ActiveMQ的發送消息,與處理消息


我們以一個簡單的用戶注冊為例,當用戶點擊注冊按鈕后,我們發送一個消息,后台php接收到該消息然后處理。

1.php代碼如下:

<?php
$stomp = new Stomp('tcp://192.168.1.222:61613');

$obj = new Stdclass();
//下面這些數據,實際中是用戶通過前端頁面post來的,這里只做演示
$obj->username = 'test';
$obj->password = '123456';
//發送一個注冊消息到隊列,我們這里模擬用戶注冊
$stomp->send('/queue/userReg', json_encode($obj));

2.php代碼如下:

<?php
$stomp = new Stomp('tcp://192.168.1.222:61613');
//訂閱只對一個有效,如果啟動多個腳本,只有一個會接收到消息
$stomp->subscribe('/queue/userReg');

while(true) {
    //判斷是否有讀取的信息
    if($stomp->hasFrame()) {
        $frame = $stomp->readFrame();

        $data = json_decode($frame->body, true);
        var_dump($data);

        //我們通過獲取的數據
        //處理相應的邏輯,比如存入數據庫,發送驗證碼等一系列操作。
        //$db->query("insert into user values('{$username}','{$password}')");
        //sendVerify();

        //表示消息被處理掉了,ack()函數很重要
        $stomp->ack($frame);
    }
    sleep(1);
}

分別運行上面兩個腳本文件

> /data/php56/bin/php 1.php
> /data/php56/bin/php 2.php

我們還可以把上面的2.php代碼分成多步執行。

2.php代碼如下:

<?php
$stomp = new Stomp('tcp://192.168.1.222:61613');
$stomp->subscribe('/queue/userReg');

while(true) {
    //判斷是否有讀取的信息
    if($stomp->hasFrame()) {
        $frame = $stomp->readFrame();

        $data = json_decode($frame->body, true);

        //注冊信息入庫
        //$ret = db->query("insert into user values('{$data['username']}', '{$data['password']}')");
        //這里演示直接設成true了
        $ret = true;
        if($ret) {
            echo $data['username'], '入庫成功', PHP_EOL;
            //如果入庫成功,再次把數據發送到另一個消息隊列中,進行下一步處理
            $stomp->send('/queue/sendVerify', $frame->body);

            $stomp->ack($frame);
        }
    }
    sleep(1);
}

3.php代碼如下:

<?php
$stomp = new Stomp('tcp://192.168.1.222:61613');
$stomp->subscribe('/queue/sendVerify');

while(true) {
    //判斷是否有讀取的信息
    if($stomp->hasFrame()) {
        $frame = $stomp->readFrame();

        $data = json_decode($frame->body, true);

        //$ret = sendVerify()發送驗證碼,實際中應該是請求某接口
        $ret = true;
        if($ret) {
            echo $data['username'], '發送驗證碼成功', PHP_EOL;

            $stomp->ack($frame);
        }
    }
    sleep(1);
}

再次分別運行上面的三個腳本

> /data/php56/bin/php 1.php
> /data/php56/bin/php 2.php
> /data/php56/bin/php 3.php


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM