zoukankan      html  css  js  c++  java
  • PHP使用RabbitMQ实例

    首先看下没有生产者和消费者的默认RabbitMQ管理界面截图:
    Connections:

    还没有任何连接(Connections)
    Channels:

    还没有任何通道(Channels)
    Exchanges:

    交换机只有系统默认的
    Queues:

    还没有任何队列
    先上消费者代码consumer.php

    <?php
    /**
     * Created by PhpStorm.
     * User: jmsite.cn
     * Date: 2019/1/15
     * Time: 13:16
     */
    //声明连接参数
    $config = array(
        'host' => '192.168.75.132',
        'vhost' => '/',
        'port' => 5672,
        'login' => 'test',
        'password' => 'test'
    );
    //连接broker
    $cnn = new AMQPConnection($config);
    if (!$cnn->connect()) {
        echo "Cannot connect to the broker";
        exit();
    }
    //在连接内创建一个通道
    $ch = new AMQPChannel($cnn);
    //创建一个交换机
    $ex = new AMQPExchange($ch);
    //声明路由键
    $routingKey = 'key_1';
    //声明交换机名称
    $exchangeName = 'exchange_1';
    //设置交换机名称
    $ex->setName($exchangeName);
    //设置交换机类型
    //AMQP_EX_TYPE_DIRECT:直连交换机
    //AMQP_EX_TYPE_FANOUT:扇形交换机
    //AMQP_EX_TYPE_HEADERS:头交换机
    //AMQP_EX_TYPE_TOPIC:主题交换机
    $ex->setType(AMQP_EX_TYPE_DIRECT);
    //设置交换机持久
    $ex->setFlags(AMQP_DURABLE);
    //声明交换机
    $ex->declareExchange();
    //创建一个消息队列
    $q = new AMQPQueue($ch);
    //设置队列名称
    $q->setName('queue_1');
    //设置队列持久
    $q->setFlags(AMQP_DURABLE);
    //声明消息队列
    $q->declareQueue();
    //交换机和队列通过$routingKey进行绑定
    $q->bind($ex->getName(), $routingKey);
    //接收消息并进行处理的回调方法
    function receive($envelope, $queue) {
    	//休眠两秒,
        sleep(2);
    	//echo消息内容
        echo $envelope->getBody()."
    ";
    	//显式确认,队列收到消费者显式确认后,会删除该消息
        $queue->ack($envelope->getDeliveryTag());
    }
    //设置消息队列消费者回调方法,并进行阻塞
    $q->consume("receive");
    //$q->consume("receive", AMQP_AUTOACK);//隐式确认,不推荐
    

    以上是消费者代码,打开两个命令行/终端
    输入php consumer.php,消费者开始阻塞获取消息,如下图

    此时再看RabbitMQ管理界面:
    Connections

    出现两个连接,这两个就是消费者,因为他们在阻塞着等待消息
    Channels

    消费者在各自的连接里都打开了一个通道
    Exchanges

    其中一个消费者创建了一个持久的直连交换机
    Queues

    消息队列已经创建,但消息数是0,因为此时还没有生产者
    生产者代码publisher.php

    <?php
    /**
     * Created by PhpStorm.
     * User: jmsite.cn
     * Date: 2019/1/15
     * Time: 13:15
     */
    
    $config = array(
        'host' => '192.168.75.132',
        'vhost' => '/',
        'port' => 5672,
        'login' => 'test',
        'password' => 'test'
    );
    $cnn = new AMQPConnection($config);
    if (!$cnn->connect()) {
        echo "Cannot connect to the broker";
        exit();
    }
    $ch = new AMQPChannel($cnn);
    $ex = new AMQPExchange($ch);
    //消息的路由键,一定要和消费者端一致
    $routingKey = 'key_1';
    //交换机名称,一定要和消费者端一致,
    $exchangeName = 'exchange_1';
    $ex->setName($exchangeName);
    $ex->setType(AMQP_EX_TYPE_DIRECT);
    $ex->setFlags(AMQP_DURABLE);
    $ex->declareExchange();
    //创建10个消息
    for ($i=1;$i<=10;$i++){
    	//消息内容
        $msg = array(
            'data'  => 'message_'.$i,
            'hello' => 'world',
        );
    	//发送消息到交换机,并返回发送结果
    	//delivery_mode:2声明消息持久,持久的队列+持久的消息在RabbitMQ重启后才不会丢失
        echo "Send Message:".$ex->publish(json_encode($msg), $routingKey, AMQP_NOPARAM, array('delivery_mode' => 2))."
    ";
    	//代码执行完毕后进程会自动退出
    }
    

    以上是生产者代码
    在执行之前,先关掉前面的两个消费者,打开一个命令行/终端,输入php publisher.php,由于生产者不需要阻塞,执行完进程便退出,所以现在RabbitMQ管理界面中既没有Connections也没有Channels,但是Queues已经被Exchanges投递过去了10条消息,如下图:

    因为我们执行生产者之前已经关掉了全部消费者,所以此时消息在队列中等待获取;
    因为在发送消息时设置了delivery_mode:2来声明消息持久化,此时如果重启RabbitMQ,消息还会恢复;此时重新执行消费者,假设还是两个,打开两个命令行/终端,输入php consumer.php,我们可以看到消息被消费,如下图:

    提醒:生产者在生产消息时,如果不存在指定队列,并且没有创建队列,或者队列存在但消息路由键和交换机与队列绑定的键(路由规则)不一致(直连交换机必须一致),则消息会被交换机丢弃。

  • 相关阅读:
    CSS3实现平行四边形
    [Python]图的遍历-DFS
    [Python]图的遍历-BFS
    [Python]图的遍历-拓扑排序
    [Python]哈夫曼编码
    [Python]贪心算法-Prim-和-Kruskal实现-最小生成树
    [Python]贪心算法-Dijkstra-实现
    [python3]稳定匹配算法实现和优化
    从csv文件构建Tensorflow的数据集
    Tensorflow 基础API
  • 原文地址:https://www.cnblogs.com/liliuguang/p/14430084.html
Copyright © 2011-2022 走看看