Guide de développement de coroutines asynchrones : création d'un système de recommandation hautes performances

Avec le développement rapide d'Internet et de l'Internet mobile, la quantité de données a explosé. Comment traiter efficacement les données est devenu un enjeu important auquel sont confrontées les équipes R&D des grandes entreprises. Les systèmes de recommandation constituent l’un des domaines d’application clés et sont largement utilisés dans de nombreuses entreprises. Les coroutines asynchrones sont une technologie importante pour obtenir un traitement de données hautes performances dans des scénarios à forte concurrence. Cet article explique comment utiliser les coroutines asynchrones pour créer un système de recommandation hautes performances et fournit des exemples de code spécifiques.

1. Qu'est-ce que la coroutine asynchrone ?

La coroutine asynchrone est un modèle de programmation concurrente très efficace. Elle a été initialement proposée et implémentée par le langage Python. Elle a été empruntée et développée par de nombreux langages, comme goroutine dans le langage Go, SwiftNIO dans Swift, etc. Les coroutines asynchrones prennent en charge les opérations d'E/S asynchrones hautement concurrentes en commutant au niveau des coroutines.

Par rapport au multi-threading, les coroutines asynchrones présentent les avantages suivants :

1. Plus efficaces : les coroutines asynchrones peuvent implémenter un modèle de thread très léger avec très peu de frais de commutation.
2. Plus flexible : la commutation entre les coroutines n'a pas besoin d'entrer dans le noyau, mais est contrôlée par le programme, de sorte que le nombre et les méthodes de planification des coroutines peuvent être contrôlés de manière plus flexible.
3. Plus facile à utiliser : par rapport au mécanisme de verrouillage multi-thread, les coroutines asynchrones peuvent éviter les problèmes multi-thread tels que les verrous grâce à la planification collaborative, rendant le code plus simple et plus facile à utiliser.

2. Scénarios d'application de coroutines asynchrones dans les systèmes de recommandation

Les systèmes de recommandation doivent traiter une grande quantité de données pendant la mise en œuvre, telles que les journaux de comportement des utilisateurs, les informations sur les attributs des éléments, etc., tandis que les coroutines asynchrones peuvent réaliser un traitement de données hautes performances. . Plus précisément, les scénarios d'application suivants dans le système de recommandation conviennent à l'utilisation de coroutines asynchrones :

1. Extraction des fonctionnalités d'intérêt de l'utilisateur : la lecture et le traitement asynchrones des journaux de comportement des utilisateurs sont implémentés via des coroutines asynchrones, et les fonctionnalités d'intérêt de l'utilisateur sont extraites pour prendre en charge des recommandations.
2. Agrégation des informations sur les articles : la lecture et le traitement asynchrones des informations sur les attributs des articles sont réalisés via des coroutines asynchrones, et diverses informations sont agrégées pour prendre en charge une recommandation complète des articles.
3. Tri des résultats des recommandations : le tri et le filtrage rapides des résultats des recommandations sont mis en œuvre via des coroutines asynchrones pour garantir un débit élevé et une faible latence du système de recommandation.

3. Guide de développement de coroutines asynchrones

Ce qui suit présentera le guide de développement des coroutines asynchrones sous trois aspects : le processus de développement de coroutines, le mécanisme de planification et les opérations d'E/S asynchrones.

1. Processus de développement de coroutines

Dans les coroutines asynchrones, vous devez utiliser une bibliothèque de coroutines pour réaliser la création, la commutation et la planification des coroutines. Actuellement, les bibliothèques de coroutines les plus populaires incluent asyncio en Python, goroutine en Go et SwiftNIO en Swift.

Prenons asyncio en Python comme exemple pour implémenter un simple programme de coroutine asynchrone :

import asyncio

async def foo():
    await asyncio.sleep(1)
    print('Hello World!')

loop = asyncio.get_event_loop()
loop.run_until_complete(foo())

Dans le programme ci-dessus, asyncio.sleep(1) signifie laisser la coroutine actuelle dormir pendant 1 seconde. simuler des opérations d'E/S asynchrones, la fonction déclarée par async def représente une fonction asynchrone. Utilisez loop.run_until_complete() dans le programme pour exécuter la coroutine, et le résultat de sortie est Hello World !. asyncio.sleep(1) 表示让当前协程休眠 1 秒钟，以模拟异步 I/O 操作，async def 声明的函数表示异步函数。在程序中使用 loop.run_until_complete() 来运行协程，输出结果为 Hello World!。

2. 调度机制

在异步协程中，协程的调度是非常重要的一环。通过异步协程的协作式调度，可以更加灵活地控制协程的数量和调度顺序，以达到最优的性能表现。

在 asyncio 中，使用 asyncio.gather() 方法来执行多个协程，例如：

import asyncio

async def foo():
    await asyncio.sleep(1)
    print('foo')

async def bar():
    await asyncio.sleep(2)
    print('bar')

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(foo(), bar()))

上述程序中，asyncio.gather() 可以同时执行多个协程，输出结果为 foo 和 bar。这里的两个协程的时间长度分别为 1 秒和 2 秒，因此输出顺序为 foo 和 bar。

3. 异步 I/O 操作

在推荐系统中，需要使用异步 I/O 操作来处理大量的用户行为日志、物品属性信息等数据。在异步协程中使用异步 I/O 操作可以大大提高数据读取和处理的效率。

在 asyncio 中，使用 asyncio.open() 方法来异步读取文件，例如：

import asyncio

async def read_file():
    async with aiofiles.open('data.log', 'r') as f:
        async for line in f:
            print(line.strip())

loop = asyncio.get_event_loop()
loop.run_until_complete(read_file())

上述程序中，使用 async with aiofiles.open() 来异步打开文件，使用 async for line in f 来异步读取文件中的每行数据。在程序中使用 loop.run_until_complete()

Mécanisme de planification

Dans les coroutines asynchrones, la planification des coroutines est une partie très importante. Grâce à la planification collaborative de coroutines asynchrones, le nombre et l'ordre de planification des coroutines peuvent être contrôlés de manière plus flexible pour obtenir des performances optimales.

Dans asyncio, utilisez la méthode asyncio.gather() pour exécuter plusieurs coroutines, par exemple :

1. import asyncio
   import json
   
   async def extract_feature(data):
       result = {}
       for item in data:
           uid = item.get('uid')
           if uid not in result:
               result[uid] = {'click': 0, 'expose': 0}
           if item.get('type') == 'click':
               result[uid]['click'] += 1
           elif item.get('type') == 'expose':
               result[uid]['expose'] += 1
       return result
   
   async def read_file():
       async with aiofiles.open('data.log', 'r') as f:
           data = []
           async for line in f:
               data.append(json.loads(line))
               if len(data) >= 1000:
                   result = await extract_feature(data)
                   print(result)
                   data = []
   
           if len(data) > 0:
               result = await extract_feature(data)
               print(result)
   
   loop = asyncio.get_event_loop()
   loop.run_until_complete(read_file())

Dans le programme ci-dessus, asyncio.gather() peut être exécuté en même temps Exécutez plusieurs coroutines et les résultats de sortie sont foo et bar. Les durées de temps des deux coroutines ici sont respectivement de 1 seconde et 2 secondes, donc l'ordre de sortie est foo et bar.

Opérations d'E/S asynchrones

🎜🎜Dans le système de recommandation, les opérations d'E/S asynchrones doivent être utilisées pour traiter une grande quantité de journaux de comportement des utilisateurs, d'informations sur les attributs des éléments et d'autres données. L'utilisation d'opérations d'E/S asynchrones dans des coroutines asynchrones peut considérablement améliorer l'efficacité de la lecture et du traitement des données. 🎜🎜Dans asyncio, utilisez la méthode asyncio.open() pour lire les fichiers de manière asynchrone, par exemple : 🎜

import asyncio
import json

async def aggregate_info(data):
    result = {}
    for item in data:
        key = item.get('key')
        if key not in result:
            result[key] = []
        result[key].append(item.get('value'))
    return result

async def read_file():
    async with aiofiles.open('data.log', 'r') as f:
        data = []
        async for line in f:
            data.append(json.loads(line))
            if len(data) >= 1000:
                result = await aggregate_info(data)
                print(result)
                data = []

        if len(data) > 0:
            result = await aggregate_info(data)
            print(result)

loop = asyncio.get_event_loop()
loop.run_until_complete(read_file())

🎜Dans le programme ci-dessus, utilisez async avec aiofiles.open() Pour ouvrir le fichier de manière asynchrone, utilisez async for line in f pour lire de manière asynchrone chaque ligne de données du fichier. Utilisez loop.run_until_complete() dans le programme pour exécuter la coroutine. 🎜🎜4. Exemples de code spécifiques🎜🎜Ce qui suit est une introduction détaillée à la méthode d'implémentation des coroutines asynchrones dans le système de recommandation. 🎜🎜🎜Extraction des fonctionnalités d'intérêt des utilisateurs🎜🎜🎜Dans le système de recommandation, l'extraction des fonctionnalités d'intérêt des utilisateurs est un lien très critique. Les journaux de comportement des utilisateurs sont l'une des données importantes dans les systèmes de recommandation, les E/S asynchrones doivent donc être utilisées pour lire et traiter les journaux de comportement afin d'extraire les caractéristiques d'intérêt des utilisateurs. 🎜

import asyncio
import json

async def extract_feature(data):
    result = {}
    for item in data:
        uid = item.get('uid')
        if uid not in result:
            result[uid] = {'click': 0, 'expose': 0}
        if item.get('type') == 'click':
            result[uid]['click'] += 1
        elif item.get('type') == 'expose':
            result[uid]['expose'] += 1
    return result

async def read_file():
    async with aiofiles.open('data.log', 'r') as f:
        data = []
        async for line in f:
            data.append(json.loads(line))
            if len(data) >= 1000:
                result = await extract_feature(data)
                print(result)
                data = []

        if len(data) > 0:
            result = await extract_feature(data)
            print(result)

loop = asyncio.get_event_loop()
loop.run_until_complete(read_file())

上述程序中，extract_feature() 函数用于从用户行为日志中提取用户兴趣特征，read_file() 函数读取用户行为日志，并调用 extract_feature() 函数进行用户特征提取。在程序中，使用 if len(data) >= 1000 判断每次读取到的数据是否满足处理的条件。

2. 物品信息聚合

在推荐系统中，物品信息的聚合是支持物品的综合推荐的必要环节。物品属性信息是推荐系统中的重要数据之一，因此需要使用异步 I/O 来进行读取和处理。

import asyncio
import json

async def aggregate_info(data):
    result = {}
    for item in data:
        key = item.get('key')
        if key not in result:
            result[key] = []
        result[key].append(item.get('value'))
    return result

async def read_file():
    async with aiofiles.open('data.log', 'r') as f:
        data = []
        async for line in f:
            data.append(json.loads(line))
            if len(data) >= 1000:
                result = await aggregate_info(data)
                print(result)
                data = []

        if len(data) > 0:
            result = await aggregate_info(data)
            print(result)

loop = asyncio.get_event_loop()
loop.run_until_complete(read_file())

上述程序中，aggregate_info() 函数用于从物品属性信息中聚合物品信息，read_file() 函数读取物品属性信息，并调用 aggregate_info() 函数进行信息聚合。在程序中，使用 if len(data) >= 1000 判断每次读取到的数据是否满足处理的条件。

3. 推荐结果排序

在推荐系统中，推荐结果的排序是支持高吞吐量和低延迟的关键环节。通过异步协程进行推荐结果的排序和过滤，可以大大提高推荐系统的性能表现。

import asyncio

async def sort_and_filter(data):
    data.sort(reverse=True)
    result = []
    for item in data:
        if item[1] > 0:
            result.append(item)
    return result[:10]

async def recommend():
    data = [(1, 2), (3, 4), (2, 5), (7, 0), (5, -1), (6, 3), (9, 8)]
    result = await sort_and_filter(data)
    print(result)

loop = asyncio.get_event_loop()
loop.run_until_complete(recommend())

上述程序中，sort_and_filter() 函数用于对推荐结果进行排序和过滤，并只返回前 10 个结果。recommend() 函数用于模拟推荐结果的生成，调用 sort_and_filter() 函数进行结果排序和过滤。在程序中，使用 0 或者 0 以下的值来模拟不需要的结果。

总结

本文介绍了异步协程的基本知识和在推荐系统中的应用，并提供了具体的代码示例。异步协程作为一种高效的并发编程技术，在大数据场景下具有广泛的应用前景。需要注意的是，在实际应用中，需要根据具体的业务需求和技术场景进行针对性的选择和调优，以达到最优的性能表现。

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!