PHP RabbitMQ相关

$message = [
    'ticket_id' => 10102
];
$message = json_encode($message);
// 交换机名称，自己命名
$exchange = 'async';
// 路由名称，自己命名
$routing = 'ticket';

// RabbitMQ连接配置信息
$setting => [
    'host'     => 'dev.rabbitmq.demo.com',
    'port'     => '5672',
    'login'    => 'admin',
    'password' => '123+-*abc'
],

// RabbitMQ连接
$conn = new \AMQPConnection($setting);  // 创建对象
$conn->connect();
$channel = new \AMQPChannel($conn);  // 创建channel
$ex = new \AMQPExchange($channel);  // 创建交换机
$ex->setName($exchange);  // 创建交换机名字 
$ex->setType(AMQP_EX_TYPE_DIRECT);  // 设置交换机类型
$ex->setFlags(AMQP_DURABLE);  // 设置模式
$ex->declareExchange();

// 发布消息到交换机并指定路由关键字
if ($ex->publish($message, $routing, 1)) {
    echo 'succ';
}
echo 'end';

/**
 * Create an instance of AMQPConnection.
 *
 * Creates an AMQPConnection instance representing a connection to an AMQP
 * broker. A connection will not be established until
 * AMQPConnection::connect() is called.
 *
 *  $credentials = array(
 *      'host'  => amqp.host The host to connect too. Note: Max 1024 characters.
 *      'port'  => amqp.port Port on the host.
 *      'vhost' => amqp.vhost The virtual host on the host. Note: Max 128 characters.
 *      'login' => amqp.login The login name to use. Note: Max 128 characters.
 *      'password' => amqp.password Password. Note: Max 128 characters.
 *      'read_timeout'  => Timeout in for income activity. Note: 0 or greater seconds. May be fractional.
 *      'write_timeout' => Timeout in for outcome activity. Note: 0 or greater seconds. May be fractional.
 *      'connect_timeout' => Connection timeout. Note: 0 or greater seconds. May be fractional.
 *
 *      Connection tuning options (see https://www.rabbitmq.com/amqp-0-9-1-reference.html#connection.tune for details):
 *      'channel_max' => Specifies highest channel number that the server permits. 0 means standard extension limit
 *                       (see PHP_AMQP_MAX_CHANNELS constant)
 *      'frame_max'   => The largest frame size that the server proposes for the connection, including frame header
 *                       and end-byte. 0 means standard extension limit (depends on librabbimq default frame size limit)
 *      'heartbeat'   => The delay, in seconds, of the connection heartbeat that the server wants.
 *                       0 means the server does not want a heartbeat. Note, librabbitmq has limited heartbeat support,
 *                       which means heartbeats checked only during blocking calls.这个参数可以设置延长心跳检查时间
 *
 *      TLS support (see https://www.rabbitmq.com/ssl.html for details):
 *      'cacert' => Path to the CA cert file in PEM format..
 *      'cert'   => Path to the client certificate in PEM foramt.
 *      'key'    => Path to the client key in PEM format.
 *      'verify' => Enable or disable peer verification. If peer verification is enabled then the common name in the
 *                  server certificate must match the server name. Peer verification is enabled by default.
 * )
 *
 * @param array $credentials Optional array of credential information for
 *                           connecting to the AMQP broker.
 */
public function __construct(array $credentials = array()) { }

$conn = new \AMQPConnection($setting); 
if (!$conn->->isConnected()) {
    $conn->connect();
}

/**
 * Establish a transient connection with the AMQP broker.
 *
 * This method will initiate a connection with the AMQP broker.
 *
 * @throws AMQPConnectionException
 * @return boolean TRUE on success or throw an exception on failure.
 */
public function connect() { }

/**
 * Check whether the connection to the AMQP broker is still valid.
 *
 * It does so by checking the return status of the last connect-command.
 *
 * @return boolean True if connected, false otherwise.
 */
public function isConnected() { }

/**
 * Create an instance of an AMQPChannel object.
 *
 * @param AMQPConnection $amqp_connection An instance of AMQPConnection
 *                                        with an active connection to a broker.
 *
 * @throws AMQPConnectionException        If the connection to the broker was lost.
 */
public function __construct(AMQPConnection $amqp_connection) { }

/**
 * Create an instance of AMQPExchange.
 *
 * Returns a new instance of an AMQPExchange object, associated with the
 * given AMQPChannel object.
 *
 * @param AMQPChannel $amqp_channel A valid AMQPChannel object, connected to a broker.
 *                                  
 * @throws AMQPExchangeException   When amqp_channel is not connected to a broker.                           
 * @throws AMQPConnectionException If the connection to the broker was lost.
 *                                 
 */
public function __construct(AMQPChannel $amqp_channel) { }

/**
 * Set the name of the exchange.
 *
 * @param string $exchange_name The name of the exchange to set as string.
 *
 * @return void
 */
public function setName($exchange_name) { }

/**
 * Set the type of the exchange.
 *
 * Set the type of the exchange. This can be any of AMQP_EX_TYPE_DIRECT,
 * AMQP_EX_TYPE_FANOUT, AMQP_EX_TYPE_HEADERS or AMQP_EX_TYPE_TOPIC.
 *
 * @param string $exchange_type The type of exchange as a string.
 *
 * @return void
 */
public function setType($exchange_type) { }

/**
 * Set the flags on an exchange.
 *
 * @param integer $flags A bitmask of flags. This call currently only
 *                       considers the following flags:
 *                       AMQP_DURABLE, AMQP_PASSIVE
 *                       (and AMQP_DURABLE, if librabbitmq version >= 0.5.3)
 *
 * @return void
 */
public function setFlags($flags) { }

/**
 * Declare a new exchange on the broker.
 *
 * @throws AMQPExchangeException   On failure.
 * @throws AMQPChannelException    If the channel is not open.
 * @throws AMQPConnectionException If the connection to the broker was lost.
 *
 * @return boolean TRUE on success or FALSE on failure.
 */
public function declareExchange() { }

/**
 * Publish a message to an exchange.
 *
 * Publish a message to the exchange represented by the AMQPExchange object.
 *
 * @param string  $message     The message to publish.
 * @param string  $routing_key The optional routing key to which to
 *                             publish to.
 * @param integer $flags       One or more of AMQP_MANDATORY and
 *                             AMQP_IMMEDIATE.
 * @param array   $attributes  One of content_type, content_encoding,
 *                             message_id, user_id, app_id, delivery_mode,
 *                             priority, timestamp, expiration, type
 *                             or reply_to, headers.
 *
 * @throws AMQPExchangeException   On failure.
 * @throws AMQPChannelException    If the channel is not open.
 * @throws AMQPConnectionException If the connection to the broker was lost.
 *
 * @return boolean TRUE on success or FALSE on failure.
 */
public function publish(
    $message,
    $routing_key = null,
    $flags = AMQP_NOPARAM,
    array $attributes = array()
) {
}

/**
 * Passing in this constant as a flag will forcefully disable all other flags.
 * Use this if you want to temporarily disable the amqp.auto_ack ini setting.
 */
define('AMQP_NOPARAM', 0);

/**
 * Passing in this constant as a flag to proper methods will forcefully ignore all other flags.
 * Do not send basic.consume request during AMQPQueue::consume(). Use this if you want to run callback on top of previously
 * declared consumers.
 */
define('AMQP_JUST_CONSUME', 1);

/**
 * Durable exchanges and queues will survive a broker restart, complete with all of their data.
 */
define('AMQP_DURABLE', 2);

/**
 * Passive exchanges and queues will not be redeclared, but the broker will throw an error if the exchange or queue does not exist.
 */
define('AMQP_PASSIVE', 4);

/**
 * Valid for queues only, this flag indicates that only one client can be listening to and consuming from this queue.
 */
define('AMQP_EXCLUSIVE', 8);

/**
 * For exchanges, the auto delete flag indicates that the exchange will be deleted as soon as no more queues are bound
 * to it. If no queues were ever bound the exchange, the exchange will never be deleted. For queues, the auto delete
 * flag indicates that the queue will be deleted as soon as there are no more listeners subscribed to it. If no
 * subscription has ever been active, the queue will never be deleted. Note: Exclusive queues will always be
 * automatically deleted with the client disconnects.
 */
define('AMQP_AUTODELETE', 16);

/**
 * Clients are not allowed to make specific queue bindings to exchanges defined with this flag.
 */
define('AMQP_INTERNAL', 32);

/**
 * When passed to the consume method for a clustered environment, do not consume from the local node.
 */
define('AMQP_NOLOCAL', 64);

/**
 * When passed to the {@link AMQPQueue::get()} and {@link AMQPQueue::consume()} methods as a flag,
 * the messages will be immediately marked as acknowledged by the server upon delivery.
 */
define('AMQP_AUTOACK', 128);

/**
 * Passed on queue creation, this flag indicates that the queue should be deleted if it becomes empty.
 */
define('AMQP_IFEMPTY', 256);

/**
 * Passed on queue or exchange creation, this flag indicates that the queue or exchange should be
 * deleted when no clients are connected to the given queue or exchange.
 */
define('AMQP_IFUNUSED', 512);

/**
 * When publishing a message, the message must be routed to a valid queue. If it is not, an error will be returned.
 */
define('AMQP_MANDATORY', 1024);

/**
 * When publishing a message, mark this message for immediate processing by the broker. (High priority message.)
 */
define('AMQP_IMMEDIATE', 2048);

/**
 * If set during a call to {@link AMQPQueue::ack()}, the delivery tag is treated as "up to and including", so that multiple
 * messages can be acknowledged with a single method. If set to zero, the delivery tag refers to a single message.
 * If the AMQP_MULTIPLE flag is set, and the delivery tag is zero, this indicates acknowledgement of all outstanding
 * messages.
 */
define('AMQP_MULTIPLE', 4096);

/**
 * If set during a call to {@link AMQPExchange::bind()}, the server will not respond to the method.The client should not wait
 * for a reply method. If the server could not complete the method it will raise a channel or connection exception.
 */
define('AMQP_NOWAIT', 8192);

/**
 * If set during a call to {@link AMQPQueue::nack()}, the message will be placed back to the queue.
 */
define('AMQP_REQUEUE', 16384);

$attributes = [
    "delivery_mode" => 2,
    "expiration" => 10000
];

// 连接名称，自己命名
$routing_key = 'ticket';

// RabbitMQ配置信息
$setting => [
    'host'     => 'dev.rabbitmq.demo.com',
    'port'     => '5672',
    'login'    => 'admin',
    'password' => '123+-*abc'
],

// RabbitMQ连接
$conn = new \AMQPConnection($setting);  // 创建对象
$conn->connect();
$channel = new \AMQPChannel($conn);  // 创建channel
$q = new \AMQPQueue($channel);  // 创建队列
$q->setName('queue_ticket');  // 设置队列名称
$q->setFlags(AMQP_DURABLE);　　// 设置模式
$q->declareQueue();
$q->bind('async', $routing_key);  // 绑定交换机
$messages = $q->get(AMQP_AUTOACK);  // 获取消息
// 处理消息
if ($messages) {
    var_dump(json_decode($messages->getBody(), true));
}
$conn->disconnect();  // 断开连接

/**
 * Create an instance of an AMQPQueue object.
 *
 * @param AMQPChannel $amqp_channel The amqp channel to use.
 *
 * @throws AMQPQueueException      When amqp_channel is not connected to a
 *                                 broker.
 * @throws AMQPConnectionException If the connection to the broker was lost.
 */
public function __construct(AMQPChannel $amqp_channel) { }

/**
 * Set the queue name.
 *
 * @param string $queue_name The name of the queue.
 *
 * @return boolean
 */
public function setName($queue_name) { }

/**
 * Set the flags on the queue.
 *
 * @param integer $flags A bitmask of flags:
 *                       AMQP_DURABLE, AMQP_PASSIVE,
 *                       AMQP_EXCLUSIVE, AMQP_AUTODELETE.
 *
 * @return boolean
 */
public function setFlags($flags) { }

/**
 * Declare a new queue on the broker.
 *
 * @throws AMQPChannelException    If the channel is not open.
 * @throws AMQPConnectionException If the connection to the broker was lost.
 *
 * @return integer the message count.
 */
public function declareQueue() { }

/**
 * Bind the given queue to a routing key on an exchange.
 *
 * @param string $exchange_name Name of the exchange to bind to.
 * @param string $routing_key   Pattern or routing key to bind with.
 * @param array  $arguments     Additional binding arguments.
 *
 * @throws AMQPChannelException    If the channel is not open.
 * @throws AMQPConnectionException If the connection to the broker was lost.
 *
 * @return boolean
 */
public function bind($exchange_name, $routing_key = null, array $arguments = array()) { }

/**
 * Retrieve the next message from the queue.
 *
 * Retrieve the next available message from the queue. If no messages are
 * present in the queue, this function will return FALSE immediately. This
 * is a non blocking alternative to the AMQPQueue::consume() method.
 * Currently, the only supported flag for the flags parameter is
 * AMQP_AUTOACK. If this flag is passed in, then the message returned will
 * automatically be marked as acknowledged by the broker as soon as the
 * frames are sent to the client.
 *
 * @param integer $flags A bitmask of supported flags for the
 *                       method call. Currently, the only the
 *                       supported flag is AMQP_AUTOACK. If this
 *                       value is not provided, it will use the
 *                       value of ini-setting amqp.auto_ack.
 *
 * @throws AMQPChannelException    If the channel is not open.
 * @throws AMQPConnectionException If the connection to the broker was lost.
 *
 * @return AMQPEnvelope|boolean
 */
public function get($flags = AMQP_NOPARAM) { }

/**
 * Closes the transient connection with the AMQP broker.
 *
 * This method will close an open connection with the AMQP broker.
 *
 * @return boolean true if connection was successfully closed, false otherwise.
 */
public function disconnect() { }

// 回调函数
function run($envelope, $queue)
{
    // 获取信息体
    $msg = $envelope->getBody();
    $msg = json_decode($msg, true);
    
    // 信息体逻辑处理
    ...
    
    // 确认响应
    $queue->ack($envelope->getDeliveryTag());
}
$callback = "run";

// 交换器名
$exchange_name = "async";
// routing_key 路由连接名称，自己命名
$routing_key = 'ticket';
// 队列名
$queue_name = "queue_ticket";
// 要预取的消息数
$offset = 1;

// RabbitMQ配置信息
$setting => [
    'host'     => 'dev.rabbitmq.demo.com',
    'port'     => '5672',
    'login'    => 'admin',
    'password' => '123+-*abc'
],

// RabbitMQ连接
$conn = new \AMQPConnection($setting);  // 创建对象
$conn->connect();
$channel = new \AMQPChannel($conn);  // 创建channel

// 声明交换器，以防声明队列时找不到要绑定的交换器
$ex = new \AMQPExchange($channel);
$ex->setName($exchange_name);
$ex->setType(AMQP_EX_TYPE_DIRECT);
$ex->setFlags(AMQP_DURABLE);
$ex->declareExchange();

// 声明队列
$q = new \AMQPQueue($channel);  // 创建队列
$q->setName($queue_name);  // 设置队列名称
$q->setFlags(AMQP_DURABLE);　　// 设置模式
$q->declareQueue();
$q->bind($exchange_name, $routing_key);  // 绑定交换机

$channel->qos(0, $offset);

$q->consume($callback);

$conn->disconnect();

/**
 * Set the Quality Of Service settings for the given channel.
 *
 * Specify the amount of data to prefetch in terms of window size (octets)
 * or number of messages from a queue during a AMQPQueue::consume() or
 * AMQPQueue::get() method call. The client will prefetch data up to size
 * octets or count messages from the server, whichever limit is hit first.
 * Setting either value to 0 will instruct the client to ignore that
 * particular setting. A call to AMQPChannel::qos() will overwrite any
 * values set by calling AMQPChannel::setPrefetchSize() and
 * AMQPChannel::setPrefetchCount(). If the call to either
 * AMQPQueue::consume() or AMQPQueue::get() is done with the AMQP_AUTOACK
 * flag set, the client will not do any prefetching of data, regardless of
 * the QOS settings.
 *
 * @param integer $size  The window size, in octets, to prefetch.
 * @param integer $count The number of messages to prefetch.
 *
 * @throws AMQPConnectionException If the connection to the broker was lost.
 *
 * @return bool TRUE on success or FALSE on failure.
 */
public function qos($size, $count) { }

/**
 * Consume messages from a queue.
 *
 * Blocking function that will retrieve the next message from the queue as
 * it becomes available and will pass it off to the callback.
 *
 * @param callable | null $callback    A callback function to which the
 *                              consumed message will be passed. The
 *                              function must accept at a minimum
 *                              one parameter, an AMQPEnvelope object,
 *                              and an optional second parameter
 *                              the AMQPQueue object from which callback
 *                              was invoked. The AMQPQueue::consume() will
 *                              not return the processing thread back to
 *                              the PHP script until the callback
 *                              function returns FALSE.
 *                              If the callback is omitted or null is passed,
 *                              then the messages delivered to this client will
 *                              be made available to the first real callback
 *                              registered. That allows one to have a single
 *                              callback consuming from multiple queues.
 * @param integer $flags        A bitmask of any of the flags: AMQP_AUTOACK,
 *                              AMQP_JUST_CONSUME. Note: when AMQP_JUST_CONSUME
 *                              flag used all other flags are ignored and
 *                              $consumerTag parameter has no sense.
 *                              AMQP_JUST_CONSUME flag prevent from sending
 *                              `basic.consume` request and just run $callback
 *                              if it provided. Calling method with empty $callback
 *                              and AMQP_JUST_CONSUME makes no sense.
 * @param string   $consumerTag A string describing this consumer. Used
 *                              for canceling subscriptions with cancel().
 *
 * @throws AMQPChannelException    If the channel is not open.
 * @throws AMQPConnectionException If the connection to the broker was lost.
 * @throws AMQPEnvelopeException   When no queue found for envelope.
 *
 * @return void
 */
public function consume(
    callable $callback = null,
    $flags = AMQP_NOPARAM,
    $consumerTag = null
) {
}

/**
 * stub class representing AMQPEnvelope from pecl-amqp
 */
class AMQPEnvelope extends AMQPBasicProperties
{
    public function __construct() { }

    /**
     * Get the body of the message.
     *
     * @return string The contents of the message body.
     */
    public function getBody() { }

    /**
     * Get the routing key of the message.
     *
     * @return string The message routing key.
     */
    public function getRoutingKey() { }

    /**
     * Get the consumer tag of the message.
     *
     * @return string The consumer tag of the message.
     */
    public function getConsumerTag() { }

    /**
     * Get the delivery tag of the message.
     *
     * @return string The delivery tag of the message.
     */
    public function getDeliveryTag() { }

    /**
     * Get the exchange name on which the message was published.
     *
     * @return string The exchange name on which the message was published.
     */
    public function getExchangeName() { }

    /**
     * Whether this is a redelivery of the message.
     *
     * Whether this is a redelivery of a message. If this message has been
     * delivered and AMQPEnvelope::nack() was called, the message will be put
     * back on the queue to be redelivered, at which point the message will
     * always return TRUE when this method is called.
     *
     * @return bool TRUE if this is a redelivery, FALSE otherwise.
     */
    public function isRedelivery() { }

    /**
     * Get a specific message header.
     *
     * @param string $header_key Name of the header to get the value from.
     *
     * @return string|boolean The contents of the specified header or FALSE
     *                        if not set.
     */
    public function getHeader($header_key) { }

    /**
     * Check whether specific message header exists.
     *
     * @param string $header_key Name of the header to check.
     *
     * @return boolean
     */
    public function hasHeader($header_key) { }
}

/**
 * Acknowledge the receipt of a message.
 *
 * This method allows the acknowledgement of a message that is retrieved
 * without the AMQP_AUTOACK flag through AMQPQueue::get() or
 * AMQPQueue::consume()
 *
 * @param string  $delivery_tag The message delivery tag of which to
 *                              acknowledge receipt.
 * @param integer $flags        The only valid flag that can be passed is
 *                              AMQP_MULTIPLE.
 *
 * @throws AMQPChannelException    If the channel is not open.
 * @throws AMQPConnectionException If the connection to the broker was lost.
 *
 * @return boolean
 */
public function ack($delivery_tag, $flags = AMQP_NOPARAM) { }

“#” 通配任何零个或多个word
“*” 通配任何单个word

Queue::produce(['indexname' => 'weberror', 'type' => 'adminWeb', 'error' => 303020], 'logstash', 'app_logs_routing');
Queue::produce(['id' => 101], 'async', 'ticket');

/**
 * 连接MQ并消费
 */
private function msgQueue()
{
    $conn = new \AMQPConnection($this->setting);
    if (!$conn->connect()) {
        echo "\n" ."Connect Failed" ."\n";
        exit();
    }
    echo "\n" ."Connect Success" ."\n";
    $channel = new \AMQPChannel($conn);
    
    $ex = new \AMQPExchange($channel);
    $ex->setName($this->bindSetting['exchange']);
    $ex->setType(AMQP_EX_TYPE_DIRECT);
    $ex->setFlags(AMQP_DURABLE);
    $ex->declareExchange();
    
    $q = new \AMQPQueue($channel);
    $q->setName($this->bindSetting['queue']);
    $q->setFlags(AMQP_DURABLE);
    $q->declareQueue();
    $q->bind(
        $this->bindSetting['exchange'],
        $this->bindSetting['routing']
    );
    
    if (!file_exists($this->root . '/workers/' . $this->consumer . '.php')) {
        echo "\n" ."workerService does not exist" ."\n";
        exit();
    }
    $workerName = 'console\\workers\\' . $this->consumer;
    $this->worker = new $workerName();
    $channel->qos(0,1);
    $q->consume(array($this->worker, 'run'));
    
    $conn->disconnect();
}

$this->setting = [
    'host' => 'test.rabbitmq.demo.com',
    'port' => '5672',
    'login' => 'mqadmin',
    'password' => 'demo_+-*123'
];

$this->bindSetting = [
    'queue' => 'queue_ticket',
    'exchange' => 'async',
    'routing' => 'ticket'
];

// console\workers\Ticket->run()
public function run($envelope, $queue)
{
    $msg = $envelope->getBody();
    $msg = json_decode($msg, true);
    $event = isset($msg['event']) ?  $msg['event'] : '';

    switch ($event) {
        case 'ticket_callback':
            TicketService::callbackRequest($msg['data']);
            break;
        default:
            break;
    }

    Yii::$app->demoDB->close();
    $queue->ack($envelope->getDeliveryTag());
}

/**
 * 连接MQ并消费
 */
private function msgQueue()
{
    $conn = new \AMQPConnection($this->setting);

    if (!$conn->connect()) {
        echo "\n" ."Connect Failed" ."\n";
        exit();
    }
    echo "\n" ."Connect Success" ."\n";

    $channel = new \AMQPChannel($conn);

    $ex = new \AMQPExchange($channel);
    $ex->setName($this->bindSetting['exchange']);
    $ex->setType(AMQP_EX_TYPE_DIRECT);
    $ex->setFlags(AMQP_DURABLE);
    $ex->declareExchange();

    $q = new \AMQPQueue($channel);
    $q->setName($this->bindSetting['queue']);
    $q->setFlags(AMQP_DURABLE);
    $q->setArguments([
        'x-dead-letter-exchange' => $this->bindSetting['x_exchange'],
        'x-dead-letter-routing-key' => $this->bindSetting['x_routing']
    ]);
    $q->declareQueue();

    $q->bind(
        $this->bindSetting['exchange'],
        $this->bindSetting['routing']
    );

    $conn->disconnect();
}

$this->setting = [
    'host' => 'test.rabbitmq.demo.com',
    'port' => '5672',
    'login' => 'mqadmin',
    'password' => 'demo_+-*123'
];

$this->bindSetting = [
    'queue' => 'queue_delay_ticket',
    'exchange' => 'delay',
    'routing' => 'delay_ticket',
    'x_exchange' => 'async',
    'x_routing' => 'ticket'
];