非常简易的一个测试程序
# -*- coding: UTF-8 -*-
from flask import Flask, request
import json
import time
import re
import sys
import asyncio
from tornado.ioloop import IOLoop
from tornado.wsgi import WSGIContainer
from tornado.httpserver import HTTPServer
import sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine('sqlite:///database.db?check_same_thread=False')
Base = declarative_base()
class ClientLog(Base):
__tablename__ = 'clientlog'
log_id = Column(Integer, primary_key=True, autoincrement=True)
message = Column(String(512))
def __repr__(self):
return "A"
from sqlalchemy.ext.declarative import DeclarativeMeta
class AlchemyEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj.__class__, DeclarativeMeta):
# an SQLAlchemy class
fields = {}
for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']:
data = obj.__getattribute__(field)
try:
json.dumps(data) # this will fail on non-encodable values, like other classes
fields[field] = data
except TypeError:
fields[field] = None
# a json-encodable dict
return fields
return json.JSONEncoder.default(self, obj)
Base.metadata.create_all(engine, checkfirst=True)
Session = sessionmaker(bind=engine)
session = Session()
app = Flask("app")
@app.route('/api', methods=['GET', 'POST'])
@app.route('/api/', methods=['GET', 'POST'])
def api1():
session.add(ClientLog(message="test"))
session.commit()
return "test"
def launch_server():
if sys.platform == 'win32':
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
http_server = HTTPServer(WSGIContainer(app))
http_server.listen(8080)
IOLoop.current().start()
launch_server()
在阿里云的 Windows 2019 模板下 100 并发就会崩溃退出,报错
Exception in thread Tornado selector:
Traceback (most recent call last):
File "C:\Users\Administrator\AppData\Local\Programs\Python\Python311\Lib\threading.py", line 1045, in _bootstrap_inner
self.run()
File "C:\Users\Administrator\AppData\Local\Programs\Python\Python311\Lib\threading.py", line 982, in run
self._target(*self._args, **self._kwargs)
File "C:\Users\Administrator\AppData\Local\Programs\Python\Python311\Lib\site-packages\tornado\platform\asyncio.py", line 574, in _run_select
rs, ws, xs = select.select(to_read, to_write, to_write)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
ValueError: too many file descriptors in select()
但是我在 Windows 10 的电脑上测试并发没问题 之前测试换了另外一个对 Windows 更友好的 web framework (想不起来名字,找了半天没找到)可以是可以,但是估计它用了不止一个进程写 SQLite ,导致冲突无法提交更改 业务需求调用一个 Windows Only 的 pip 包 ,无法更换 Linux 服务器
这是一个专为移动设备优化的页面(即为了让你能够在 Google 搜索结果里秒开这个页面),如果你希望参与 V2EX 社区的讨论,你可以继续到 V2EX 上打开本讨论主题的完整版本。
V2EX 是创意工作者们的社区,是一个分享自己正在做的有趣事物、交流想法,可以遇见新朋友甚至新机会的地方。
V2EX is a community of developers, designers and creative people.