PHP实现RabbitMQ的Publish/Subscribe

<?php

/**

 * Created by PhpStorm.

 * User: 豆腐居士

 * Date: 2018/5/30

 * Time: 上午11:01

 */

class AqiTask extends BaseTask

{

    const EX_NAME = 'aqi_fanout';

    function clientAction()

    {

        $conn = new AMQPConnection([

            'host' => '0.0.0.0',

            'port' => 5672,

            'login' => 'wangwang',

            'password' => '123456'

        ]);

        if (!$conn->connect()) {

            exit('fail');

        }

        $channel = new AMQPChannel($conn);

        $ex = new AMQPExchange($channel);

        $ex->setName(self::EX_NAME);

        $ex->setType(AMQP_EX_TYPE_FANOUT); //direct类型

        $ex->setFlags(AMQP_DURABLE); //持久化

        $ex->declareExchange();

        for ($i = 0; $i < 10; $i++) {

            $msg = 'hello fanout:' . $i;

            echo $msg . PHP_EOL;

            $ex->publish($msg);

        }

        $conn->disconnect();

    }

    function server1Action()

    {

        $conn = new AMQPConnection([

            'host' => '0.0.0.0',

            'port' => 5672,

            'login' => 'wangwang',

            'password' => '123456'

        ]);

        if (!$conn->connect()) {

            exit('fail');

        }

        $channel = new AMQPChannel($conn);

        $ex = new AMQPExchange($channel);

        $ex->setName(self::EX_NAME);

        $ex->setType(AMQP_EX_TYPE_FANOUT); //fanout类型

        $ex->setFlags(AMQP_DURABLE); //持久化

        $ex->declareExchange();

        $queue = new AMQPQueue($channel);

        $queue->setName('q1');

        $queue->setFlags(AMQP_DURABLE);

        $queue->declareQueue();

        $queue->bind($ex->getName());try {

             $queue->consume(function ($envelope, $q) {

                 echo 'server1:' . $envelope->getBody(), PHP_EOL;

                 $q->ack($envelope->getDeliveryTag()); //这里是手动应答  如果使用默认的自动应答，进程退出后消息会丢失

             });

        } catch (\Exception $e) {

            echo $e->getMessage();

        }

    }

    function server2Action()

    {

        $conn = new AMQPConnection([

            'host' => '0.0.0.0',

            'port' => 5672,

            'login' => 'wangwang',

            'password' => '123456'

        ]);

        if (!$conn->connect()) {

            exit('fail');

        }

        $channel = new AMQPChannel($conn);

        $ex = new AMQPExchange($channel);

        $ex->setName(self::EX_NAME);

        $ex->setType(AMQP_EX_TYPE_FANOUT);

        $ex->setFlags(AMQP_DURABLE);

        $ex->declareExchange();

        $queue = new AMQPQueue($channel);

        $queue->setName('q2');

        $queue->setFlags(AMQP_DURABLE);

        $queue->declareQueue();

        $queue->bind($ex->getName());$queue->consume(function ($envelope, $q) {

             echo 'server2:' . $envelope->getBody(), PHP_EOL;

             $q->ack($envelope->getDeliveryTag());

        });

    }

}