IOS APNS 推送，利用HTTP2.0 特性以及异步实现多用户推送

矫情吗；* 2022-11-24 15:29 145阅读 0赞

### IOS APNS 推送，利用HTTP2.0 特性以及异步实现多用户推送 ###

1.  原生的`apns`不支持直接多用户推送，如果单纯的使用`HTTP2.0`， 而不考虑其特性直接利用循环推送会非常慢（每个推送大概会花费1秒左右的时间）
2.  `httpx`，python中，支持`http2.0`请求的第三方包。[简单教程][Link 1]
3.  `http2.0`的特性，这里就不仔细说明了，可以自行查询。
4.  以下提供了`Django`和`aiohttp`两个版本的服务器代码，可自行选择。（`django`只支持3.0以上版本）

import httpx
    import asyncio
    
    async def http2conn(device_token, client, content):
        r = await client.post(f"/3/device/{device_token}", json=content)
        logger.info('ios push: ' + device_token)
        logger.info(r.status_code)
    
    # 您可以参考apple apns官方文档，重组下面的推送参数
    # 此函数可以在协程中直接调用
    async def ios_push(title, text, product, product_id, device_tokens='', **kwargs):
        """
        kwargs={'product':'', 'product_id': '产品id', 'device_token': 'device_token'}
        """
        content = {
        
            "aps": {
        
                "alert": {
        
                    "title": title,
                    "body": text,
                },
            },
            "classify": product,
            "carousel_id": product_id,
        }
        path = 'dev_certificates.pem'
        device_tokens = device_tokens.split(',')
        number = len(device_tokens) // 1000
        async with httpx.AsyncClient(base_url=settings.IOS_URL, cert=path, http2=True) as client:
            http2conn_def = partial(http2conn, client=client, content=content)
            for i in range(number + 1):
                limit = i * 1000
                await asyncio.gather(*[http2conn_def(device_token) for device_token in device_tokens[limit:limit + 1000]])
              
    # Django 版本
    async def ios(request):
        title = request.POST.get('title', None)
        text = request.POST.get('text', None)
        product = request.POST.get('product', None)
        product_id = request.POST.get('product_id', None)
        device_token = request.POST.get('device_token', None)
        await ios_push(text, title, product, product_id, device_tokens=device_token)
        return JsonResponse({
        'status': '200'}, safe=False)
    
    # 借助aiohttp实现一个简单的客户端
    @routes.post('/')
    async def ios(request):
        data = await request.post()
        title = data.get('title', None)
        text = data.get('text', None)
        product = data.get('product', None)
        product_id = data.get('product_id', None)
        device_token = data.get('device_token', None)
        print(title, text, product_id, product, device_token)
        await ios_push(text, title, product, product_id, device_tokens=device_token)
        return web.json_response({
        'msg': 'success'})
    
    
    app = web.Application()
    app.add_routes(routes)
    web.run_app(app)

[Link 1]: https://blog.csdn.net/wei_bo_cai/article/details/107846010