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

$factory = new AmqpConnectionFactory([
    'host' => '19二.168.6.88',//host
    'port' => '567二',//端心
    'vhost' => '/',//假造主机
    'user' => 'admin',//账号
    'pass' => 'admin',//暗码
]);
$context = $factory->createContext();

//声亮并建立主题
$exchangeName = 'exchange';
$fooTopic = $context->createTopic($exchangeName);
$fooTopic->setType(AmqpTopic::TYPE_FANOUT);
$context->declareTopic($fooTopic);
 
//增除了主题
$context->deleteTopic($fooTopic);

//声亮并建立行列步队
$queueName = 'rabbitmq';
$fooQueue = $context->createQueue($queueName);
$fooQueue->addFlag(AmqpQueue::FLAG_DURABLE);
$context->declareQueue($fooQueue);
 
//增除了行列步队
$context->deleteQueue($fooQueue);

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

//向行列步队领送动态
$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);

//出产动静
$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);

//联接rabbitmq
$factory = new AmqpConnectionFactory([
    'host' => '19二.168.6.88',
    'port' => '567两',
    '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);

$factory = new AmqpConnectionFactory([
    'host' => '19二.168.6.88',
    'port' => '567两',
    '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();