不良网站浏览窗口,seo短视频新地址在哪里?,wordpress如何添加关键词和描述,无锡建设局评职称网站代码的参数说明在 第一小节的代码中#xff0c;如果需要可移步到第一节中查看
工作队列 工作队列#xff08;又称#xff1a;任务队列——Task Queues#xff09;是为了避免等待一些占用大量资源、时间的操作。当我们把任务#xff08;Task#xff09;当作消息发送到队列…代码的参数说明在 第一小节的代码中如果需要可移步到第一节中查看
工作队列 工作队列又称任务队列——Task Queues是为了避免等待一些占用大量资源、时间的操作。当我们把任务Task当作消息发送到队列中一个运行在后台的工作者worker进程就会取出任务然后处理。当你运行多个工作者workers任务就会在它们之间共享。 使用工作队列的一个好处就是它能够并行的处理队列。如果堆积了很多任务我们只需要添加更多的工作者workers就可以了扩展很简单。 当我们在 n个Terminal 窗口中运行消费者程序就可以有多个消费者处理生产者生产的消息了 当队列中的消息发送给消费者1的时候就不会再发送给消费者2了。
消息确认
当我们处理消息的时候 我们想知道消费者在处理的过程中是否已经处理完成没有出现消费者挂机的状态这里就需要消息确认了 不然队列中的消息是否处理完成不能明确 有可能会丢失重要的数据。 消息响应默认是开启的。之前的例子中我们可以使用no_ackTrue标识把它关闭。是时候设置的第四个参数basic_consume为false (true 意味着不响应ack) 当工作者worker完成了任务就发送一个响应。
公平调度
RabbitMQ只管分发进入队列的消息不会关心有多少消费者consumer没有作出响应。它盲目的把第n-th条消息发给第n-th个消费者。 不会等侍是否处理完成 我们可以使用basic.qos方法并设置prefetch_count1。这样是告诉RabbitMQ再同一时刻不要发送超过1条消息给一个工作者worker直到它已经处理了上一条消息并且作出了响应。这样RabbitMQ就会把消息分发给下一个空闲的工作者worker。
$channel-basic_qos(null, 1, null);生产者
使用了 第一小节中的生产者一次生成了20个消息
消费者
?php
declare (strict_types 1);namespace app\command;use Exception;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;//工作队列带消息确认的 开几个终端就会有几个消息者来消费。
//生产者可以使用 SimpleMQProduce 中的生产者
class WorkerWithAck extends Command
{protected function configure(){// 指令配置$this-setName(workerwithack)-setDescription(这是一个工作队列带应答的);}protected function execute(Input $input, Output $output){//获取连接$connection $this-getConnection();//获取通道$channel $connection-channel();$channel-queue_declare(hello,false,false,false,false,false);$callback function($msg){$msgbody $msg-body;$msgbydyArr json_decode($msgbody,true);echo $msgbydyArr[name].--.$msgbydyArr[age].--.$msgbydyArr[sex].PHP_EOL;$msg-delivery_info[channel]-basic_ack($msg-delivery_info[delivery_tag]); //这里让就是消息的应答了};//如果我们要让消费者一条一条的处理消费也就是说 只有consumer已经处理并确认了上一条message时queue才分派新的message给它//我们可以加上下面的这个代码 注意是可以加也可以不加。因为我们有了应答机制消息是不会丢失的//$channel-basic_qos(null,1,null); //这句可加可不加$channel-basic_consume(hello,,false,false,false,false,$callback);while(count($channel-callbacks)){$channel-wait();}}protected function getConnection(){try{return new AMQPStreamConnection(192.168.3.228,5672,admin,123456);}catch(Exception $e){throw new \Exception(创建队列连接失败);}}}
测试结果两个 工作队列分别处理了同一个生产者的数据并且没有重复
