以色列队列

以色列队列的快速Python (Cython) 实现。

目录

1. 概述
2. 什么是以色列队列？
   • 为什么这很有用？
3. 安装
4. 类
   • IsraeliQueue
   • AsyncIsraeliQueue
5. 文档
6. 复杂度
7. 同步示例
8. 异步示例
9. 许可证

概述

本项目实现了一个队列系统，其中每个项目都与一个“组”相关联。队列按组处理项目，确保同一组的项目一起出队。此实现提供了一种同步版本（IsraeliQueue）和一种异步版本（AsyncIsraeliQueue），使用asyncio。

什么是以色列队列？

以色列队列是一种优先队列类型，其中任务在同一优先级下分组。将新任务添加到队列会导致它们跳过队列并一起分组。然后，按顺序，分组取出。

为什么这很有用？

以色列队列从批量处理分组任务中受益匪浅。例如，想象一个需要登录远程存储库以获取文件的机器人或API

def login(repo_name: str):
  return Session("repo_name")  # Expensive operation

def download_file(session: Session, filename: str):
  return Session.download(filename)

def logout(session: Session):
  session.logout

现在，我们有一个线程或asyncio任务将待下载的文件添加到队列中

from israeliqueue import IsraeliQueue
queue = IsraeliQueue()
queue.put("cpython", "build.bat")
queue.put("black", "pyproject.toml")
queue.put("black", "bar")  # Same repo as the second item
queue.put("cpython", "index.html")  # Same repository as the first item

普通队列会导致我们的机器人登录和登出四次，逐个处理每个项目。以色列队列将存储库分组在一起，节省了设置成本，并允许在同一个请求中下载它们

while True:
  group, items = queue.get_group()
  session = login(group)
  for item in items:
    download_file(session, item)
  logout(session)

如果下载过程一次接受多个文件，那就更高效了

session.download_files(*items)

其他用途可能包括批量处理AWS查询、批量处理numpy计算等等！

安装

要安装此包，只需pip install cisraeliqueue。

您可以在项目中如下使用这些类

from israeliqueue import IsraeliQueue, AsyncIsraeliQueue

类

IsraeliQueue

这是以色列队列的同步实现。它提供基于组的任务处理，并支持阻塞和非阻塞队列操作。该类是线程安全的，可用于多线程环境。

AsyncIsraeliQueue

这是以色列队列的异步版本，使用Python的asyncio框架构建。它提供非阻塞、异步的队列操作方法，适用于需要高并发和异步任务管理的应用程序。

文档

完整的文档可以在我们的RTD页面上找到。

复杂度

• put / put_nowait：O(1) - 在队列末尾插入。
• get / get_nowait：O(1) - 从队列前端出队。
• get_group / get_group_nowait：O(group) - 从同一组出队所有项目。
• task_done：O(1) - 简单的账务记录，用于跟踪完成的任务。
• join：O(1) - 阻塞，直到所有任务完成

同步示例

from israeliqueue import IsraeliQueue

# Initialize the queue
queue = IsraeliQueue(maxsize=10)

# Add items to the queue
queue.put('group1', 'task1')
queue.put('group1', 'task2')
queue.put('group2', 'task3')

# Get items from the queue
group, task = queue.get()
print(f"Processing {task} from {group}")

# Get all items from the same group
group, tasks = queue.get_group()
print(f"Processing all tasks from {group}: {tasks}")

# Mark the task as done
queue.task_done()

# Wait for all tasks to complete
queue.join()

异步示例

import asyncio
from israeliqueue import AsyncIsraeliQueue

async def main():
    # Initialize the queue
    queue = AsyncIsraeliQueue(maxsize=10)

    # Add items to the queue
    await queue.put('group1', 'task1')
    await queue.put('group1', 'task2')
    await queue.put('group2', 'task3')

    # Get items from the queue
    group, task = await queue.get()
    print(f"Processing {task} from {group}")

    # Get all items from the same group
    group, tasks = await queue.get_group()
    print(f"Processing all tasks from {group}: {tasks}")

    # Mark the task as done
    queue.task_done()

    # Wait for all tasks to complete
    await queue.join()

# Run the async example
asyncio.run(main())

许可证

本项目采用MIT许可证。