在 mysql 中实现消息队列表的最佳方法是什么
声明：本页面是StackOverFlow热门问题的中英对照翻译，遵循CC BY-SA 4.0协议，如果您需要使用它，必须同样遵循CC BY-SA许可，注明原文地址和作者信息，同时你必须将它归于原作者（不是我）：StackOverFlow
原文地址: http://stackoverflow.com/questions/423111/
Warning: these are provided under cc-by-sa 4.0 license. You are free to use/share it, But you must attribute it to the original authors (not me):
StackOverFlow
What's the best way of implementing a messaging queue table in mysql
提问by taw
It's probably the tenth time I'm implementing something like this, and I've never been 100% happy about solutions I came up with.
这可能是我第十次实施这样的事情，而且我从来没有对我提出的解决方案感到 100% 满意。
The reason using mysql table instead of a "proper" messaging system is attractive is primarily because most application already use some relational database for other stuff (which tends to be mysql for most of the stuff I've been doing), while very few applications use a messaging system. Also - relational databases have very strong ACID properties, while messaging systems often don't.
使用 mysql 表而不是“适当的”消息传递系统之所以有吸引力，主要是因为大多数应用程序已经将一些关系数据库用于其他东西（对于我一直在做的大部分事情，它往往是 mysql），而很少有应用程序使用消息系统。此外 - 关系数据库具有非常强的 ACID 属性，而消息传递系统通常没有。
The first idea is to use:
第一个想法是使用：
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) );
And then enqueue looks like this:
然后入队看起来像这样：
insert into jobs(message) values('blah blah');
And dequeue looks like this:
出队看起来像这样：
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
Table and enqueue look nice, but dequeue kinda bothers me. How likely is it to rollback? Or to get blocked? What keys I should use to make it O(1)-ish?
表和入队看起来不错，但出队有点困扰我。回滚的可能性有多大？还是要被屏蔽？我应该使用什么键来使它成为 O(1)-ish？
Or is there any better solution that what I'm doing?
或者有没有比我正在做的更好的解决方案？
采纳答案by jerebear
I've built a few message queuing systems and I'm not certain what type of message you're referring to, but in the case of the dequeuing (is that a word?) I've done the same thing you've done. Your method looks simple, clean and solid. Not that my work is the best, but it's proven very effective for large-monitoring for many sites. (error logging, mass email marketing campaigns, social networking notices)
我已经建立了一些消息队列系统，我不确定你指的是什么类型的消息，但在出队的情况下（这是一个词吗？）我做了你做过的同样的事情. 你的方法看起来简单、干净、可靠。并不是说我的工作是最好的，但事实证明它对于许多站点的大型监控非常有效。（错误记录、大量电子邮件营销活动、社交网络通知）
My vote: no worries!
我的投票：别担心！
回答by pawstrong
Your dequeue could be more concise. Rather than relying on the transaction rollback, you could do it in one atomic statement without an explicit transaction:
你的出队可以更简洁。您可以在没有显式事务的情况下在一个原子语句中完成，而不是依赖事务回滚：
UPDATE jobs SET process_id = ? WHERE process_id IS NULL ORDER BY ID ASC LIMIT 1;
Then you can pull jobs with (brackets [] mean optional, depending on your particulars):
然后，您可以使用（括号 [] 表示可选，具体取决于您的详细信息）拉取作业：
SELECT * FROM jobs WHERE process_id = ? [ORDER BY ID LIMIT 1];
回答by Gary Richardson
Brian Aker talked about a queue enginea while ago. There's been talk about a SELECT table FROM DELETEsyntax, too.
不久前，Brian Aker 谈到了队列引擎。也有人讨论过SELECT table FROM DELETE语法。
If you're not worried about throughput, you can always use SELECT GET_LOCK()as a mutex. For example:
如果您不担心吞吐量，您始终可以使用SELECT GET_LOCK()作为互斥锁。例如：
SELECT GET_LOCK('READQUEUE');
SELECT * FROM jobs;
DELETE FROM JOBS WHERE ID = ?;
SELECT RELEASE_LOCK('READQUEUE');
And if you want to get really fancy, wrap it in a stored procedure.
如果你想变得更花哨，把它包装在一个存储过程中。
回答by jgmjgm
In MySQL 8 you can use the new NOWAITand SKIP LOCKEDkeywords to avoid complexity with special locking mechanisms:
在 MySQL 8 中，您可以使用 newNOWAIT和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;
Traditionally this was hard to achieve without hacks and unusual special tables or columns, unreliable solutions or losing concurrency.
传统上，如果没有黑客攻击和不寻常的特殊表或列、不可靠的解决方案或失去并发性，这是很难实现的。
SKIP LOCKEDmay cause performance issues with extremely large numbers of consumers.
SKIP LOCKED可能会导致大量消费者出现性能问题。
This still does not however handle automatically marking the job complete on transaction rollback. For this you may need save points. That however might not solve all cases. You would really want to set an action to execute on transaction failure but as part of the transaction!
然而，这仍然不能处理在事务回滚时自动标记作业完成。为此，您可能需要保存点。然而，这可能无法解决所有情况。您真的希望设置一个操作以在事务失败时执行，但作为事务的一部分！
In future it's possible there may be more features to help optimise with cases such as an update that can also return the matched rows. It's important to keep apprised of new features and capabilities in the change log.
将来可能会有更多功能来帮助优化案例，例如也可以返回匹配行的更新。在更改日志中随时了解新特性和功能非常重要。
回答by Shumoapp
Here is a solution I used, working without the process_id of the current thread, or locking the table.
这是我使用的解决方案，在没有当前线程的 process_id 或锁定表的情况下工作。
SELECT * from jobs ORDER BY ID ASC LIMIT 0,1;
Get the result in a $row array, and execute:
在 $row 数组中获取结果，并执行：
DELETE from jobs WHERE ID=$row['ID'];
Then get the affected rows(mysql_affected_rows). If there are affected rows, process the job in the $row array. If there are 0 affected rows, it means some other process is already processing the selected job. Repeat the above steps until there are no rows.
然后获取受影响的行（mysql_affected_rows）。如果有受影响的行，则处理 $row 数组中的作业。如果有 0 个受影响的行，则意味着某个其他进程已经在处理所选作业。重复上述步骤，直到没有行。
I've tested this with a 'jobs' table having 100k rows, and spawning 20 concurrent processes that do the above. No race conditions happened. You can modify the above queries to update a row with a processing flag, and delete the row after you actually processed it:
我已经使用具有 100k 行的“jobs”表对此进行了测试，并产生了 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'];
}
Needless to say, you should use a proper message queue (ActiveMQ, RabbitMQ, etc) for this kind of work. We had to resort to this solution though, as our host regularly breaks things when updating software, so the less stuff to break the better.
不用说，您应该使用合适的消息队列（ActiveMQ、RabbitMQ 等）来完成此类工作。但是我们不得不求助于这个解决方案，因为我们的主机在更新软件时经常破坏东西，所以破坏的东西越少越好。
回答by dkretz
This threadhas design information that should be mappable.
该线程具有应该可映射的设计信息。
To quote:
报价：
Here's what I've used successfully in the past:
这是我过去成功使用的：
MsgQueue table schema
MsgQueue 表模式
MsgId identity -- NOT NULL
MsgTypeCode varchar(20) -- NOT NULL
SourceCode varchar(20) -- process inserting the message -- NULLable
State char(1) -- 'N'ew if queued, 'A'(ctive) if processing, 'C'ompleted, default 'N' -- NOT NULL
CreateTime datetime -- default GETDATE() -- NOT NULL
Msg varchar(255) -- NULLable
MsgId identity -- NOT NULL
MsgTypeCode varchar(20) -- NOT NULL
SourceCode varchar(20) -- 插入消息的过程 -- NULLable
State char(1) -- 'New if queued, 'A'(ctive) if处理，'C'完成，默认'N'--NOT NULL
CreateTime datetime--默认GETDATE()--NOT NULL
Msg varchar(255)--NULLable
Your message types are what you'd expect - messages that conform to a contract between the process(es) inserting and the process(es) reading, structured with XML or your other choice of representation (JSON would be handy in some cases, for instance).
您的消息类型是您所期望的 - 符合流程插入和流程读取之间的契约的消息，使用 XML 或您选择的其他表示形式（JSON 在某些情况下会很方便，对于实例）。
Then 0-to-n processes can be inserting, and 0-to-n processes can be reading and processing the messages, Each reading process typically handles a single message type. Multiple instances of a process type can be running for load-balancing.
然后0到n个进程可以插入，0到n个进程可以读取和处理消息，每个读取进程通常处理单个消息类型。可以运行一个进程类型的多个实例以进行负载平衡。
The reader pulls one message and changes the state to "A"ctive while it works on it. When it's done it changes the state to "C"omplete. It can delete the message or not depending on whether you want to keep the audit trail. Messages of State = 'N' are pulled in MsgType/Timestamp order, so there's an index on MsgType + State + CreateTime.
阅读器在处理消息时拉取一条消息并将状态更改为“A”活动。完成后，它会将状态更改为“C”完成。它可以根据您是否要保留审计跟踪来删除消息。状态 = 'N' 的消息按 MsgType/Timestamp 的顺序拉取，因此在 MsgType + State + CreateTime 上有一个索引。
Variations:
State for "E"rror.
Column for Reader process code.
Timestamps for state transitions.
变化：
状态为“E”错误。
Reader 进程代码列。
状态转换的时间戳。
This has provided a nice, scalable, visible, simple mechanism for doing a number of things like you are describing. If you have a basic understanding of databases, it's pretty foolproof and extensible. There's never been an issue with locks roll-backs etc. because of the atomic state transition transactions.
这提供了一种很好的、可扩展的、可见的、简单的机制来执行您所描述的许多事情。如果您对数据库有基本的了解，那么它就非常万无一失且可扩展。由于原子状态转换事务，锁回滚等从来没有问题。
回答by Mitch Wheat
I would suggest using Quartz.NET
我建议使用Quartz.NET
It has providers for SQL Server, Oracle, MySql, SQLite and Firebird.
它有 SQL Server、Oracle、MySql、SQLite 和 Firebird 的提供程序。
回答by Harshit
You can have an intermediate table to maintain the offset for the queue.
您可以有一个中间表来维护队列的偏移量。
create table scan(
scan_id int primary key,
offset_id int
);
You might have multiple scans going on as well, hence one offset per scan. Initialise the offset_id = 0 at the start of the scan.
您可能还会进行多次扫描，因此每次扫描有一个偏移量。在扫描开始时初始化 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;
All you need to do is just to maintain the last offset. This would also save you significant space (process_id per record). Hope this sounds logical.
您需要做的只是保持最后一个偏移量。这也将为您节省大量空间（每条记录的 process_id）。希望这听起来合乎逻辑。
