zoukankan      html  css  js  c++  java
  • php运用PhpAmqpLib操作队列

    //安装类库
    composer require php-amqplib/php-amqplib
    //consumer.php消费者代码
    require __DIR__ . "/../vendor/autoload.php";
    
    use PhpAmqpLibConnectionAMQPStreamConnection;
    
    $conf         = [
        'host'  => '127.0.0.1',
        'port'  => 5672,
        'user'  => '',
        'pwd'   => '',
        'vhost' => '/',
    ];
    $exchangeName = '_ex'; //交换机名
    $queueName    = '_que'; //队列名称
    $routingKey   = '_send'; //路由关键字(也可以省略)
    
    $conn    = new AMQPStreamConnection( //建立生产者与mq之间的连接
        $conf['host'], $conf['port'], $conf['user'], $conf['pwd'], $conf['vhost']
    );
    $channel = $conn->channel(); //在已连接基础上建立生产者与mq之间的通道
    $channel->exchange_declare($exchangeName, 'direct', false, true, false); //声明初始化交换机
    $channel->queue_declare($queueName, false, true, false, false); //声明初始化一条队列
    $channel->queue_bind($queueName, $exchangeName, $routingKey); //将队列与某个交换机进行绑定,并使用路由关键字
    //1:queue 要取得消息的队列名
    //2:consumer_tag 消费者标签
    //3:no_local false这个功能属于AMQP的标准,但是rabbitMQ并没有做实现.参考
    //4:no_ack  false收到消息后,是否不需要回复确认即被认为被消费
    //5:exclusive false排他消费者,即这个队列只能由一个消费者消费.适用于任务不允许进行并发处理的情况下.比如系统对接
    //6:nowait  false不返回执行结果,但是如果排他开启的话,则必须需要等待结果的,如果两个一起开就会报错
    //7:callback  null回调函数
    //8:ticket  null
    //9:arguments null
    $channel->basic_consume($queueName, '', false, true, false, false, function($msg){
        var_dump($msg->delivery_info);
        $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);//手动应答
    });
    //监听消息
    while(count($channel->callbacks)){
        $channel->wait();
    }
    //publisher.php生产者代码
    require __DIR__ . "/../vendor/autoload.php";
    
    use PhpAmqpLibConnectionAMQPStreamConnection;
    use PhpAmqpLibMessageAMQPMessage;
    
    $conf         = [
        'host'  => '127.0.0.1',
        'port'  => 5672,
        'user'  => '',
        'pwd'   => '',
        'vhost' => '/',
    ];
    $exchangeName = '_ex'; //交换机名
    $routingKey   = '_send'; //路由关键字(也可以省略)
    
    $conn    = new AMQPStreamConnection( //建立生产者与mq之间的连接
        $conf['host'], $conf['port'], $conf['user'], $conf['pwd'], $conf['vhost']
    );
    $channel = $conn->channel(); //在已连接基础上建立生产者与mq之间的通道
    $channel->exchange_declare($exchangeName, 'direct', false, true, false); //声明初始化交换机
    
    $msgBody = json_encode(["msg" => "hello"]);
    $msg     = new AMQPMessage($msgBody, ['content_type' => 'text/plain', 'delivery_mode' => 2]); //生成消息
    $r       = $channel->basic_publish($msg, $exchangeName, $routingKey); //推送消息到某个交换机
    $channel->close();
    $conn->close();
  • 相关阅读:
    Delphi中多线程同步过程Synchronize的一些说明
    property中的read,write是什么意思?
    如何用delphi读写csv文件
    将DBGrid中的数据导入Excel表格中
    如何获取combobox显示的值
    WaitForSingleObject 的返回值
    关于GetOverlappedResult函数的一些知识
    串口编程:COMSTAT 结构
    windows 官方镜像下载地址
    ffmpeg 命令行改变视频分辨率
  • 原文地址:https://www.cnblogs.com/godehi/p/14493720.html
Copyright © 2011-2022 走看看