Server Queue 提高 QPS

#!/usr/bin/env python
# encoding: utf-8
# 猴子补丁
from gevent import monkey
monkey.patch_all()
import json
import logging
from flask import Flask
from flask import request
from gevent import pywsgi
from kafka import KafkaProducer
app = Flask(__name__)
# kafka 生产者 配置
topic = "test"
host = ["127.0.0.1:8009"]
produc = KafkaProducer(hosts=host)
logging.basicConfig(filename=__name__, filemode="w", format="%(asctime)s-%(name)s-%(levelname)s-%(message)s",
                    level=logging.INFO)
logger = logging.getLogger(__name__)
@app.route("/api/v1/show/", methods=["GET"])
def api_v1_show():
    try:
        data = json.dumps(request.args.to_dict())
        logger.info("data")
        # 因为 Kafka 使用的是，异步批量写入 Kafka 中，所以需要
        produc.send(topic=topic, value=data)
    except Exception as e:
        logger.error("/api/v1/show/ err")
        pass
    return "succ"
if __name__ == '__main__':
    # 使用 pywsgi.WSGIServer 异步 方式提高 性能
    server = pywsgi.WSGIServer(('', 8015), app)
    server.serve_forever()

#!/usr/bin/env python
# encoding: utf-8
# 猴子补丁
from gevent import monkey
monkey.patch_all()
import json
import gevent
import logging
from flask import Flask
from flask import request
from gevent import pywsgi
from kafka import KafkaProducer
from gevent.queue import JoinableQueue
app = Flask(__name__)
REQ_Q = JoinableQueue(100000)  # 中间数据缓存
# kafka 生产者 配置
topic = "test"
host = ["127.0.0.1:8009"]
produc = KafkaProducer(hosts=host)
logging.basicConfig(filename=__name__, filemode="w", format="%(asctime)s-%(name)s-%(levelname)s-%(message)s",
                    level=logging.INFO)
logger = logging.getLogger(__name__)
def main_worker():
    while True:
        try:
            item = REQ_Q.get()
            produc.send(item)
        except Exception as e:
            logger.error(f"main_work err {e}")
            print(e)
        finally:
            REQ_Q.task_done()
@app.route("/api/v1/show/", methods=["GET"])
def api_v1_show():
    try:
        # 因为 Kafka 使用的是，异步批量写入 Kafka 中，所以需要
        data = json.dumps(request.args.to_dict())
        logger.info("data")
        REQ_Q.put(data)
    except Exception as e:
        logger.error("/api/v1/show/ err")
        pass
    return "succ"
def init_workers(size):
    for i in range(size):
        gevent.spawn(main_worker)
if __name__ == '__main__':
    # 异步轮训消费
    init_workers(50)
    # 使用 pywsgi.WSGIServer 异步 方式提高 性能
    server = pywsgi.WSGIServer(('', 8015), app)
    server.serve_forever()