Panduan Pembangunan Coroutine Asynchronous: Membina Sistem Pengesyoran Berprestasi Tinggi

Dengan perkembangan pesat Internet dan Internet mudah alih, jumlah data telah meletup. Cara memproses data dengan cekap telah menjadi isu penting yang dihadapi oleh pasukan R&D syarikat utama. Sistem pengesyoran adalah salah satu bidang aplikasi utama dan digunakan secara meluas dalam banyak perusahaan. Coroutine tak segerak ialah teknologi penting untuk mencapai pemprosesan data berprestasi tinggi dalam senario konkurensi tinggi Artikel ini akan memperkenalkan cara menggunakan coroutine tak segerak untuk membina sistem pengesyoran berprestasi tinggi dan menyediakan contoh kod khusus.

1. Apakah coroutine tak segerak?

Asynchronous coroutine ialah model pengaturcaraan serentak yang sangat cekap Ia pada asalnya dicadangkan dan dilaksanakan oleh bahasa Python Ia telah dipinjam dan dibangunkan oleh banyak bahasa, seperti goroutine dalam bahasa Go, SwiftNIO dalam Swift, dll. Coroutine tak segerak menyokong operasi I/O tak segerak yang sangat serentak dengan menukar pada tahap coroutine.

Berbanding dengan multi-threading, coroutine tak segerak mempunyai kelebihan berikut:

1. Lebih cekap: Coroutine tak segerak boleh melaksanakan model benang yang sangat ringan dengan overhed pensuisan yang sangat sedikit.
2. Lebih fleksibel: Bertukar antara coroutine tidak perlu memasuki kernel, tetapi dikawal oleh program, jadi kaedah nombor dan penjadualan coroutine boleh dikawal dengan lebih fleksibel.
3. Lebih mudah digunakan: Berbanding dengan mekanisme kunci berbilang benang, coroutine tak segerak boleh mengelakkan masalah berbilang benang seperti kunci melalui penjadualan kolaboratif, menjadikan kod lebih ringkas dan mudah digunakan.

2. Senario aplikasi coroutine tak segerak dalam sistem pengesyoran

Sistem pengesyoran perlu memproses sejumlah besar data semasa pelaksanaan, seperti log tingkah laku pengguna, maklumat atribut item, dll., manakala coroutine tak segerak boleh mencapai pemprosesan data berprestasi tinggi . Khususnya, senario aplikasi berikut dalam sistem pengesyoran adalah sesuai untuk penggunaan coroutine tak segerak:

1. Pengeluaran ciri minat pengguna: Pembacaan tidak segerak dan pemprosesan log tingkah laku pengguna dilaksanakan melalui coroutine tak segerak, dan ciri minat pengguna diekstrak untuk menyokong diperibadikan. cadangan .
2. Penggabungan maklumat item: Pembacaan tidak segerak dan pemprosesan maklumat atribut item direalisasikan melalui coroutine tak segerak, dan pelbagai maklumat diagregatkan untuk menyokong pengesyoran komprehensif item.
3. Isih hasil pengesyoran: Pengisihan pantas dan penapisan hasil pengesyoran dilaksanakan melalui coroutine tak segerak untuk memastikan daya pemprosesan tinggi dan kependaman rendah sistem pengesyoran.

3. Panduan Pembangunan Coroutine Asynchronous

Berikut akan memperkenalkan panduan pembangunan untuk coroutine tak segerak dari tiga aspek: proses pembangunan coroutine, mekanisme penjadualan dan operasi I/O tak segerak.

1. Proses pembangunan Coroutine

Dalam coroutine tak segerak, anda perlu menggunakan perpustakaan coroutine untuk merealisasikan penciptaan, penukaran dan penjadualan coroutine. Pada masa ini, perpustakaan coroutine yang lebih popular termasuk asyncio dalam Python, goroutine dalam Go dan SwiftNIO dalam Swift.

Ambil asyncio dalam Python sebagai contoh untuk melaksanakan program coroutine tak segerak yang mudah:

import asyncio

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

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

Dalam program di atas, asyncio.sleep(1) bermaksud membiarkan coroutine semasa tidur selama 1 saat Untuk simulasi operasi I/O tak segerak, fungsi yang diisytiharkan oleh async def mewakili fungsi tak segerak. Gunakan loop.run_until_complete() dalam program untuk menjalankan coroutine dan hasil output ialah 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()

Mekanisme penjadualan

Dalam coroutine tak segerak, penjadualan coroutine adalah bahagian yang sangat penting. Melalui penjadualan kolaboratif coroutine tak segerak, bilangan dan susunan penjadualan coroutine boleh dikawal dengan lebih fleksibel untuk mencapai prestasi optimum.

Dalam asyncio, gunakan kaedah asyncio.gather() untuk melaksanakan berbilang coroutine, contohnya:

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())

Dalam atur cara di atas, asyncio.gather() boleh dilaksanakan pada masa yang sama Laksanakan berbilang coroutine, dan hasil output ialah foo dan bar. Tempoh masa kedua-dua coroutine di sini ialah masing-masing 1 saat dan 2 saat, jadi susunan output ialah foo dan bar.

Operasi I/O tak segerak

🎜🎜Dalam sistem pengesyoran, operasi I/O tak segerak perlu digunakan untuk memproses sejumlah besar log tingkah laku pengguna, maklumat atribut item dan data lain. Menggunakan operasi I/O tak segerak dalam coroutine tak segerak boleh meningkatkan kecekapan pembacaan dan pemprosesan data. 🎜🎜Dalam asyncio, gunakan kaedah asyncio.open() untuk membaca fail secara tidak segerak, contohnya: 🎜

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())

🎜Dalam atur cara di atas, gunakan async dengan aiofiles.open() Untuk membuka fail secara tak segerak, gunakan async for line in f untuk membaca secara tak segerak setiap baris data dalam fail. Gunakan loop.run_until_complete() dalam program untuk menjalankan coroutine. 🎜🎜4. Contoh kod khusus🎜🎜Berikut ialah pengenalan terperinci kepada kaedah pelaksanaan coroutine tak segerak dalam sistem pengesyoran. 🎜🎜🎜Pengeluaran ciri minat pengguna🎜🎜🎜Dalam sistem pengesyoran, pengekstrakan ciri minat pengguna ialah pautan yang sangat kritikal. Log tingkah laku pengguna ialah salah satu data penting dalam sistem pengesyoran, jadi I/O tak segerak perlu digunakan untuk membaca dan memproses log tingkah laku untuk mengekstrak ciri minat pengguna. 🎜

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 以下的值来模拟不需要的结果。

总结

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

Atas ialah kandungan terperinci Panduan Pembangunan Coroutine Asynchronous: Membina Sistem Pengesyoran Berprestasi Tinggi. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!