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

熊.Mir

目录

• 一：拓展安装
  
• 二：方法介绍
  
  
  
  • 1：连接rabbitmq
    
  • 2：声明主题
    
  • 3：声明队列
    
  • 4：将队列绑定到主题
    
  • 5：发送消息
    
  • 6：消费消息【接收消息】
    
  
  
• 三：简单实现
  
  
  
  • 1：发送消息
    
  • 2：消费消息
    
  
  

一：拓展安装

1. composer require enqueue/amqp-lib

复制代码

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

二：方法介绍


1：连接rabbitmq

1. $factory = new AmqpConnectionFactory([
   
2.     'host' => '192.168.6.88',//host
   
3.     'port' => '5672',//端口
   
4.     'vhost' => '/',//虚拟主机
   
5.     'user' => 'admin',//账号
   
6.     'pass' => 'admin',//密码
   
7. ]);
   
8. $context = $factory->createContext();

复制代码

2：声明主题

1. //声明并创建主题
   
2. $exchangeName = 'exchange';
   
3. $fooTopic = $context->createTopic($exchangeName);
   
4. $fooTopic->setType(AmqpTopic::TYPE_FANOUT);
   
5. $context->declareTopic($fooTopic);
   
6. 
   
7. //删除主题
   
8. $context->deleteTopic($fooTopic);

复制代码

3：声明队列

1. //声明并创建队列
   
2. $queueName = 'rabbitmq';
   
3. $fooQueue = $context->createQueue($queueName);
   
4. $fooQueue->addFlag(AmqpQueue::FLAG_DURABLE);
   
5. $context->declareQueue($fooQueue);
   
6. 
   
7. //删除队列
   
8. $context->deleteQueue($fooQueue);

复制代码

4：将队列绑定到主题

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

复制代码

5：发送消息

1. //向队列发送消息
   
2. $message = $context->createMessage('Hello world!');
   
3. $context->createProducer()->send($fooQueue, $message);
   
4. 
   
5. //向队列发送优先消息
   
6. $queueName = 'rabbitmq';
   
7. $fooQueue = $context->createQueue(queueName);
   
8. $fooQueue->addFlag(AmqpQueue::FLAG_DURABLE);
   
9. //设置队列的最大优先级
   
10. $fooQueue->setArguments(['x-max-priority' => 10]);
    
11. $context->declareQueue($fooQueue);
    
12. 
    
13. $message = $context->createMessage('Hello world!');
    
14. 
    
15. $context->createProducer()
    
16.     ->setPriority(5) //设置优先级，优先级越高，消息越快到达消费者
    
17.     ->send($fooQueue, $message);
    
18. 
    
19. //向队列发送延时消息
    
20. $message = $context->createMessage('Hello world!');
    
21. 
    
22. $context->createProducer()
    
23.     ->setDelayStrategy(new RabbitMqDlxDelayStrategy())
    
24.     ->setDeliveryDelay(5000) //消息延时5秒
    
25.     ->send($fooQueue, $message);

复制代码

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

1. //消费消息
   
2. $consumer = $context->createConsumer($fooQueue);
   
3. 
   
4. $message = $consumer->receive();
   
5. 
   
6. // process a message
   
7. //业务代码
   
8. 
   
9. $consumer->acknowledge($message);//ack应答，通知rabbitmq成功，删除对应任务
   
10. // $consumer->reject($message);ack应答，通知rabbitmq失败，不删除对应任务
    
11. 
    
12. 
    
13. //订阅消费者
    
14. $fooConsumer = $context->createConsumer($fooQueue);
    
15. 
    
16. $subscriptionConsumer = $context->createSubscriptionConsumer();
    
17. $subscriptionConsumer->subscribe($fooConsumer, function(Message $message, Consumer $consumer) {
    
18.     // process message
    
19.     //业务代码
    
20.     $consumer->acknowledge($message);//ack应答，通知rabbitmq成功，删除对应任务
    
21.     // $consumer->reject($message);ack应答，通知rabbitmq失败，不删除对应任务
    
22. 
    
23.     return true;
    
24. });
    
25. $subscriptionConsumer->consume();
    
26. 
    
27. //清除队列消息
    
28. $queueName = 'rabbitmq';
    
29. $queue = $context->createQueue($queueName);
    
30. $context->purgeQueue($queue);

复制代码

三：简单实现


1：发送消息

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：消费消息

1. $factory = new AmqpConnectionFactory([
   
2.     'host' => '192.168.6.88',
   
3.     'port' => '5672',
   
4.     'vhost' => '/',
   
5.     'user' => 'admin',
   
6.     'pass' => 'admin',
   
7.     'persisted' => false,
   
8. ]);
   
9. $context = $factory->createContext();
   
10. 
    
11. 
    
12. $queueName = 'rabbitmq';
    
13. $fooQueue = $context->createQueue($queueName);
    
14. 
    
15. 
    
16. 
    
17. $fooConsumer = $context->createConsumer($fooQueue);
    
18. 
    
19. $subscriptionConsumer = $context->createSubscriptionConsumer();
    
20. $subscriptionConsumer->subscribe($fooConsumer, function(Message $message, Consumer $consumer) {
    
21.     // process message
    
22.     //业务代码
    
23.     $consumer->acknowledge($message);//ack应答，通知rabbitmq成功，删除对应任务
    
24.     // $consumer->reject($message);ack应答，通知rabbitmq失败，不删除对应任务
    
25. 
    
26.     return true;
    
27. });
    
28. $subscriptionConsumer->consume();

复制代码

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

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

chenlie1168434

我完全同意你的观点

ning0888

谢谢楼主分享

小和尚懵了

友善的讨论氛围是非常重要的。

五更琉璃赛高

能给个链接吗？我想深入了解一下。

奔跑的不锈钢

666666666666666666

飞翔的小鸟123

好用好用

无双瑶夏

这个话题很有趣，我想多了解一些

七彩祥云

让我们一起努力

一个人的故事

保持尊重和礼貌对待其他成员是必要的。