安装rabbitmq和php扩展

落日映苍穹つ 2023-10-18 16:15 223阅读 0赞

### 一、安装rabbitmq服务端 ###

1、安装erlang

试了几次erlang官方的包安装后，在安装rabbitmq时都报错要依赖R14B02，干脆使用rabbitmq官方提供的包：[https://github.com/rabbitmq/erlang-rpm][https_github.com_rabbitmq_erlang-rpm] [点击下载][Link 1]，终于不报错了。

**rabbitmq官方提供安装包的步骤：**  
1、下载后进入安装包，直接输入命令：make  
2、安装过程需要3分钟左右，编译过程中需要用到rpm-build和autoconf，没有安装的需要安装。安装完成后，会在目录RPMS/x86\_64/下找到erlang的安装包，rpm -ivh \*\*\* 安装即可。

2、安装rabbitmq

我直接使用rabbitmq 的rpm安装包。如果这时还报错socat找不到，下载socat 的安装包安装即可。  
注意：如果使用yum安装socat，需安装epel。

3、测试rabbitmq是否安装成功

service rabbitmq-server start #开启rabbitmq
    service rabbitmq-server status #查看rabbitmq状态

### 二、安装rabbitmq的php扩展 ###

1、安装扩展依赖库\#\#\#\#\#

注意:扩展是C写的,由于C与RabbitMQ通信一般需要依赖rabbitmq-c库(也就是librabbitmq),所以编译扩展前需要先装依赖库。不同版本的扩展，对php版本和librabbitmq兼容性不一样。  
[rabbitmq-c依赖库下载：][rabbitmq-c] [https://github.com/alanxz/rabbitmq-c/archive/v0.8.0.tar.gz][rabbitmq-c]

mkdir build && cd build # 这一步是在rabbitmq-c的根目录下创建一个build子目录
    cmake -DCMAKE_INSTALL_PREFIX=/usr/local/librabbitmq ..  # 这一步是让cmake根据../CMakeList.txt，即rabbitmq-c的根目录下的CMakeList.txt创建Makefile文件，Makefile文件会被创建到build目录中
    cmake --build . # 这一步是真正的build rabbitmq-c库的，注意，不要漏掉结尾的点 '.'
    make
    make install

2、安装amqp扩展

[amqp扩展下载：][amqp] [http://pecl.php.net/get/amqp-1.9.0.tgz][amqp]

tar zvxf amqp-1.9.0.tgz #解压
    cd amqp-1.9.9 #打开目录
    
    ./configure --with-php-config=/usr/local/php56/bin/php-config --with-amqp --with-librabbitmq-dir=/usr/local/librabbitmq
    
    make
    make install

修改php.ini

extension=amqp.so #extension_dir自己定义

重启php-fpm

service php-fpm restart #重启

如无意外，则可以在phpinfo页面看到如下所示：

![Image 1][]

图一

### 三、监控及代码测试 ###

**启动rabbitmq监控页面**

rabbitmq-plugins enable rabbitmq_management #启动监控页面插件

**生产者代码 (创建一个rabbit\_publisher.php的文件)**  
创建连接-->创建channel-->创建交换机对象-->发送消息

<?php 
    //配置信息
    $conn_args = array(
        'host' => '192.168.1.93',
        'port' => '5672',
        'login' => 'guest',
        'password' => 'guest',
        'vhost'=>'/'
    );
    $e_name = 'e_linvo'; //交换机名
    //$q_name = 'q_linvo'; //无需队列名
    $k_route = 'key_1'; //路由key
    
    //创建连接和channel
    $conn = new AMQPConnection($conn_args);
    if (!$conn->connect()) {
        die("Cannot connect to the broker!\n");
    }
    $channel = new AMQPChannel($conn);
    
    //消息内容
    $message = "TEST MESSAGE! 测试消息！";
    
    //创建交换机对象
    $ex = new AMQPExchange($channel);
    $ex->setName($e_name);
    
    //发送消息
    //$channel->startTransaction(); //开始事务
    for($i=0; $i<5; ++$i){
        echo "Send Message:".$ex->publish($message, $k_route)."\n";
    }
    //$channel->commitTransaction(); //提交事务
    
    $conn->disconnect();

**消费者代码(创建一个rabbit\_consumer.php的文件)**  
创建连接-->创建channel-->创建交换机-->创建队列-->绑定交换机/队列/路由键-->接收消息

<?php 
    //配置信息
    $conn_args = array(
        'host' => '192.168.1.93',
        'port' => '5672',
        'login' => 'guest',
        'password' => 'guest',
        'vhost'=>'/'
    );
    $e_name = 'e_linvo'; //交换机名
    $q_name = 'q_linvo'; //队列名
    $k_route = 'key_1'; //路由key
    
    //创建连接和channel
    $conn = new AMQPConnection($conn_args);
    if (!$conn->connect()) {
        die("Cannot connect to the broker!\n");
    }
    $channel = new AMQPChannel($conn);
    
    //创建交换机
    $ex = new AMQPExchange($channel);
    $ex->setName($e_name);
    $ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型
    $ex->setFlags(AMQP_DURABLE); //持久化
    echo "Exchange Status:".$ex->declare()."\n";
    
    //创建队列
    $q = new AMQPQueue($channel);
    $q->setName($q_name);
    $q->setFlags(AMQP_DURABLE); //持久化
    echo "Message Total:".$q->declare()."\n";
    
    //绑定交换机与队列，并指定路由键
    echo 'Queue Bind: '.$q->bind($e_name, $k_route)."\n";
    
    //阻塞模式接收消息
    echo "Message:\n";
    while(True){
        $q->consume('processMessage');
        //$q->consume('processMessage', AMQP_AUTOACK); //自动ACK应答
    }
    $conn->disconnect();
    
    /**
    * 消费回调函数
    * 处理消息
    */
    function processMessage($envelope, $queue) {
        $msg = $envelope->getBody();
        echo $msg."\n"; //处理消息
        $queue->ack($envelope->getDeliveryTag()); //手动发送ACK应答
    }
    
    /**
     * 消费回调函数
     * 处理消息
     */
    function processMessage($envelope, $queue) {
        var_dump($envelope->getRoutingKey);
        $msg = $envelope->getBody();
        echo $msg."\n"; //处理消息
    }

需要注意的地方是：  
queue对象有两个方法可用于取消息：consume和get。前者是阻塞的，无消息时会被挂起，适合循环中使用；  
后者则是非阻塞的，取消息时有则取，无则返回false。  
测试截图  
运行消费者：

![Image 1][]

运行生产者，发消息：

![Image 1][]

消费者接收到消息：

![Image 1][]

执行两个文件，再打开RabbitMQ的管理中心 [http://127.0.0.1:15672/][http_127.0.0.1_15672]

![Image 1][]

作者：无极生两仪  
链接：https://www.jianshu.com/p/65490900a937  
来源：简书  
简书著作权归作者所有，任何形式的转载都请联系作者获得授权并注明出处。

[https_github.com_rabbitmq_erlang-rpm]: https://link.jianshu.com?t=https://github.com/rabbitmq/erlang-rpm
[Link 1]: https://link.jianshu.com?t=https://codeload.github.com/rabbitmq/erlang-rpm/zip/master
[rabbitmq-c]: https://link.jianshu.com?t=https://github.com/alanxz/rabbitmq-c/archive/v0.8.0.tar.gz
[amqp]: https://link.jianshu.com?t=http://pecl.php.net/get/amqp-1.9.0.tgz
[Image 1]: 
[http_127.0.0.1_15672]: https://link.jianshu.com?t=http://127.0.0.1:15672/