我有一个表，其中的行仅附加(不更新或删除)事务(我将解释为什么这很重要)，我需要获取该表的新的、以前未获取的行，每个用 cron 分钟。

我该怎么做？在任何编程语言中(我使用 Perl，但这无关紧要。)

我列出了我想到的如何解决这个问题的方法，并请你告诉我正确的方法(必须有一个...)

我想到的第一个方法是保存(在一个文件中)所获取行的最大 auto_incrementing id，因此在下一分钟我可以使用:WHERE id > $last_id .但这可能会错过行。因为在事务中插入了新行，所以保存 id = 5 行的事务有可能在保存 id = 4 行的事务之前提交。因此，cron 脚本可能检索到第 5 行而不是第 4 行，并且当第 4 行在一瞬间后提交时，它永远不会被提取(因为 4 不大于 5，即 $last_id)。

然后我想我可以让 cron 作业获取在最后两分钟内具有日期字段的所有行，检查在 cron 作业的上一次运行中再次检索了这些行中的哪些行(为此我需要将检索到的行 ID 保存在某处)，比较并仅处理新行 ID。不幸的是，这很复杂，而且也没有解决如果某个插入事务由于某些奇怪的数据库原因需要两分半钟才能提交的问题，这将导致日期对于下一次迭代来说太旧了要获取的 cron 作业。

然后我想到安装一个消息队列 (MQ)，如 RabbitMQ 或任何其他的。执行插入事务的同一进程会通知 RabbitMQ 新行，然后 RabbitMQ 会通知一个始终运行的处理新行的进程。因此，该过程不会在最后一分钟插入一批行，而是在写入新行时逐行获取。这听起来不错，但有太多的失败点——RabbitMQ 可能会停机一秒钟(例如在重启时)，在这种情况下，插入事务将在接收进程还没有接收到新行的情况下提交。所以新行将被错过。不好。

我只是想到了另一个解决方案:接收进程(有 30 个，在完全相同的数据上做完全相同的工作，所以相同的行被处理 30 次，每个接收进程一次)可以写入另一个当他们处理它时，他们已经处理了 X 行，然后当时间到了时，他们可以使用 OUTER JOIN 查询请求主表中不存在于“have_processed”表中的所有行。但我相信(如果我错了请纠正我)这样的查询会在数据库服务器上消耗大量 CPU 和 HD，因为它必须比较两个表的整个 ID 列表以找到新条目(和 table 很大，而且每分钟都在变大)。如果接收过程只有一个，那会很快——然后我就可以在主表中添加一个名为“have_read”的索引字段，这将使在数据库服务器上查找新行变得非常快速和容易。

正确的做法是什么？你有什么建议？问题很简单，但(对我而言)似乎很难找到解决方案。

谢谢。

最佳答案

我相信做到这一点的“最佳”方法是使用一个进程来检查新行并将它们委托(delegate)给三十个消费者进程。然后，从数据库的角度来看，您的问题将变得更易于管理，并且委托(delegate)流程的编写也不会那么困难。

如果您无法通过数据库与 30 个消费者进程通信，我能想到的最佳选择是在表上创建触发器，它将每一行复制到辅助表。将每一行复制到副表三十次(每个消费者进程一次)。向此辅助表添加一列，指示“目标”消费者进程(例如从 1 到 30 的数字)。每个消费者进程使用其唯一编号检查新行，然后删除这些行。如果担心有些行在处理之前就被删除了(因为consumer在处理中途crash)，可以一个一个的fetch，处理，删除。

由于通过不断删除已处理的行使辅助表保持较小，INSERT、SELECT 和 DELETE 会非常快。此辅助表上的所有操作也将由主键索引(如果您将消费者 ID 作为主键的第一个字段)。

在 MySQL 语句中，这看起来像这样:

CREATE TABLE `consumer`(
    `id` INTEGER NOT NULL,
    PRIMARY KEY (`id`)
);
INSERT INTO `consumer`(`id`) VALUES
(1),
(2),
(3)
-- all the way to 30
;

CREATE TABLE `secondaryTable` LIKE `primaryTable`;
ALTER TABLE `secondaryTable` ADD COLUMN `targetConsumerId` INTEGER NOT NULL FIRST;
-- alter the secondary table further to allow several rows with the same primary key (by adding targetConsumerId to the primary key)

DELIMTER //
CREATE TRIGGER `mark_to_process` AFTER INSERT ON `primaryTable`
FOR EACH ROW
BEGIN
    -- by doing a cross join with the consumer table, this automatically inserts the correct amount of rows and adding or deleting consumers is just a matter of adding or deleting rows in the consumer table
    INSERT INTO `secondaryTable`(`targetConsumerId`, `primaryTableId`, `primaryTableField1`, `primaryTableField2`) SELECT `consumer`.`id`, `primaryTable`.`id`, `primaryTable`.`field1`, `primaryTable`.`field2` FROM `consumer`, `primaryTable` WHERE `primaryTable`.`id` = NEW.`id`;
END//
DELIMITER ;

-- loop over the following statements in each consumer until the SELECT doesn't return any more rows
START TRANSACTION;
SELECT * FROM secondaryTable WHERE targetConsumerId = MY_UNIQUE_CONSUMER_ID LIMIT 1;
-- here, do the processing (so before the COMMIT so that crashes won't let you miss rows)
DELETE FROM secondaryTable WHERE targetConsumerId = MY_UNIQUE_CONSUMER_ID AND primaryTableId = PRIMARY_TABLE_ID_OF_ROW_JUST_SELECTED;
COMMIT;

关于mysql - 如何每分钟检索表的新行，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/18647323/