Python结合多线程与协程实现高效异步请求处理

import aiohttp
import asyncio

async def has_orders_async(phone, cookie, timestamp):
    async with aiohttp.ClientSession(cookies={'session': cookie}) as session:
        async with session.get(f'https://api.example.com/orders?phone={phone}') as resp:
            data = await resp.json()
            return data['total'] > 0

# 批量查询
async def batch_check(phones):
    tasks = [has_orders_async(p) for p in phones]
    return await asyncio.gather(*tasks)

from concurrent.futures import ThreadPoolExecutor

def async_to_sync(async_func):
    """装饰器：异步函数转同步"""
    def wrapper(*args, kwargs):
        loop = asyncio.new_event_loop()
        try:
            return loop.run_until_complete(async_func(*args, kwargs))
        finally:
            loop.close()
    return wrapper

@async_to_sync
async def has_orders_sync(phone, cookie, timestamp):
    # 同方案1的异步实现
    pass

def thread_pool_check(phones, workers=4):
    with ThreadPoolExecutor(workers) as executor:
        return list(executor.map(has_orders_sync, phones))

strategies = [
    {
        "name": "精确匹配", 
        "query": lambda: query_by_city(prefix, suffix, city)
    },
    {
        "name": "同省匹配",
        "query": lambda: query_by_province(prefix, suffix, province)
    },
    {
        "name": "全国匹配",
        "query": lambda: query_nationwide(prefix, suffix)
    }
]

​​​​​​​async def hierarchical_match(prefix, suffix):
    for strategy in strategies:
        phones = await strategy["query"]()
        if not phones:
            continue
            
        results = await batch_check(phones)
        if any(results):
            return phones[results.index(True)]

import java.net.http.*;
import java.util.concurrent.*;

public class OrderChecker {
    private static final HttpClient httpClient = HttpClient.newHttpClient();

    public static CompletableFuture<Boolean> hasOrder(String phone, String cookie) {
        HttpRequest request = HttpRequest.newBuilder()
            .uri(URI.create("https://api.example.com/orders?phone=" + phone))
            .header("Cookie", "session=" + cookie)
            .build();

        return httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString())
            .thenApply(response -> {
                JsonObject json = JsonParser.parseString(response.body()).getAsJsonObject();
                return json.get("total").getAsInt() > 0;
            });
    }

    public static CompletableFuture<String> batchCheck(List<String> phones) {
        List<CompletableFuture<Pair<String, Boolean>>> futures = phones.stream()
            .map(phone -> hasOrder(phone).thenApply(result -> Pair.of(phone, result)))
            .collect(Collectors.toList());

        return CompletableFuture.anyOf(futures.toArray(new CompletableFuture[0]))
            .thenApply(firstResult -> ((Pair<String, Boolean>) firstResult).getKey());
    }
}

ExecutorService executor = Executors.newFixedThreadPool(4);

List<Future<Boolean>> results = phones.stream()
    .map(phone -> executor.submit(() -> {
        // 同步HTTP请求实现
        return checkOrderSync(phone, cookie);
    }))
    .collect(Collectors.toList());

String matchedPhone = results.stream()
    .filter(future -> {
        try {
            return future.get();
        } catch (Exception e) {
            return false;
        }
    })
    .findFirst()
    .orElse(null);

# 连接池配置
connector = aiohttp.TCPConnector(limit=100, force_close=True)
session = aiohttp.ClientSession(connector=connector)

// Java重试机制
.handle((result, ex) -> {
    if (ex != null) {
        return retry(phone);
    }
    return result;
})