在使用rabbimq之前先理解rabbitmq的基本通信方式：
发布过程：
connection(mq链接) -> channel(创建channel) -> exchange（创建exchange路由） -> queue（创建队列） -> bind（绑定exchangeName 和 routekey） -> publish(exchange发布消息) -> disconnect（关闭链接）
订阅过程：
connection(mq链接) -> channel(创建channel) -> exchange（创建exchange路由） -> queue（创建队列） -> bind（绑定exchangeName 和 routekey） -> consume(回调函数消费消息) -> disconnect（关闭链接）
发布（publish.php）
$exchangeName = "testex";
$queueName = "testqueue";
$routeKey = "testkey";
$message = "hello rabbitmq message!";
$connection = new AMQPConnection(['host'=>'127.0.0.1','port'=>'5672','vhost'=>'/','login'=>'ysp','password'=>'123123']);
$connection->connect() or die('can not connect broker server !');
$channel = new AMQPChannel($connection);
//$channel->setPerfetchCount(1);
$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->setFlags(AMQP_DURABLE);
$exchange->declareExchange();
$queue = new AMQPQueue($channel);
$queue->setName($queueName);
$queue->setFlags(AMQP_DURABLE);
$queue->declareQueue();
$queue->bind($exchangeName, $routeKey);
$exchange->publish($message, $routeKey);
$connection->disconnect();
订阅(subscribe.php)
$exchangeName = 'testex';
$queueName="queuetest";
$routeKey = 'testkey';
$connection = new AMQPConnection(['host'=>'127.0.0.1','port'=>'5672','vhost'=>'/','login'=>'ysp','password'=>'123123']);
$connection->connect() or die('can not connect broker server !');
$channel = new AMQPChannel($connection);
//$channel->setPerfetchCount(1);
$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->setFlags(AMQP_DURABLE);
$exchange->declareExchange();
$queue = new AMQPQueue($channel);
$queue->setName($queueName);
$queue->setFlags(AMQP_DURABLE);
$queue->declareQueue();
$queue->bind($exchangeName, $routeKey);
echo "message ........";
while(true){
$queue->consume('dealMsg');
}
$connection->disconnect();
function dealMsg($event, $q){
$msg = $event->getBody();
echo $msg."\r\n";
$q->ack($event->getDeliveryTag());
}
rabbitmq消息机制：
消息确认：为了保证数据不被丢失，RabbitMQ 支持消息确认机制，这种机制下不能采用 no-ack，而应该是在处理完数据后发送 ack。如果处理中途 Consumer 退出了，但是没有发送 ack，那么 RabbitMQ 就会把这个 Message 发送到下一个 Consumer，这样就保证了在 Consumer 异常退出的情况下数据也不会丢失。
代码中使用了 event->getDeliveryTag());，这就是消息确认机制的应用，这种情况下，即使中断任务执行，也不会影响 RabbitMQ 中消息的处理，RabbitMQ 会将其发送给下一个 Consumer 进行处理。
消息持久化：
为了保证在 RabbitMQ 退出或者 crash 了数据不丢失，需要将 Queue 和 Message 持久化。
exchange持久化
$exchange->setFlags(AMQP_DURABLE);
queue持久化
$queue->setFlags(AMQP_DURABLE);
公平分发：
如果Consumer数量很多或者希望每个Consumer同时只处理一个任务可以通过在Consumer中设置PrefetchCount来实现更加均匀的任务分发。
告诉 RabbitMQ，在同一时刻，不要发送超过 1 条消息给一个工作者（worker），直到它已经处理了上一条消息并且作出了响应。这样，RabbitMQ 就会把消息分发给下一个空闲的工作者（worker）。
$channel = new AMQPChannel($connection);
$channel->setPerfetchCount(1);
exchange将消息加入队列有四种方式：direct、topic 和 fanout。anout 就是广播模式，会将 Message 都放到它所知道的所有 Queue 中。
$exchange->setType(AMQP_EX_TYPE_DIRECT); 或 $exchange->setType(AMQP_EX_TYPE_TOPIC); 或 $exchange->setType(AMQP_EX_TYPE_FANOUT);
rabbitmq封装github地址: https://github.com/yspwf/rabbimq
参考： https://blog.csdn.net/super_rd/article/category/6869439
网友评论