[{"_1":2,"_3":-5,"_4":-5},"loaderData",{"_5":6,"_7":8},"actionData","errors","root",{"_11":-5,"_58":59,"_60":61,"_62":63},"routes/user.$name.replies",{"_9":10,"_11":12,"_13":14,"_15":16,"_17":18,"_19":20},"replies",[35],"user",{"_21":22,"_13":14,"_23":24,"_25":14,"_26":27,"_28":29,"_30":20,"_31":32,"_33":34},"loginname","ZoeQuan","page",1,"limit",50,"total",3,"id","19324","avatar_url","https://avatars.githubusercontent.com/u/15337365?v=3","githubUsername","create_at","2015-10-27T07:05:44.157Z","score",80,"topic_count","reply_count",13,"is_star",false,{"_21":36,"_37":38,"_39":40,"_41":42,"_43":44,"_45":46,"_47":48,"_49":34,"_31":50,"_51":52,"_26":53,"_54":55},"14587","author_id","4922","tab","share","content","最近有个朋友要实现队列任务方面的工作，我们就 mysql(innodb) 的事务和锁的特性聊了一些有趣的话题。\r\n\r\n其中，最终的解决方案来自大神 https://github.com/fengmk2 之前的一个队列实现。\r\n我做了一个小改进，使得之前表级锁的表现可以恢复到行级锁水平。\r\n\r\n任务的大致描述是这样的：\r\n\r\n有一个表，里面存了很多的用户id，大概100w条，表的结构简化如下：\r\n\r\n```sql\r\ncreate table user_block_status {\r\n user_id bigint // 用户的id\r\n status int // 用户的状态。1 ok 2 not ok\r\n updated_time timestamp // 更新时间戳\r\n}\r\n```\r\n\r\n这个表里面，每隔10秒就要去检查用户是否存在违规页面。如果存在的话，则需要把 status 置为 2，默认是 1。\r\n\r\n有 100 个 worker 会并发地从表里面读取 user_id，所以我们要设计一个策略，使得这 100 个 worker 在并发时，\r\n读到的是独立的 100 个条目。\r\n\r\n## 方案1\r\n\r\n一开始的方案是这样的：\r\n\r\n```\r\n// 这一句不一定会发请求，可能会优化成跟接下来的第一个 query 一起发出\r\nsql.begin_transaction\r\n\r\n// 第一次io发生。\r\n// 如果一个用户在 10s 内没有被更新，那么取出来\r\n// 这时候由于程序拿得到 user_id 的值，所以网络io是发生了的。否则拿不到 user_id 的值\r\noutdate_time = now() - 10s\r\nline = sql.query('select user_id where updated_time < ? order by updated_time asc limit 1', [outdate_time])\r\n\r\n// 第二次 io 发生\r\n// 更新这一行的 updated_time，免得被其他worker重复读取\r\nuser_id = line.user_id\r\nsql.query('update user_block_status set updated_time=now() where user_id = ?',\r\n [user_id])\r\n\r\n// 第三次 io 发生\r\nsql.commit\r\n\r\n// do something with user_id\r\n```\r\n\r\n可以看到，这个地方我们发起了 3 次 io 请求。当然，请求数不是很关键，因为请求数以及对应的时间是一个恒定量，\r\n而随着 worker 的增加，这一块并不会带来额外的性能瓶颈。但由于我们使用了事务，所以当 worker 由 100 增加到\r\n1000 的时候，数据库由于存在大量的事务操作，这些事务都需要掌握写锁，所以有潜在的写锁排队问题。\r\n\r\n而且关键是，方案是不可行的，根本没有起到队列的效果。\r\n\r\n为什么呢？我们假设网络io无限快，而数据库每条语句的执行时间是1s，那么我们这个事务的执行时间是 2s。\r\n这时如果 3 个 worker 并发地在同一秒（00:00）执行，那么假设 worker1 读到的 user_id 是 10086，\r\n由于读锁是共享的，worker2 和 worker3 读到的 user_id 也是 10086。这时他们三个都想要更新 10086 的值，\r\n而 worker1 抢先加了写锁，所以 worker2 和 worker3 就需要等待 worker1 的事务执行完毕，\r\n才能重新获得 10086 的写锁并进行写入。\r\n所以当 worker2 执行的时候，是 00:02 的时候，当 worker3 执行的时候，是 00:04 的时刻。\r\n而且由于他们都是在对 10086 进行更新，所以没有起到队列的效果。\r\n\r\n这里的查询条件太特殊，导致所有并发的事务需要的都是同一条数据，\r\n这时候 innodb 行级锁的特性也没有发挥出来。\r\n\r\n这个方案不仅并发时的表现类似表级锁的特性，而且也没有达到队列的效果。\r\n\r\n\r\n## 方案2\r\n\r\n将 update 语句在先，select 语句在后。\r\n\r\nupdate 语句改成\r\n\r\n```\r\noutdate_time = now() - 10s\r\nresult = sql.query('update user_block_status set updated_time=now() where updated_time < ? order by updated_time asc limit 1',\r\n [outdate_time])\r\n\r\n## each worker can get different result.user_id\r\n```\r\n\r\n这样在 update 的时候，3 个 worker 会排队，分别更新不同的 user_id 条目。然后返回来的\r\n也是不同的 user_id。\r\n\r\n可关键是，update 语句并不会将被 update 了的 id 返回给程序，所以我们后面的 select 语句拿不到对应的 user_id。\r\n这个方案先否决。\r\n\r\n## 方案3\r\n\r\n方案1的基础上，在 select 语句中，手工地干扰一下，使得不同的 worker 取到不同的条目\r\n\r\n```\r\noutdate_time = now() - 10s\r\nrandom_number = random_int(0, worker_count * 2)\r\nline = sql.query('select user_id where updated_time < ? order by updated_time asc limit 1 offset ?',\r\n [outdate_time, random_number])\r\n```\r\n\r\n这时，我们的 worker 有很大的几率可以取出不同 user_id。但这里也还有个问题就是，很可能两个 worker 的\r\nrandom_number 是同一个值。那么就发生了两次重复读取，不过对于我们的业务来说，重复读取只会造成资源的浪费，\r\n而不会带来数据一致性的问题。只要尽量减少重复读的几率，那么这个方案就是可被接受的。\r\n\r\n其中 worker_count * 2 是拍脑袋决定的数，如果数据库中始终有大量需要处理的数据，可以加大点。\r\n\r\n## 方案4\r\n\r\n方案3还是挺不完美的，虽然能解决问题，但是从概念上来说，我们需要的是队列。\r\n队列的意思就是：排队！排队！排队！\r\n\r\n方案3只是从业务逻辑层面出发，做出了一些规避，模拟了我们需要的效果。\r\n\r\n那么回到方案2，其实方案2是更接近队列的。因为不同的 worker 真正在等待另一个 worker 更新东西。\r\n可方案2无奈的是，我们拿不到被更新的id。那么有没有办法拿到呢？\r\n\r\n其实是有的，用 mysql 的 LAST_INSERT_ID() 函数。\r\n\r\n`LAST_INSERT_ID():\tValue of the AUTOINCREMENT column for the last INSERT`\r\n\r\n关于这个函数可以看看 https://dev.mysql.com/doc/refman/5.7/en/information-functions.html 这里的详细介绍。\r\n\r\n这个函数本来的含义是，拿到 AUTO_INCREMENT 那一列的最新值。也就是我们最新 insert 进表的那个 id。\r\n但实际上，它也可以作为一个 sql 语句中的变量来使用，它可以被赋值，然后取出。\r\n而且它的作用域是同一 connection 内，这样我们多个 worker 如果对 LAST_INSERT_ID 赋了不同的值，\r\n也不会互相干扰，因为不同的 worker 使用不同的 connection。\r\n\r\n这时，我们的查询在方案2的基础上就变成：\r\n\r\n```\r\nsql.begin_transaction\r\n\r\noutdate_time = now() - 10s\r\nsql.query('update user_block_status set updated_time=now()，\r\n id=LAST_INSERT_ID(id) where updated_time < ? order by updated_time asc limit 1',\r\n [outdate_time])\r\n\r\nline = sql.query('select user_id where id = LAST_INSERT_ID()')\r\n\r\n## do sth with line.user_id\r\n\r\nsql.commit\r\n```\r\n\r\nok，已经能排队了，业务上已经可以满足了。\r\n\r\n目前性能上说，网络io还是三个，而且，【行级锁】没有被利用的特定依然存在。\r\n写锁依然要排队，为什么这么说？因为不管 worker 有多少个，当他们并发的时候，where 条件都始终把它们\r\n指向同一行数据，所以还是要为了同一行数据排队。即使目前我们已经达成了【排队之后，互相更新不同条目】这个目的。\r\n\r\n方案4就总的性价比来说，目前跟方案3相比，还不一定谁好谁坏。\r\n方案4的性能在于多个worker抢一个锁，大家总是等；方案3是无脑乱取，造成资源浪费，降低worker的效率，浪费机器。\r\n\r\n什么情况下方案3好？\r\n如果总是有一大堆数据没有被处理的话，那么把方案3的乱取范围开大点，就能更好避免浪费。\r\n而当一大堆数据等待处理的时候，方案4却不停在排队，这就等于堵住了。\r\n\r\n还有一种情况就是，方案4的写锁排队已经成为瓶颈。但其实这跟上面是一回事，当总是有一大堆 worker 来取\r\n东西的话，说明就是有一大堆数据没有被处理。否则开那么多 worker 干嘛。\r\n\r\n什么情况下方案4好？\r\n前提就是，写锁排队并不成为瓶颈。如果要处理的数据并不是那么多，那么使用方案4的话，可以降低我们需要的 worker 数量，节约机器。\r\n而且 worker 数量评估可以更加理性。\r\n\r\n## 方案5\r\n\r\n那么，我们把方案3的 offset 思想加进来吧。可惜啊可惜，update 语法只支持 limit，不支持 offset。\r\n\r\n```\r\nUPDATE [LOW_PRIORITY] [IGNORE] table_reference\r\n SET col_name1={expr1|DEFAULT} [, col_name2={expr2|DEFAULT}] ...\r\n [WHERE where_condition]\r\n [ORDER BY ...]\r\n [LIMIT row_count]\r\n```\r\n\r\n那就绕一绕。\r\n\r\n不用 offset，而是通过更改 outdate_time 的值，让他们获得不同的行数据。\r\n\r\n我们的程序是要求 10s 算作过期，那么 11s、20s、30s 肯定也算过期吧。那就这样写：\r\n\r\n```\r\n// 在 10 到 30s 之间随机取值\r\noutdate_time = now() - (random(10, 30))s\r\nsql.query('update user_block_status set updated_time=now()，\r\n id=LAST_INSERT_ID(id) where updated_time < ? order by updated_time asc limit 1',\r\n [outdate_time])\r\n```\r\n\r\n`where updated_time < now() - 10s` 与 `where updated_time < now() - 12s` 与 `where updated_time < now() - 15s`（不要在 where 条件里面写计算，这只是示例）\r\n还是有可能锁定同一条数据。但至少，这个方案既利用上了行级锁，也不会造成多个 worker 处理同一 user_id 的\r\n资源浪费。\r\n\r\n\r\n## 方案6\r\n\r\n锁的问题差不多就这么解决了。\r\n\r\n我们再回头看看，发现还有个 io 问题可以再弄弄。现在还是 3 个 io 嘛。\r\n\r\n其实到了现在这步，\r\n\r\nbegin_transaction 可以去掉了。因为我们只有一个涉及写锁的操作在里面，这个操作本身作为单一语句，\r\n就已经是原子性的了。\r\n\r\n但由于我们利用了 LAST_INSERT_ID，所以我们要保证 update 语句和它之后的 select 语句在同一个 connection 中。\r\n\r\n很多的 mysql 库实现都是用了连接池的，所以同一段代码中的两条 sql 有可能会利用两条 connection，\r\n导致得到我们非预期的 user_id。\r\n\r\n但就我们的业务来说，LAST_INSERT_ID 混了其实是没关系的。每个 worker 始终还是会得到一个 unique 的 user_id。\r\n这就够了。那么我们也不必加一些多余的逻辑，保证这两条语句取到同一个 connection。\r\n\r\n这时，io 操作从 3，降低到了 2。\r\n\r\n那么，有没有可能降到 1 呢。\r\n\r\n其实也可以啊............因为基本所有 mysql 库都支持 multistatements 特性。\r\n\r\n我们可以在一条 query 写两个语句，返回接口会是一个数组，分别表示这两个语句的值。\r\n\r\n类似这样，`sql.query('update .....; select ....;')`。这是支持的。而且这么一来，\r\n同一 connection 的问题也解决了。避免为以后留坑。\r\n\r\n## 重写方案\r\n\r\n```\r\noutdate_time = now() - (random(10, 30))s\r\nresult = sql.query('update user_block_status set updated_time=now()，\r\n user_id=LAST_INSERT_ID(user_id) where updated_time < ? order by updated_time asc limit 1;\r\n\r\n select * from user_block_status where user_id = LAST_INSERT_ID()',\r\n [outdate_time])\r\n\r\n// do something with result[1].user_id\r\n```\r\n\r\n。。。。。。。。。。。。。\r\n\r\n\r\n还是有坑的。。。。。。。。。。。。。。。\r\n\r\n如果 `where updated_time < ?` 一条都不命中，那么会发生什么结果？\r\n\r\n首先，update 没有改变任何行。而 LAST_INSERT_ID 还是会返回一个合理的 id，有可能是真正的 LAST_INSERT_ID，\r\n也可能是这条 connection 中上次手工设置的。\r\n\r\n在这里可以多说一下 LAST_INSERT_ID 的特性。默认情况下，LAST_INSERT_ID() 不带参数会返回最新插入那条的 id。\r\n带参数的情况下 LAST_INSERT_ID(id) 本身的返回值就是参数，然后在接下来的调用中，如果不发生任何 insert，那么\r\n值会在 connection 中一直保持。如果发生了 insert，就会被更新。\r\n\r\n如果不处理这个 update nothing 的异常情况，当队列全部被处理完的时候，\r\n我们的 worker 会一直工作，不会停下来。所以我们要在取 LAST_INSERT_ID 的值时，\r\n判断一下上一条 update 语句到底有没有发生作用。\r\n\r\n这时候我们需要用到另一个跟 LAST_INSERT_ID 一起出现在文档中的函数，\r\n\r\nROW_COUNT():\tThe number of rows updated\r\n\r\n判断一下 ROW_COUNT，如果是 0 的话，就条件不符，这时候我们在程序里面拿到的值就是空。\r\n\r\n## 最终方案\r\n\r\n```\r\noutdate_time = now() - (random(10, 30))s\r\nresult = sql.query('update user_block_status set updated_time=now()，\r\n user_id=LAST_INSERT_ID(user_id) where updated_time < ? order by updated_time asc limit 1;\r\n\r\n select * from user_block_status where user_id = LAST_INSERT_ID()\r\n and ROW_COUNT() <> 0',\r\n [outdate_time])\r\n\r\n// do something with result[1].user_id\r\n```\r\n\r\n当然，mysql 用来解决这种队列问题可能不是一个好的方案。队列相关的知识，我还在努力学习中。\r\n\r\n\r\n## 参考资料：\r\n\r\n1. http://www.cnblogs.com/zhoujinyi/p/3437475.html\r\n2. https://dev.mysql.com/doc/refman/5.7/en/information-functions.html","title","一个简单的 mysql 队列问题","last_reply_at","2017-02-07T15:07:41.671Z","good",true,"top",19,"visit_count",13884,"2015-10-24T19:19:19.614Z","author",{"_13":56,"_23":57},"alsotang","https://avatars.githubusercontent.com/u/1147375?v=4","zones",[107],"tabs",[78,79,80,81,82,83],"publicConfig",{"_64":65,"_66":67,"_68":69},"apiBaseUrl","https://api.cnodejs.org","turnstileSiteKey","","build",{"_70":71,"_72":73,"_74":75,"_76":77},"service","cnode-web","version","0.1.0","commit","09940aa1bc0ad3dfe64235540ced544300ba2865","buildTime","2026-08-02T04:10:22Z",{"_21":16,"_85":40,"_87":106,"_89":48,"_90":16,"_92":100},{"_21":103,"_85":104,"_87":105,"_89":48,"_90":103,"_92":100},{"_21":20,"_85":101,"_87":102,"_89":48,"_90":20,"_92":100},{"_21":98,"_85":47,"_87":99,"_89":48,"_90":98,"_92":100},{"_21":94,"_85":95,"_87":96,"_89":34,"_90":97,"_92":93},{"_21":84,"_85":86,"_87":88,"_89":34,"_90":91,"_92":93},10,"key","test","label","测试","visible","sort_order",91,"scope","admin",9,"dev","开发",90,4,"精华","public","job","招聘",2,"ask","问答","分享",{"_21":16,"_108":109,"_110":102,"_111":112,"_113":114,"_89":48,"_90":16},"slug","jobs","name","description","Node.js 招聘信息专区","icon","briefcase"]