<?php

require_once "./vendor/autoload.php";

use PhpAmqpLib\Connection\AMQPStreamConnection;

use PhpAmqpLib\Message\AMQPMessage;

$conf = [

	'host' => '192.168.33.50',

	'port' => 5672,

	'user' => 'admin',

	'pwd' => 'password',

	'vhost' => '/',

];

$exchangeName = 'tpc'; //交换机名

//$queueName = 'ttt'; //队列名称

$routingKey = 'lazy.a.b'; //路由关键字(也可以省略)

$conn = new AMQPStreamConnection( //建立生产者与mq之间的连接

	$conf['host'], $conf['port'], $conf['user'], $conf['pwd'], $conf['vhost']

);

$channel = $conn->channel(); //在已连接基础上建立生产者与mq之间的通道

$channel->exchange_declare($exchangeName, 'topic', false, true, false); //声明初始化交换机

//$channel->queue_declare($queueName, false, true, false, false); //声明初始化一条队列

//$channel->queue_bind($queueName, $exchangeName, $routingKey); //将队列与某个交换机进行绑定，并使用路由关键字

$msgBody = json_encode(["name" => "iGoo", "age" => 22]);

$msg = new AMQPMessage($msgBody, ['content_type' => 'text/plain', 'delivery_mode' => 2]); //生成消息

$r = $channel->basic_publish($msg, $exchangeName, $routingKey); //推送消息到某个交换机

$channel->close();

$conn->close();

<?php

/**

 * Created by PhpStorm.

 * User: jmsite.cn

 * Date: 2019/1/15

 * Time: 13:16

 */

//声明连接参数

$config = array(

	'host' => '192.168.33.50',

	'vhost' => '/',

	'port' => 5672,

	'login' => 'admin',

	'password' => 'password'

);

//连接broker

$cnn = new AMQPConnection($config);

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

	echo "Cannot connect to the broker";

	exit();

}

//在连接内创建一个通道

$ch = new AMQPChannel($cnn);

//创建一个交换机

$ex = new AMQPExchange($ch);

//声明路由键

$routingKey = 'lazy.#';

//声明交换机名称

$exchangeName = 'tpc';

//设置交换机名称

$ex->setName($exchangeName);

//设置交换机类型

//AMQP_EX_TYPE_DIRECT:直连交换机

//AMQP_EX_TYPE_FANOUT:扇形交换机

//AMQP_EX_TYPE_HEADERS:头交换机

//AMQP_EX_TYPE_TOPIC:主题交换机

$ex->setType(AMQP_EX_TYPE_TOPIC);

//设置交换机持久

$ex->setFlags(AMQP_DURABLE);

//声明交换机

$ex->declareExchange();

////创建一个消息队列

$q = new AMQPQueue($ch);

//设置队列名称

$q->setName('Q2');

//设置队列持久

$q->setFlags(AMQP_DURABLE);

//声明消息队列

$q->declareQueue();

//交换机和队列通过$routingKey进行绑定

//$q->bind($ex->getName(), $routingKey);

//接收消息并进行处理的回调方法

function receive($envelope, $queue) {

	//休眠两秒，

	sleep(2);

	//echo消息内容

	echo $envelope->getBody()."\n";

	//显式确认，队列收到消费者显式确认后，会删除该消息

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

}

//设置消息队列消费者回调方法，并进行阻塞

$q->consume("receive");

//$q->consume("receive", AMQP_AUTOACK);//隐式确认,不推荐