在 mysql 中实现消息传递队列表的最佳方法是什么

Question

这可能是我第十次实施这样的事情了，而且我从来没有对我提出的解决方案感到 100% 满意。

使用 mysql 表而不是“正确的”消息传递系统之所以有吸引力，主要是因为大多数应用程序已经使用一些关系数据库来处理其他事情（对于我一直在做的大部分事情来说，这往往是 mysql ），而很少有应用程序使用消息系统。 此外，关系数据库具有非常强的 ACID 属性，而消息传递系统通常没有。

第一个想法是使用：

create table jobs(
  id auto_increment not null primary key,
  message text not null,
  process_id varbinary(255) null default null,
  key jobs_key(process_id) 
);

然后入队看起来像这样：

insert into jobs(message) values('blah blah');

出队看起来像这样：

begin;
select * from jobs where process_id is null order by id asc limit 1;
update jobs set process_id = ? where id = ?; -- whatever i just got
commit;
-- return (id, message) to application, cleanup after done

表和入队看起来不错，但出队有点困扰我。 回滚的可能性有多大？ 还是要被封杀？ 我应该使用什么键来使其 O(1) 左右？

或者有比我正在做的更好的解决方案吗？

Answer 1

你的出队可以更简洁。 您可以在没有显式事务的情况下在一个原子语句中完成此操作，而不是依赖事务回滚：

UPDATE jobs SET process_id = ? WHERE process_id IS NULL ORDER BY ID ASC LIMIT 1;

然后您可以使用以下命令拉取作业（括号 [] 表示可选，具体取决于您的具体情况）：

SELECT * FROM jobs WHERE process_id = ? [ORDER BY ID LIMIT 1];

Answer 2

我已经构建了一些消息队列系统，我不确定您指的是哪种类型的消息，但是在出队的情况下（这是一个词吗？）我做了和您一样的事情。 你的方法看起来简单、干净、可靠。 并不是说我的工作是最好的，但事实证明它对于许多站点的大型监控非常有效。 （错误记录、群发电子邮件营销活动、社交网络通知）

我的投票：不用担心！

Answer 3

Brian Aker 不久前谈到了队列引擎。 也有人讨论过 SELECT table FROM DELETE 语法。

如果您不担心吞吐量，则可以随时使用 SELECT GET_LOCK() 作为互斥体。 例如：

SELECT GET_LOCK('READQUEUE');
SELECT * FROM jobs;
DELETE FROM JOBS WHERE ID = ?;
SELECT RELEASE_LOCK('READQUEUE');

如果您想要变得非常奇特，请将其包装在存储过程中。

Answer 4

在 MySQL 8 中，您可以使用新的 NOWAIT 和 SKIP LOCKED 关键字来避免特殊锁​​定机制的复杂性：

START TRANSACTION;
SELECT id, message FROM jobs
 WHERE process_id IS NULL
 ORDER BY id ASC LIMIT 1
 FOR UPDATE SKIP LOCKED;
UPDATE jobs
 SET process_id = ?
 WHERE id = ?;
COMMIT;

传统上，如果没有黑客和不寻常的特殊表或列，这是很难实现的，不可靠的解决方案或失去并发性。

SKIP LOCKED 可能会导致大量消费者出现性能问题。

然而，这仍然不能处理在事务回滚时自动标记作业完成的情况。 为此，您可能需要保存点。 然而，这可能并不能解决所有情况。 您确实希望设置一个在事务失败时执行的操作，但作为事务的一部分！

将来可能会有更多功能来帮助优化某些情况，例如也可以返回匹配行的更新。 在更改日志中随时了解新特性和功能非常重要。

Answer 5

这是我使用的解决方案，在没有当前线程的 process_id 的情况下工作，或者锁定表。

SELECT * from jobs ORDER BY ID ASC LIMIT 0,1;

获取 $row 数组中的结果，并执行：

DELETE from jobs WHERE ID=$row['ID'];

然后获取受影响的行(mysql_affected_rows)。 如果有受影响的行，则处理 $row 数组中的作业。 如果有 0 行受影响，则意味着其他进程已经在处理所选作业。 重复以上步骤，直到没有行为止。

我已经使用具有 100k 行的“作业”表对此进行了测试，并生成了 20 个执行上述操作的并发进程。 没有发生竞争条件。 您可以修改上述查询来更新带有处理标志的行，并在实际处理后删除该行：

while(time()-$startTime<$timeout)
{
SELECT * from jobs WHERE processing is NULL ORDER BY ID ASC LIMIT 0,1;
if (count($row)==0) break;
UPDATE jobs set processing=1 WHERE ID=$row['ID'];
if (mysql_affected_rows==0) continue;
//process your job here
DELETE from jobs WHERE ID=$row['ID'];
}

不用说，您应该使用适当的消息队列（ActiveMQ、RabbitMQ 等）来完成此类工作。 不过，我们不得不求助于这个解决方案，因为我们的主机在更新软件时经常会破坏一些东西，所以破坏的东西越少越好。

Answer 6

我建议使用 Quartz.NET

它有 SQL Server、Oracle、MySql、SQLite 和 Firebird 的提供程序。

Answer 7

此线程有设计应可映射的信息。

引用：

这是我过去成功使用过的内容：

MsgQueue 表架构

MsgId 标识 -- NOT NULL
MsgTypeCode varchar(20) -- NOT NULL
SourceCode varchar(20) -- 插入消息的过程 -- NULLable
State char(1) -- 如果已排队，则为“新”；如果正在处理，则为“A”（活动）；如果已完成，则为“C”；默认为“N”—— NOT NULL
CreateTime 日期时间 -- 默认 GETDATE() -- NOT NULL
Msg varchar(255) -- NULLable

您的消息类型是您所期望的 - 符合插入进程和读取进程之间契约的消息，使用 XML 或您选择的其他表示形式（JSON例如，在某些情况下会很方便）。

然后0到n个进程可以插入，并且0到n个进程可以读取和处理消息。每个读取进程通常处理单个消息类型。 可以运行某个进程类型的多个实例以实现负载平衡。

读取器拉取一条消息，并在处理该消息时将状态更改为“A”active。 完成后，它将状态更改为“C”完成。 它可以删除该消息或不删除该消息，具体取决于您是否要保留审核跟踪。 State = 'N' 的消息按照 MsgType/Timestamp 顺序拉取，因此在 MsgType + State + CreateTime 上有一个索引。

变体：
“E”错误状态。
Reader 进程代码列。
状态转换的时间戳。

这提供了一个很好的、可扩展的、可见的、简单的机制来完成您所描述的许多事情。 如果您对数据库有基本的了解，那么它是非常简单且可扩展的。 由于原子状态转换事务，从来没有出现过锁回滚等问题。

Answer 8

您可以使用中间表来维护队列的偏移量。

create table scan(
  scan_id int primary key,
  offset_id int
);

您可能还会进行多次扫描，因此每次扫描会有一个偏移量。 在扫描开始时初始化 offset_id = 0。

begin;
select * from jobs where order by id where id > (select offset_id from scan where scan_id = 0)  asc limit 1;
update scan set offset_id = ? where scan_id = ?; -- whatever i just got
commit;

您所需要做的只是保持最后的偏移量。 这也将为您节省大量空间（每条记录的 process_id）。 希望这听起来合乎逻辑。