脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 网络编程 → PHP编程 → php技巧 → PHP rabbitmq任务处理

PHP使用enqueue/amqp-lib实现rabbitmq任务处理

 更新时间：2024年03月11日 11:58:28   作者：huaweichenai  
这篇文章主要为大家详细介绍了PHP如何使用enqueue/amqp-lib实现rabbitmq任务处理,文中的示例代码讲解详细,感兴趣的小伙伴可以学习一下

一：拓展安装

composer require enqueue/amqp-lib

文档地址：https://github.com/php-enqueue/enqueue-dev/blob/master/docs/transport/amqp_lib.md

二：方法介绍

1：连接rabbitmq

$factory = new AmqpConnectionFactory([
    'host' => '192.168.6.88',//host
    'port' => '5672',//端口
    'vhost' => '/',//虚拟主机
    'user' => 'admin',//账号
    'pass' => 'admin',//密码
]);
$context = $factory->createContext();

2：声明主题

//声明并创建主题
$exchangeName = 'exchange';
$fooTopic = $context->createTopic($exchangeName);
$fooTopic->setType(AmqpTopic::TYPE_FANOUT);
$context->declareTopic($fooTopic);
 
//删除主题
$context->deleteTopic($fooTopic);

3：声明队列

//声明并创建队列
$queueName = 'rabbitmq';
$fooQueue = $context->createQueue($queueName);
$fooQueue->addFlag(AmqpQueue::FLAG_DURABLE);
$context->declareQueue($fooQueue);
 
//删除队列
$context->deleteQueue($fooQueue);

4：将队列绑定到主题

$context->bind(new AmqpBind($fooTopic, $fooQueue));

5：发送消息

//向队列发送消息
$message = $context->createMessage('Hello world!');
$context->createProducer()->send($fooQueue, $message);
 
//向队列发送优先消息
$queueName = 'rabbitmq';
$fooQueue = $context->createQueue(queueName);
$fooQueue->addFlag(AmqpQueue::FLAG_DURABLE);
//设置队列的最大优先级
$fooQueue->setArguments(['x-max-priority' => 10]);
$context->declareQueue($fooQueue);
 
$message = $context->createMessage('Hello world!');
 
$context->createProducer()
    ->setPriority(5) //设置优先级，优先级越高，消息越快到达消费者
    ->send($fooQueue, $message);
 
//向队列发送延时消息
$message = $context->createMessage('Hello world!');
 
$context->createProducer()
    ->setDelayStrategy(new RabbitMqDlxDelayStrategy())
    ->setDeliveryDelay(5000) //消息延时5秒
    ->send($fooQueue, $message);

6：消费消息【接收消息】

//消费消息
$consumer = $context->createConsumer($fooQueue);
 
$message = $consumer->receive();
 
// process a message
//业务代码
 
$consumer->acknowledge($message);//ack应答，通知rabbitmq成功，删除对应任务
// $consumer->reject($message);ack应答，通知rabbitmq失败，不删除对应任务
 
 
//订阅消费者
$fooConsumer = $context->createConsumer($fooQueue);
 
$subscriptionConsumer = $context->createSubscriptionConsumer();
$subscriptionConsumer->subscribe($fooConsumer, function(Message $message, Consumer $consumer) {
    // process message
    //业务代码
    $consumer->acknowledge($message);//ack应答，通知rabbitmq成功，删除对应任务
    // $consumer->reject($message);ack应答，通知rabbitmq失败，不删除对应任务
 
    return true;
});
$subscriptionConsumer->consume();
 
//清除队列消息
$queueName = 'rabbitmq';
$queue = $context->createQueue($queueName);
$context->purgeQueue($queue);

三：简单实现 

1：发送消息

//连接rabbitmq
$factory = new AmqpConnectionFactory([
    'host' => '192.168.6.88',
    'port' => '5672',
    'vhost' => '/',
    'user' => 'admin',
    'pass' => 'admin',
    'persisted' => false,
]);
 
$context = $factory->createContext();
//声明主题
$exchangeName = 'exchange';
$fooTopic = $context->createTopic($exchangeName);
$fooTopic->setType(AmqpTopic::TYPE_FANOUT);
$context->declareTopic($fooTopic);
 
//声明队列
$queueName = 'rabbitmq';
$fooQueue = $context->createQueue($queueName);
$fooQueue->addFlag(AmqpQueue::FLAG_DURABLE);
$context->declareQueue($fooQueue);
 
//将队列绑定到主题
$context->bind(new AmqpBind($fooTopic, $fooQueue));
 
//发送消息到队列
$message = $context->createMessage('Hello world!');
 
$context->createProducer()->send($fooQueue, $message);

2：消费消息

$factory = new AmqpConnectionFactory([
    'host' => '192.168.6.88',
    'port' => '5672',
    'vhost' => '/',
    'user' => 'admin',
    'pass' => 'admin',
    'persisted' => false,
]);
$context = $factory->createContext();
 
 
$queueName = 'rabbitmq';
$fooQueue = $context->createQueue($queueName);
 
 
 
$fooConsumer = $context->createConsumer($fooQueue);
 
$subscriptionConsumer = $context->createSubscriptionConsumer();
$subscriptionConsumer->subscribe($fooConsumer, function(Message $message, Consumer $consumer) {
    // process message
    //业务代码
    $consumer->acknowledge($message);//ack应答，通知rabbitmq成功，删除对应任务
    // $consumer->reject($message);ack应答，通知rabbitmq失败，不删除对应任务
 
    return true;
});
$subscriptionConsumer->consume();

到此这篇关于PHP使用enqueue/amqp-lib实现rabbitmq任务处理的文章就介绍到这了,更多相关PHP rabbitmq任务处理内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

相关文章

  • PHP使用TCPDF处理pdf

    PHP使用TCPDF处理pdf

    TCPDF是一个开源的PHP库,用于创建和处理PDF文件,这篇文章主要为大家详细介绍了PHP如何使用TCPDF进行pdf的相关操作,需要的小伙伴可以参考下
    2023-12-12
  • 两种php去除二维数组的重复项方法

    两种php去除二维数组的重复项方法

    这篇文章主要介绍了两种php去除二维数组的重复项方法，大家可以进行比较看哪一种更适合自己，需要的朋友可以参考下
    2015-11-11
  • php抽象类和接口知识点整理总结

    php抽象类和接口知识点整理总结

    这篇文章主要介绍了php抽象类和接口知识点,整理总结了php抽象类与接口的概念、原理、操作技巧及相关使用注意事项,需要的朋友可以参考下
    2019-08-08
  • 使用PHP json_decode可能遇到的坑与解决方法

    使用PHP json_decode可能遇到的坑与解决方法

    在我们日常使用php开发的时候，经常会用到json_decode函数，最近在使用json_decode函数的发现了一个坑，想着总结分享出来，所以下面这篇文章主要给大家介绍了关于使用PHP json_decode可能遇到的坑与解决方法，需要的朋友可以参考借鉴。
    2017-08-08
  • 详解PHP中数组函数的巧用

    详解PHP中数组函数的巧用

    PHP 的数组是一种很强大的数据类型，与此同时 PHP 内置了一系列与数组相关的函数可以很轻易的实现日常开发的功能。所以本文便总结了一些在常见场景中利用 PHP 内置函数的实现方法，希望对你有所帮助
    2022-09-09
  • php读取文件内容的三种可行方法示例介绍

    php读取文件内容的三种可行方法示例介绍

    这篇文章主要介绍了php读取文件内容的三种方法,需要的朋友可以参考下
    2014-02-02
  • php中异常处理方法小结

    php中异常处理方法小结

    这篇文章主要介绍了php中异常处理方法,以实例形式总结归纳了常见的php异常处理技巧,具有一定参考借鉴价值,需要的朋友可以参考下
    2015-01-01
  • 一些PHP写的小东西

    一些PHP写的小东西

    一些PHP写的小东西...
    2006-12-12
  • PHP 模拟登陆MSN并获得用户信息

    PHP 模拟登陆MSN并获得用户信息

    一个可以获取msn上用户信息的代码
    2009-05-05
  • PHP开发环境配置(MySQL数据库安装图文教程)

    PHP开发环境配置(MySQL数据库安装图文教程)

    下载完软件后开始PHP开发环境的配置。注意是开发环境，不是服务器环境。
    2010-04-04

最新评论