Python中如何实现一个简单的业务场景

业务描述

  • 有很多 client 与服务器 socket 通信
  • 有一个 redis 队列存了特定 client 需要的信息
  • client 发送请求连接给服务器,服务器在 redis 中找到这个 client 需要的信息(如果没有一直等待),找到后返回给这个 client,结束.

我原本的想法

  • 服务器是 socketserver 多线程,当有个 client 来的时候,就循环在 redis 中找,一直到找到后再发送给 client

弊端

  • 我想了下,如果我有 10000 个 client,会出现两个问题:
    • 10000 个 client 一直与我的服务器连接中(会不会服务器不支持这个多的连接同时存在)
    • 10000 个 client 每一个都要循环去 redis 中找数据 相同的操作重复了 10000 次,好蠢
    • 其实只要有一个循环一直查 redis,查到参数,就发给 client 与服务器连接的那个 socket 就可以(不知道如果实现这一步的想法)

希望大佬能指点一二呀= =,


Python中如何实现一个简单的业务场景

19 回复

循环查 redis 是什么意思?


帖子内容比较宽泛,我来举一个最常见的业务场景例子:用户注册。这个场景会涉及数据验证、业务逻辑处理和简单的数据持久化。

下面是一个完整的、可运行的实现,使用SQLite数据库和Python内置模块,不依赖外部框架。

import sqlite3
import hashlib
import re
from datetime import datetime

class UserRegistrationSystem:
    def __init__(self, db_path='users.db'):
        """初始化,连接数据库并创建用户表"""
        self.conn = sqlite3.connect(db_path)
        self.cursor = self.conn.cursor()
        self._create_table()

    def _create_table(self):
        """创建用户表,如果不存在"""
        self.cursor.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                username TEXT UNIQUE NOT NULL,
                email TEXT UNIQUE NOT NULL,
                password_hash TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        self.conn.commit()

    def validate_username(self, username):
        """验证用户名:长度3-20,只允许字母数字和下划线"""
        if not re.match(r'^[a-zA-Z0-9_]{3,20}$', username):
            return False, "用户名必须是3-20位的字母、数字或下划线"
        return True, ""

    def validate_email(self, email):
        """简单的邮箱格式验证"""
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(pattern, email):
            return False, "邮箱格式不正确"
        return True, ""

    def validate_password(self, password):
        """密码验证:至少8位,包含字母和数字"""
        if len(password) < 8:
            return False, "密码至少需要8位"
        if not re.search(r'[a-zA-Z]', password) or not re.search(r'\d', password):
            return False, "密码必须包含字母和数字"
        return True, ""

    def hash_password(self, password):
        """使用SHA-256哈希密码"""
        return hashlib.sha256(password.encode()).hexdigest()

    def check_user_exists(self, username, email):
        """检查用户名或邮箱是否已存在"""
        self.cursor.execute(
            "SELECT COUNT(*) FROM users WHERE username = ? OR email = ?",
            (username, email)
        )
        count = self.cursor.fetchone()[0]
        return count > 0

    def register_user(self, username, email, password):
        """核心注册逻辑"""
        # 1. 数据验证
        valid, msg = self.validate_username(username)
        if not valid:
            return False, msg

        valid, msg = self.validate_email(email)
        if not valid:
            return False, msg

        valid, msg = self.validate_password(password)
        if not valid:
            return False, msg

        # 2. 业务规则检查
        if self.check_user_exists(username, email):
            return False, "用户名或邮箱已存在"

        # 3. 业务处理
        password_hash = self.hash_password(password)
        
        try:
            self.cursor.execute(
                "INSERT INTO users (username, email, password_hash) VALUES (?, ?, ?)",
                (username, email, password_hash)
            )
            self.conn.commit()
            return True, "注册成功"
        except sqlite3.IntegrityError:
            return False, "注册失败,用户可能已存在"
        except Exception as e:
            return False, f"系统错误: {str(e)}"

    def close(self):
        """关闭数据库连接"""
        self.conn.close()

# 使用示例
if __name__ == "__main__":
    # 创建注册系统实例
    reg_system = UserRegistrationSystem()
    
    # 测试用例
    test_cases = [
        ("user123", "user@example.com", "password123"),  # 有效
        ("ab", "user@example.com", "password123"),       # 用户名太短
        ("user123", "invalid-email", "password123"),     # 邮箱无效
        ("user123", "user@example.com", "123"),          # 密码太短
        ("user123", "user@example.com", "password123"),  # 重复注册
    ]
    
    for username, email, password in test_cases:
        print(f"\n尝试注册: {username}, {email}")
        success, message = reg_system.register_user(username, email, password)
        print(f"结果: {'成功' if success else '失败'} - {message}")
    
    # 清理
    reg_system.close()

这个实现展示了业务场景的几个关键部分:

  1. 数据验证层:对用户名、邮箱、密码进行格式检查
  2. 业务规则层:检查用户是否已存在(唯一性约束)
  3. 数据处理层:密码哈希化、数据库操作
  4. 错误处理:捕获并处理各种异常情况

代码结构清晰,每个方法职责单一,方便后续扩展。比如要添加手机号验证、发送验证邮件等功能,只需要在相应位置添加方法即可。

总结:业务场景实现要分层处理,先验证再执行业务规则最后处理数据。

因为 redis 中的数据是动态加入进去的,当那个 socket 没有查到他想要的数据的时候,就会过个两秒钟接着去 redis 中查数据,一直到查到为止

redis 不是有发布和订阅机制吗?
再不然使用消息中间件,订阅 topic 然后监听接收消息啊

redis 是 kv 结构,为啥是循环查找而不是按 key 取
10000 个 socket 连接是支持的

说实在,不知道具体业务。不知道 client 是面对用户还是内部接口,不知道请求频率,不知道数据结构,这种问题发出来也没什么人帮得了你

你只是想实现一个队列消息加服务端推送吧。。。

不用多线程的模型,用异步的方式
服务器这边维护一个{clientID->socket 句柄}的哈希表
然后启一个线程去循环查 redis 每个元素,元素的在上面哈希表中就用对应的 socket 句柄发送过去

0.0 客户端知道自己想要什么吧? 客户端想要的称之为 Key, 然后服务端可以根据 Key 去 redis 取? 取到的值叫 value ? 酱紫。。不需要循环吧?

我倒是很想知道什么样的人会设计 redis 用循环的方式读取数据, 莫不是来搞笑的

你这样是 client 主动拉取,为什么不主动推送。
不想自己写,可以用现成的 比如 MQTT 协议,ZMQ , T-IO 等开业项目。

非常感谢,发布和订阅机制能帮助我

10000 个 cient 也不过是 10000 个 list 而已
每个 client 去查找自己的 list 找到数据就返回 找不到就阻塞 直到 超时就好啊
你只要根据数据不同 分发到不同的 list 去就可以了

你这个 redis 没学好啊 redis 支持那么多种用法跟结构啊

感谢,对,可以理解是 client 主动拉取,但是不还是需要 client 主动过来建立链接,然后才能发送过去吗= =

可以,我愚笨了,感谢!

打错字了,是开源。
client 当然要连接 server 端了,但是 client 只接收消息,server 端 有数据的时候,主动发送到 client。
和其他人说的 发布订阅一个道理 。可以自己写,可以采用其他开源的项目。

好的,谢谢

如果没有就是要循环去查了哦= =,

redis 如果需要循环查找的话一定是你的使用方式哪里错了

回到顶部