Flask 实现异步服务

Posted 2023-03-02

参考技术A Flask的服务，默认是同步的，在接收多个请求是会发生阻塞的，导致打开页面变的很慢，很卡，如下一个flask服务：

# -*- coding: utf-8 -*-

import sys

from seleniumimport webdriver

from  bs4import BeautifulSoup

import time

import json

from pymongoimport MongoClient

from zhimaipimport getdailione

from flaskimport Flask, jsonify, request

#创建一个Flask对象

app = Flask(__name__)

def browserini():

global driver

# 启动浏览器

# ip, exttime = getdailione(3)  #调用芝麻IP 3-6小时

    ip ='47.96.225.239:4111'

    while ip ==0:

# ip, exttime = getdailione(3)

        ip ='47.96.225.239:4111'

        time.sleep(3)

chrome_options = webdriver.ChromeOptions()

chrome_options .add_argument('--headless')

chrome_options.add_argument('--disable-gpu')

chrome_options.add_argument('--proxy-server=' + ip)

driver = webdriver.Chrome(chrome_options=chrome_options)# 加载浏览器驱动

# driver.set_window_size(1920, 1080)

    url ='https://www.qichacha.com/'

#flask服务，供单点查询调用

@app.route('/api/search/baseinfo/',methods=['POST'])

def baseinfo():

url ='https://www.qichacha.com/'

    driver.get(url)

search_name = request.form.get('searchname')# 获取表单请求参数

    if(len(search_name)<5):

return json.dumps('error_message':'sorry, please enter the correct company name!')

else:

print ('query company_name is:',search_name)

element = driver.find_element_by_id('searchkey')#搜索框

        time.sleep(3)

element.send_keys(search_name)#输入搜索公司名称

        clickbutton = driver.find_element_by_id('V3_Search_bt')#搜索按钮

        clickbutton.click()

#进入搜索公司列表，点击公司链接

        soup = BeautifulSoup(driver.page_source,'lxml')

num = soup.find('span', 'id':'countOld').find('span', 'class':'text-danger').text.strip()# 搜寻到该公司数量列表

        if  '0'==num:

print u'抱歉！查询不到该公司信息,请确认后重新获取!'

            return json.dumps('message':u'抱歉！查询不到该公司信息,请确认后重新获取!')

else:

hrefbutton = driver.find_element_by_class_name('ma_h1')#找到第一个公司链接点击

            hrefbutton.click()

handles = driver.window_handles#当前所有句柄

            print handles

first_handles = handles[0]

driver.switch_to.window(handles[1])#将浏览器驱动跳转到当前窗口

#开始解析页面获取基本信息

            soup = BeautifulSoup(driver.page_source)

company_name = soup.find('div', 'id':'company-top').find('div','class' :'content').find('div', 'class':'row title').text.strip()#公司名称

#print company_name, type(company_name)

            company_name = company_name.split(' ')[0].split('\n')[0]

#基础信息标签

            baselist = soup.find('section', 'id':'Cominfo').find_all('table', 'class':'ntable')[1]#

            tr_list = baselist.find_all('tr')

# 插入基本信息数据表

            data= json.dumps(

# 注册资本

                'register_capital': tr_list[0].find_all('td')[1].text.strip(),

# 实缴资本

                'real_capital': tr_list[0].find_all('td')[3].text.strip(),

# 经营状态

                'operate_state': tr_list[1].find_all('td')[1].text.strip(),

# 公司名称     

'company_name' : company_name,

# 爬取时间

                'crawl_time': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))

)

# 关闭新窗口

    driver.close()

driver.switch_to.window(first_handles)# 将浏览器驱动跳转到当前窗口

    return data

if __name__ =='__main__':

try:

# 启动浏览器

# ip, exttime = getdailione(3)  #调用芝麻IP 3-6小时

        ip ='47.96.225.239:4111'

        while ip ==0:

ip, exttime = getdailione(1)

#ip = '47.96.225.239:4111'

            time.sleep(3)

chrome_options = webdriver.ChromeOptions()

chrome_options .add_argument('--headless')

chrome_options.add_argument('--disable-gpu')

chrome_options.add_argument('--proxy-server=' + ip)

driver = webdriver.Chrome(chrome_options=chrome_options)# 加载浏览器驱动

#driver.set_window_size(1920, 1080)

        time.sleep(5)

except:

browserini()

app.run(host='0.0.0.0',port =5008,debug=True,use_reloader=False)

比较简单的实现异步的方式就是借助第三方库：gevent

# -*- coding: utf-8 -*-

from gevent import monkey

from gevent import pywsgi 或者 from gevent.pywsgi import WSGIServer

from flask import Flask

import requests

app = Flask(__name__)

@app.route('/')

def index():

        #具体的处理逻辑

http_server = pywsgi.WSGIServer(('127.0.0.1', 5000), app)

http_server.serve_forever()