使用Python怎么實現一個Web靜態服務器非堵塞模式

使用Python怎么實現一個Web靜態服務器非堵塞模式？相信很多沒有經驗的人對此束手無策，為此本文總結了問題出現的原因和解決方法，通過這篇文章希望你能解決這個問題。

python主要應用領域有哪些

1、云計算，典型應用OpenStack。2、WEB前端開發，眾多大型網站均為Python開發。3.人工智能應用，基于大數據分析和深度學習而發展出來的人工智能本質上已經無法離開python。4、系統運維工程項目，自動化運維的標配就是python+Django/flask。5、金融理財分析，量化交易，金融分析。6、大數據分析。

單進程非堵塞 模型

#coding=utf-8
from socket import *
import time
# 用來存儲所有的新鏈接的socket
g_socket_list = list()
def main():
  server_socket = socket(AF_INET, SOCK_STREAM)
  server_socket.setsockopt(SOL_SOCKET, SO_REUSEADDR , 1)
  server_socket.bind(('', 7890))
  server_socket.listen(128)
  # 將套接字設置為非堵塞
  # 設置為非堵塞后，如果accept時，恰巧沒有客戶端connect，那么accept會
  # 產生一個異常，所以需要try來進行處理
  server_socket.setblocking(False)
  while True:
    # 用來測試
    time.sleep(0.5)
    try:
      newClientInfo = server_socket.accept()
    except Exception as result:
      pass
    else:
      print("一個新的客戶端到來:%s" % str(newClientInfo))
      newClientInfo[0].setblocking(False) # 設置為非堵塞
      g_socket_list.append(newClientInfo)
    for client_socket, client_addr in g_socket_list:
      try:
        recvData = client_socket.recv(1024)
        if recvData:
          print('recv[%s]:%s' % (str(client_addr), recvData))
        else:
          print('[%s]客戶端已經關閉' % str(client_addr))
          client_socket.close()
          g_socket_list.remove((client_socket,client_addr))
      except Exception as result:
        pass
    print(g_socket_list) # for test
if __name__ == '__main__':
  main()

web靜態服務器-單進程非堵塞

import time
import socket
import sys
import re
class WSGIServer(object):
  """定義一個WSGI服務器的類"""
  def __init__(self, port, documents_root):
    # 1. 創建套接字
    self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # 2. 綁定本地信息
    self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    self.server_socket.bind(("", port))
    # 3. 變為監聽套接字
    self.server_socket.listen(128)
    self.server_socket.setblocking(False)
    self.client_socket_list = list()
    self.documents_root = documents_root
  def run_forever(self):
    """運行服務器"""
    # 等待對方鏈接
    while True:
      # time.sleep(0.5) # for test
      try:
        new_socket, new_addr = self.server_socket.accept()
      except Exception as ret:
        print("-----1----", ret) # for test
      else:
        new_socket.setblocking(False)
        self.client_socket_list.append(new_socket)
      for client_socket in self.client_socket_list:
        try:
          request = client_socket.recv(1024).decode('utf-8')
        except Exception as ret:
          print("------2----", ret) # for test
        else:
          if request:
            self.deal_with_request(request, client_socket)
          else:
            client_socket.close()
            self.client_socket_list.remove(client_socket)
      print(self.client_socket_list)
  def deal_with_request(self, request, client_socket):
    """為這個瀏覽器服務器"""
    if not request:
      return
    request_lines = request.splitlines()
    for i, line in enumerate(request_lines):
      print(i, line)
    # 提取請求的文件(index.html)
    # GET /a/b/c/d/e/index.html HTTP/1.1
    ret = re.match(r"([^/]*)([^ ]+)", request_lines[0])
    if ret:
      print("正則提取數據:", ret.group(1))
      print("正則提取數據:", ret.group(2))
      file_name = ret.group(2)
      if file_name == "/":
        file_name = "/index.html"
    # 讀取文件數據
    try:
      f = open(self.documents_root+file_name, "rb")
    except:
      response_body = "file not found, 請輸入正確的url"
      response_header = "HTTP/1.1 404 not found\r\n"
      response_header += "Content-Type: text/html; charset=utf-8\r\n"
      response_header += "Content-Length: %d\r\n" % (len(response_body))
      response_header += "\r\n"
      # 將header返回給瀏覽器
      client_socket.send(response_header.encode('utf-8'))
      # 將body返回給瀏覽器
      client_socket.send(response_body.encode("utf-8"))
    else:
      content = f.read()
      f.close()
      response_body = content
      response_header = "HTTP/1.1 200 OK\r\n"
      response_header += "Content-Length: %d\r\n" % (len(response_body))
      response_header += "\r\n"
      # 將header返回給瀏覽器
      client_socket.send( response_header.encode('utf-8') + response_body)
# 設置服務器服務靜態資源時的路徑
DOCUMENTS_ROOT = "./html"
def main():
  """控制web服務器整體"""
  # python3 xxxx.py 7890
  if len(sys.argv) == 2:
    port = sys.argv[1]
    if port.isdigit():
      port = int(port)
  else:
    print("運行方式如: python3 xxx.py 7890")
    return
  print("http服務器使用的port:%s" % port)
  http_server = WSGIServer(port, DOCUMENTS_ROOT)
  http_server.run_forever()
if __name__ == "__main__":
  main()

看完上述內容，你們掌握使用Python怎么實現一個Web靜態服務器非堵塞模式的方法了嗎？如果還想學到更多技能或想了解更多相關內容，歡迎關注億速云行業資訊頻道，感謝各位的閱讀！