如何使用Flask-SocketIO向离线用户推送浏览器通知？

核心原理 #

SocketIO仅能向在线用户推送消息，离线场景需要结合浏览器Push API+Service Worker，同时后端需暂存离线消息，待用户重新上线或浏览器处于可推送状态时触发通知。

具体实现步骤 #

1. 后端配置与离线消息存储 #

• 为每个用户维护一个离线消息队列，可使用Redis、数据库（如MySQL/PostgreSQL）存储，结构包含：用户ID、消息内容、创建时间、推送状态。
• Flask-SocketIO中，发送消息时先检查用户在线状态：

  from flask_socketio import SocketIO, emit
  import json
  import redis
  
  # 初始化socketio和Redis实例
  socketio = SocketIO()
  redis_client = redis.Redis(host='localhost', port=6379, db=0)
  online_users = set()
  
  @socketio.on('connect')
  def handle_connect():
      user_id = request.args.get('user_id')  # 前端连接时携带用户ID
      online_users.add(user_id)
  
  @socketio.on('disconnect')
  def handle_disconnect():
      user_id = request.args.get('user_id')
      online_users.remove(user_id)
  
  def send_chat_message(target_user_id, message):
      if target_user_id in online_users:
          # 在线用户直接通过SocketIO推送
          emit('new_message', message, room=target_user_id)
      else:
          # 离线用户存储消息到Redis队列
          redis_client.rpush(f'offline_messages:{target_user_id}', json.dumps(message))
  

2. 前端集成Push API与Service Worker #

• 请求推送权限：页面加载时申请通知权限，获取设备推送订阅信息并发送给后端存储：

  async function initPushNotification() {
      const permission = await Notification.requestPermission();
      if (permission !== 'granted') return;
  
      const registration = await navigator.serviceWorker.register('/sw.js');
      const subscription = await registration.pushManager.subscribe({
          userVisibleOnly: true,
          applicationServerKey: urlBase64ToUint8Array('你的VAPID公钥')
      });
  
      // 发送订阅信息到后端关联用户ID
      await fetch('/save-push-sub', {
          method: 'POST',
          headers: {'Content-Type': 'application/json'},
          body: JSON.stringify({
              user_id: '当前登录用户ID',
              subscription: subscription
          })
      });
  }
  
  // 转换VAPID密钥格式的辅助函数
  function urlBase64ToUint8Array(base64String) {
      const padding = '='.repeat((4 - base64String.length % 4) % 4);
      const base64 = (base64String + padding).replace(/-/g, '+').replace(/_/g, '/');
      const rawData = window.atob(base64);
      return Uint8Array.from([...rawData].map(char => char.charCodeAt(0)));
  }
  

• 编写Service Worker文件（sw.js）：监听推送事件并展示通知，点击通知跳转至聊天页面：

  self.addEventListener('push', (event) => {
      const payload = event.data?.json() || {content: '新消息'};
      const options = {
          body: payload.content,
          icon: '/static/notify-icon.png',
          data: {url: '/chat'}
      };
      event.waitUntil(self.registration.showNotification('新聊天消息', options));
  });
  
  self.addEventListener('notificationclick', (event) => {
      event.notification.close();
      event.waitUntil(clients.openWindow(event.notification.data.url));
  });
  

3. 后端触发离线推送 #

• 使用pywebpush库结合VAPID密钥向浏览器推送服务发送通知：

  import pywebpush
  
  def send_push_notification(subscription, message):
      vapid_private_key = '你的VAPID私钥'
      vapid_claims = {'sub': 'mailto:你的联系邮箱'}
      try:
          pywebpush.webpush(
              subscription_info=subscription,
              data=json.dumps({'content': message}),
              vapid_private_key=vapid_private_key,
              vapid_claims=vapid_claims
          )
      except Exception as e:
          # 推送失败可标记消息待重试，或清理无效订阅
          print(f'推送失败: {str(e)}')
  
  # 定时任务或用户上线时检查离线消息并推送
  def process_offline_messages(user_id):
      messages = redis_client.lrange(f'offline_messages:{user_id}', 0, -1)
      if not messages:
          return
      # 从数据库获取用户的推送订阅信息
      user_sub = get_user_push_subscription(user_id)  # 自定义查询函数
      if user_sub:
          for msg in messages:
              send_push_notification(user_sub, json.loads(msg))
          # 推送成功后删除离线消息
          redis_client.delete(f'offline_messages:{user_id}')
  

4. 用户上线时的消息同步 #

• 用户重新连接SocketIO时，同步离线消息到前端：

  @socketio.on('connect')
  def handle_connect():
      user_id = request.args.get('user_id')
      online_users.add(user_id)
      # 同步离线消息
      messages = redis_client.lrange(f'offline_messages:{user_id}', 0, -1)
      if messages:
          emit('sync_offline_msgs', [json.loads(msg) for msg in messages])
          redis_client.delete(f'offline_messages:{user_id}')
  

注意事项 #

• VAPID密钥可通过web-push工具生成：web-push generate-vapid-keys
• Push API仅支持HTTPS环境（localhost可用于开发测试）
• 需处理推送订阅失效的情况，比如用户清除浏览器数据后，后端要清理无效订阅

内容的提问来源于stack exchange，提问作者Aakarsh Kumar