RabiitMQ命令使用

刺骨的言语ヽ痛彻心扉 2023-05-31 07:47 6阅读 0赞

$cnn = news AMQPStreamConnection();
    //建立mq之间的连接 (host:ip,port:端口,user:账户,pwd:密码) 
    
    $channel = $cnn->channel(); //创建一个信道 AMQP的命令都是通过信道发送出去的
    
    $channel->exchange_declare(); 
    //创建交换机 (exchange:交换机名,type:交换机类型,passive:只判断不创建,durable:重启重建,auto_delete:无用自动销毁)
    
    $tale = new AMQPTable(); 
    $tale->set('x-dead-letter-exchange', '交换机名称');
    $tale->set('x-dead-letter-routing-key','交换机名称');
    $tale->set('x-message-ttl',15000); //队列存活时间
    //表示过期后由哪个交换机处理
    
    $channel->queue_declare('cache_queue',false,true,false,false,false,$tale); //声明队列
    //1.queue 是队列的名字.而存在默认值的意思是.你可以创建一个不重复名称的一个临时队列.
    //2.passive 只查询不创建.如果为true,如果存在这个队列,则会返回队列的信息.如果不存在这个队列..则会抛异常
    //3.durable 重启后是否会重建这个队列
    //4.exclusive 排他队列,如果你希望创建一个队列,并且只有你当前这个程序(或进程)进行消费处理.不希望别的客户端读取到这个队列.用这个方法甚好.而同时如果当进程断开连接.这个队列也会被销毁.不管是否设置了持久化或者自动删除.
    //5.auto_delete  自动删除
    //6.nowait 不等待处理结果
    //7.arguments 额外的参数
    
    $channel->queue_bind('cache_queue', 'cache_exchange','cache_exchange'); //绑定交换机
    //1.queue 队列名
    //2.exchange 交换机名
    //3.routing_key 路由名
    
    $msg = new AMQPMessage("消息内容",['配置项'=>'配置值']); //创建消息
    
    $channel->basic_publish($msg,'cache_exchange','cache_exchange'); //发送消息
    //1.msg 消息对象,创建的消息
    //2.exchange 交换机
    //3.routing_key 路由
    
    //回调函数
    $callback = function ($msg){ 
        //在这里可以添加消费消息的具体逻辑 
        $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
        //根据消息的delivery_info['channel']找到通道,并调用通道的basic_ack方法发送消息的确认内容.
    } 
    
    $channel->basic_qos(); //预取方法
    
    $channel->basic_consume(); //只有consumer已经处理并确认了上一条message时queue才分派新的message给它
    //1.队列名称
    //2.consumer_tag 消费者标签
    //3.no_local false 这个功能属于AMPQ的标准，但是rabbitMq并没有做实现
    //4.no_ack false 收到消息后，是否不需要回复确认即被认为是被消费
    //5.exclusive false 排他消费者，即这个队列只能有一个消费者消费，适用于人物不允许进行并打处理的情况下，比如系统对接
    //6.callback 回调函数
    
    //判断是否存在回调函数
    while(count($channel->callbacks)) {
        // 此处为执行回调函数
        $channel->wait();
    }
    
    $channel->close(); //关闭通道
    $cnn->close(); //关闭连接