# 队列 * * * * * **简介** 消息队列中间件是分布式系统中重要的组件，主要解决应用耦合，异步消息，流量削锋等问题。 实现高性能，高可用，可伸缩和最终一致性架构。 是大型分布式系统不可缺少的中间件。 场景：异步处理、应用解耦、流量削锋、日志处理 * * * 以上就是关于为什么要使用队列的大致说明 下面介绍在OneBase使用TP官方开发的队列think-queue。 首先需要仔细理解下面这张图  执行流程 1. 命令行`Command`开始监听队列`queue:work` 2. 执行进程`Worker`获取新消息`Queue::pop()` 3. 消息队列`Queue`返回一个可用的`Job`实例`$job` 3.1 生产者推送`Queue::push()`新消息到消息队列`Queue` 3.2 消息队列`Queue`返回是否推送成功给生产者 4. 执行进程`Worker`调用`$job`的`fire()`方法 5. 消息`Job`解析`$job`的`payload`，实例化一个消费者，并调用消费者实例的`fire($job, $data)`方法。 6. 消费者读取消息内容`$data`，处理业务逻辑，删除或重发该消息 `$job->delete()` 或 `$job->release()`。 7. 消息`Job`从Database或Redis中删除消息或重发消息 8. 消息`Job`返回消息处理结果给执行进程`Worker` 9. 执行进程`Worker`在终端输出响应或结束运行 **一、队列配置** 队列配置在app目录下config.php配置文件中，可以配置驱动，建议Redis。 此处为演示方便使用的数据库驱动。 ``` // +---------------------------------------------------------------------- // | 队列配置 // +---------------------------------------------------------------------- 'queue' => [ // sync驱动表示取消消息队列还原为同步执行 // 'connector' => 'Sync', // Redis驱动 // 'connector' => 'redis', // "expire"=>60,//任务过期时间默认为秒，禁用为null // "default"=>"default",//默认队列名称 // "host"=>Env::get("redis.host", "127.0.0.1"),//Redis主机IP地址 // "port"=>Env::get("redis.port", 6379),//Redis端口 // "password"=>Env::get("redis.password", "123456"),//Redis密码 // "select"=>5,//Redis数据库索引 // "timeout"=>0,//Redis连接超时时间 // "persistent"=>false,//是否长连接 // Database驱动 "type"=>"Database",//数据库驱动 "expire"=>60,//任务过期时间，单位为秒，禁用为null "default"=>"default",//默认队列名称 "table"=>"jobs",//存储消息的表明，不带前缀 "dsn"=>[], ] ``` **二、消息创建与推送** ``` // 将任务加入队列 public function push() { $job_data = []; $job_data["member_id"] = time(); $job_data["to_member_id"] = time(); $job_data["params"] = ['xx' => 'cc', 'vv' => 'bb']; $is_pushed = Queue::push("app\queue\controller\Test", $job_data, 'test_job_queue'); if($is_pushed !== false ) { echo date("Y-m-d H:i:s")." a new job is pushed to the message queue"; } else { echo date("Y-m-d H:i:s")." a new job pushed fail"; } } ``` $job_data 是队列中依赖的业务数据 app\queue\controller\Test 是处理队列数据的类路径 test_job_queue 是队列的名称 **三、启动消息队列**  此时执行多次消息推送代码。   可以看到 消息已经入库，但是没有处理。 **四、消费与删除** 打开 queue 模块下的控制器与逻辑目录，参考Test与TestLogic 来编写相关数据处理业务逻辑代码。  **五、启动监听处理** 命令行执行：php think queue:listen --queue test_job_queue  就能看到 数据表中的任务迅速消失，此时再调用消息创建推送到表中，也会迅速处理。 到此队列基本上算是使用起来了，任务种类不同比如批量下载任务和批量消息发送任务 可以写多个不同的消费者类进行处理。^_^