zoukankan      html  css  js  c++  java
  • rabitmq + php

    消费者

    <?php
    //配置信息
    $conn_args = array(
    'host' => '127.0.0.1',
    '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! ");
    }
    $channel = new AMQPChannel($conn);

    //创建交换机
    $ex = new AMQPExchange($channel);
    $ex->setName($e_name);
    $ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型(绝对匹配) topic类型(模糊匹配【"#" 匹配0个或多个 、""匹配有且一个】 ) fanout类型(所有范围最广)
    $ex->setFlags(AMQP_DURABLE); //持久化
    echo "Exchange Status:".$ex->declare()." ";

    //创建队列
    $q = new AMQPQueue($channel);
    $q->setName($q_name);
    $q->setFlags(AMQP_DURABLE); //持久化
    echo "Message Total:".$q->declare()." ";

    //绑定交换机与队列,并指定路由键
    echo 'Queue Bind: '.$q->bind($e_name, $k_route)." ";

    //阻塞模式接收消息
    echo "Message: ";
    while(True){
    $q->consume('processMessage');
    //$q->consume('processMessage', AMQP_AUTOACK); //自动ACK应答
    }
    $conn->disconnect();

    /**
    * 消费回调函数
    * 处理消息
    */
    function processMessage($envelope, $queue) {
    $msg = $envelope->getBody();
    echo $msg." "; //处理消息
    $queue->ack($envelope->getDeliveryTag()); //手动发送ACK应答
    }
    ?>

    生产者

    <?php
    //配置信息
    $conn_args = array(
    'host' => '127.0.0.1',
    '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! ");
    }
    $channel = new AMQPChannel($conn);

    //创建交换机对象
    $ex = new AMQPExchange($channel);
    $ex->setName($e_name);
    date_default_timezone_set("Asia/Shanghai");
    //发送消息
    //$channel->startTransaction(); //开始事务
    for($i=0; $i<5; ++$i){
    sleep(1);//休眠1秒
    //消息内容
    $message = "TEST MESSAGE!".date("h:i:sa");
    echo "Send Message:".$ex->publish($message, $k_route)." ";
    }
    //$channel->commitTransaction(); //提交事务

    $conn->disconnect();
    ?>

  • 相关阅读:
    腾讯为什么会出Q立方浏览器?
    String,StringBuffer与StringBuilder的区别??
    Linux Socket编程(不限Linux)
    将div显示在屏幕正中央
    计算鼠标坐标是否在指定范围内
    正则
    ajax异步通信
    CSS Float 换行
    jQuery强大的jQuery选择器
    给display字段增加筛选功能
  • 原文地址:https://www.cnblogs.com/setevn/p/12156599.html
Copyright © 2011-2022 走看看