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

牢记使命不忘初心

一：拓展安装

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任务处理内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

来源:https://www.jb51.net/program/317443dm3.htm
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！