我希望使用数据库表作为队列。我想按照插入的顺序(FIFO)在其中插入元素并从中取出元素。我的主要考虑因素是性能,因为每秒钟会有成千上万个此类事务。因此,我希望使用 SQL 查询来获取第一个元素,而无需搜索整个表。当我读取数据时,不会删除行。 使用 SELECT TOP 1 ….. 在这里有帮助吗? 我应该使用任何特殊的索引吗?
我希望使用数据库表作为队列。我想按照插入的顺序(FIFO)在其中插入元素并从中取出元素。我的主要考虑因素是性能,因为每秒钟会有成千上万个此类事务。因此,我希望使用 SQL 查询来获取第一个元素,而无需搜索整个表。当我读取数据时,不会删除行。 使用 SELECT TOP 1 ….. 在这里有帮助吗? 我应该使用任何特殊的索引吗?
我会使用一个IDENTITY字段作为主键,为每个排队的项目提供唯一递增的ID,并在其上放置聚集索引。这将表示排队项目的顺序。
为了在处理它们时保留项目在队列表中,您需要一个“状态”字段来指示特定项目的当前状态(例如,0=等待,1=正在处理,2=已处理)。这是为了防止对项目进行两次处理所必需的。
在处理队列中的项目时,您需要找到表中下一个目前未被处理的项目。这需要以一种方式进行,以避免多个进程同时选择相同的项目进行处理,如下所示。请注意,在实现队列时,需要了解UPDLOCK和READPAST这些表提示。
例如,在sproc中,可以像这样:
DECLARE @NextID INTEGER
BEGIN TRANSACTION
-- Find the next queued item that is waiting to be processed
SELECT TOP 1 @NextID = ID
FROM MyQueueTable WITH (UPDLOCK, READPAST)
WHERE StateField = 0
ORDER BY ID ASC
-- if we've found one, mark it as being processed
IF @NextId IS NOT NULL
UPDATE MyQueueTable SET Status = 1 WHERE ID = @NextId
COMMIT TRANSACTION
-- If we've got an item from the queue, return to whatever is going to process it
IF @NextId IS NOT NULL
SELECT * FROM MyQueueTable WHERE ID = @NextID
如果处理一个项目失败,你想要之后能够再次尝试处理吗?如果是这样的话,你将需要将状态重置回0或其他值。这需要更多的思考。
或者,不要使用数据库表作为队列,而是使用像 MSMQ 这样的东西 - 只是我随便提出来的一个想法!
一切都取决于您的数据库引擎/实现。
对我来说,简单的队列在具有以下列的表上:
id / task / priority / date_added
通常有效。
我使用优先级和任务来分组任务,在出现重复任务的情况下,我选择优先级更高的那个。
不用担心 - 对于现代数据库来说,“数千条”并不算特别多。
ORDER BY
语句的块,循环并记住上一个日期时间,以便在获取下一个块时使用。由于您不从表中删除记录,因此需要在(processed, id)
上创建一个组合索引,其中processed
是指示当前记录是否已处理的列。
最好的方法是为记录创建一个分区表,并将PROCESSED
字段设置为分区键。这样,您可以保留三个或更多本地索引。
然而,如果您总是按照id
顺序处理记录,并且只有两种状态,则更新记录意味着只需将记录从索引的第一个叶子节点取出并附加到最后一个叶子节点。
当前处理的记录始终具有所有未处理记录中最小的id
和所有已处理记录中最大的id
。
WHERE
和ORDER BY
子句即可。module.exports.pickBatchInstructions = (db, batchSize) => {
const buf = crypto.randomBytes(8); // Create a unique batch identifier
const q_pickBatch = `
UPDATE
instructions
SET
status = '${status.INSTRUCTION_INPROGRESS}',
run_id = '${buf.toString("hex")}',
mdate = datetime(datetime(), 'localtime')
WHERE
id IN (SELECT id
FROM instructions
WHERE
status is not '${status.INSTRUCTION_COMPLETE}'
and run_id is null
ORDER BY
length(targetpath), id
LIMIT ${batchSize});
`;
db.run(q_pickBatch); // Change the status and set the run id
const q_getInstructions = `
SELECT
*
FROM
instructions
WHERE
run_id = '${buf.toString("hex")}'
`;
const rows = db.all(q_getInstructions); // Get all rows with this batch id
return rows;
};
针对此问题,一个非常简单的解决方案是使用更改跟踪机制(而不是数据捕获),以避免事务、锁等。它利用版本控制来跟踪每个添加/更新/删除的行,因此您可以在特定版本之后跟踪发生的更改。
因此,您需要持久化最新版本并查询新的更改。
如果查询失败,您可以随时返回并查询上一个版本的数据。 此外,如果您不想通过一次查询获取所有更改,您可以按照最新版本的顺序获取前n个,并存储最大版本ID,以便下次再次查询。
例如,请参见在SQL Server 2008中使用更改跟踪。