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

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

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

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

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

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

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

> /data/php56/bin/php 1.php
> /data/php56/bin/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);

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

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

<?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()發(fā)送驗(yàn)證碼，實(shí)際中應(yīng)該是請(qǐng)求某接口
  $ret = true;
  if($ret) {
   echo $data['username'], '發(fā)送驗(yàn)證碼成功', 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