<?php
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
class RabbitMQ_InventoryBatch
{
protected $_conn;
public function __construct()
{
$mq_config = new Zend_Config_Ini(APPLICATION_PATH . '/../application/configs/mq.ini', 'production',true);
$this->host = $mq_config->mq->host;
$this->port = $mq_config->mq->port;
$this->user = $mq_config->mq->user;
$this->pass = $mq_config->mq->pass;
//获得连接
$this->_conn = new AMQPStreamConnection($this->host, $this->port, $this->user, $this->pass);
}
public function pusher($msg_body){
$exchange = 'exchange1';
$queue = 'Inventory';
//建立mq通道
$channel = $this->_conn->channel();
/*
*
声明通道
name: $queue
passive: false
durable: true // the queue will survive server restarts
exclusive: false // the queue can be accessed in other channels
auto_delete: false //the queue won't be deleted once the channel is closed.
*/
$channel->queue_declare($queue, false, false, false, false);
/*
*
name: $exchange
type: direct
passive: false
durable: true // the exchange will survive server restarts
auto_delete: false //the exchange won't be deleted once the channel is closed.
*/
//fanout
$channel->exchange_declare($exchange, 'direct', false, true, false);
//绑定exchange
$channel->queue_bind($queue, $exchange);
if(empty($msg_body)) $msg_body = "Hello World!";
if(is_array($msg_body)){
$msg_body=serialize($msg_body);
}
$msg = new AMQPMessage($msg_body, array('content_type' => 'text/plain', 'delivery-mode' => 2));
$channel->basic_publish($msg, $exchange);
$channel->close();
$this->_conn->close();
}
}
有疑问加站长微信联系（非本文作者））
