如何优化基于SQLite的Telebot程序以支持多用户并发？

我开发了一个电商Telebot，使用SQLite数据库筛选数据。当两名及以上用户同时使用时，机器人出现卡顿、无法发送消息的问题，控制台还会重复输出用户ID。以下是实现代码：

import telebot
from telebot import types
import sqlite3

db = sqlite3.connect('database.db', check_same_thread = False)
sql = db.cursor()
sql.execute("""CREATE TABLE IF NOT EXISTS users(
    chatid TEXT,
    username TEXT,
    balance INT
    tovary TEXT
)""")
db.commit()

bot = telebot.TeleBot('Token')

mainmenu = types.ReplyKeyboardMarkup(resize_keyboard=True, row_width=2)
rules = types.KeyboardButton('Rules')
shop = types.KeyboardButton('Shop')
reviews = types.KeyboardButton('Reviews')
profile = types.KeyboardButton('Profile')
close = types.KeyboardButton('Close')
mainmenu.add(rules, shop, reviews, profile, close)

@bot.message_handler(commands=['start'])
def start(message):
    def start(message):
        greeting = f'<b>Test greeting</b>'
        bot.send_message(message.chat.id, f'{greeting}', parse_mode='html', reply_markup=mainmenu)
    sql.execute(f"SELECT chatid FROM users WHERE chatid = '{message.chat.id}'")
    if sql.fetchone() is None:
        sql.execute(f"INSERT INTO users VALUES(?,?,?)", (message.chat.id, message.from_user.username, 0))
        db.commit()
    else:
        for value in sql.execute("SELECT * FROM users"):
            print(value)

@bot.message_handler(content_types=['text'])
def handler(message):
    if message.text == "$money.test":
        sql.execute(f"UPDATE users SET balance = balance + 100 WHERE chatid = {message.chat.id}")
        db.commit()
        for value in sql.execute(f"SELECT balance FROM users WHERE chatid = {message.chat.id}"):
            bot.send_message(message.chat.id, f"Хорошо, я выдал тебе 100 гривен на баланс.

Твой баланс: {value[0]}", parse_mode='html')
    elif message.text == "$del.account":
        sql.execute(f"DELETE FROM users WHERE chatid = {message.chat.id}")
        db.commit()
        bot.send_message(message.chat.id, "Аккаунт удалён.")

bot.polling(none_stop=True)

核心优化步骤 #

1. 修复SQLite线程安全问题，避免共享游标 #

• 问题：全局共享单个游标+check_same_thread=False会导致多线程下操作冲突，引发数据混乱和卡顿。
• 优化：每次数据库操作创建独立连接和游标，操作完成后关闭，确保上下文隔离。
• 修改示例：

def get_db_connection():
    conn = sqlite3.connect('database.db')
    return conn

# 在start函数中使用
@bot.message_handler(commands=['start'])
def start(message):
    conn = get_db_connection()
    sql = conn.cursor()
    try:
        sql.execute("SELECT chatid FROM users WHERE chatid = ?", (message.chat.id,))
        if sql.fetchone() is None:
            sql.execute("INSERT INTO users (chatid, username, balance, tovary) VALUES(?,?,?,?)", 
                       (message.chat.id, message.from_user.username, 0, ""))
            conn.commit()
        else:
            sql.execute("SELECT * FROM users")
            for value in sql.fetchall():
                print(value)
    finally:
        conn.close()

2. 修正SQL语法错误与注入风险 #

• 问题：
  • 创建表语句中balance INT后缺少逗号，导致表结构创建失败；
  • 字符串拼接SQL语句存在注入风险，易引发语法错误。
• 优化：
  • 补全表结构语句的逗号；
  • 始终使用参数化查询（?占位符）替代字符串拼接。
• 修改示例：

def init_db():
    conn = get_db_connection()
    sql = conn.cursor()
    sql.execute("""CREATE TABLE IF NOT EXISTS users(
        chatid TEXT,
        username TEXT,
        balance INT,
        tovary TEXT
    )""")
    conn.commit()
    conn.close()

# 初始化数据库
init_db()

3. 移除冗余嵌套函数，优化消息逻辑 #

• 问题：start函数内部嵌套同名函数，导致欢迎消息无法发送；重复全表查询加重数据库负载。
• 优化：删除嵌套函数，直接发送欢迎消息；仅执行必要查询，避免无意义的全表扫描。
• 修改示例：

@bot.message_handler(commands=['start'])
def start(message):
    # 直接发送欢迎消息
    greeting = f'<b>Test greeting</b>'
    bot.send_message(message.chat.id, greeting, parse_mode='html', reply_markup=mainmenu)
    
    conn = get_db_connection()
    sql = conn.cursor()
    try:
        sql.execute("SELECT chatid FROM users WHERE chatid = ?", (message.chat.id,))
        if sql.fetchone() is None:
            sql.execute("INSERT INTO users (chatid, username, balance, tovary) VALUES(?,?,?,?)", 
                       (message.chat.id, message.from_user.username, 0, ""))
            conn.commit()
    finally:
        conn.close()

4. 简化数据查询逻辑，避免冗余循环 #

• 问题：查询用户余额时用循环遍历单条数据，浪费资源。
• 优化：使用fetchone()直接获取单条结果。
• 修改示例：

@bot.message_handler(content_types=['text'])
def handler(message):
    if message.text == "$money.test":
        conn = get_db_connection()
        sql = conn.cursor()
        try:
            sql.execute("UPDATE users SET balance = balance + 100 WHERE chatid = ?", (message.chat.id,))
            conn.commit()
            
            sql.execute("SELECT balance FROM users WHERE chatid = ?", (message.chat.id,))
            balance = sql.fetchone()[0]
            bot.send_message(message.chat.id, f"Хорошо, я выдал тебе 100 гривен на баланс.\n\nТвой баланс: {balance}", parse_mode='html')
        finally:
            conn.close()
    elif message.text == "$del.account":
        conn = get_db_connection()
        sql = conn.cursor()
        try:
            sql.execute("DELETE FROM users WHERE chatid = ?", (message.chat.id,))
            conn.commit()
            bot.send_message(message.chat.id, "Аккаунт удалён.")
        finally:
            conn.close()

5. 切换异步轮询或Webhook提升并发能力 #

• 问题：同步polling无法高效处理多用户并发请求，导致卡顿。
• 优化：使用异步轮询或Webhook替代长轮询，提升并发处理效率。
• 异步轮询示例（需安装异步版本pip install pytelegrambotapi[asyncio]）：

import asyncio

async def main():
    await bot.polling(none_stop=True)

if __name__ == "__main__":
    asyncio.run(main())

内容的提问来源于stack exchange，提问作者Данііл Єщенко