SoFunction
Updated on 2024-11-12

Example of using the python Tornado framework

Tornado is a python open source web framework, it is much lighter than django, there is no component, only to apply to the corresponding business scenarios I use this framework, it is a single-process single-threaded to asynchronous non-blocking model, applicable to long connections and long rounds, high concurrency, asynchronous non-blocking

Installation:

pip install tornado

View Layer

'''
@File    : views_service.py
@Copyright : rainbol
@Date    : 2020/8/31
@Desc    :
'''
import threading
import time
import 
import tornado
import 
import 
import 
from  import run_on_executor
from  import ThreadPoolExecutor
from uuid import uuid4
import random

all_count = 0
big_list = {}


class ServiceHandler():
  executor = ThreadPoolExecutor(20) # max_threads An executor property must be defined before the run_on_executor decorator is useful.

  @run_on_executor # Under this method, run in-thread; the query function is wrapped by run_on_executor (syntactic sugar), which passes the execution of this function to the thread pool executor's thread for execution, optimized to handle time-consuming tasks such that they don't block the main thread.
  def time_demo(self, tid, uid):
    (tid)
    threading_id = threading.current_thread().ident
    big_list[uid] = threading_id

  @ # Asynchronous, concurrent processing; increased concurrency
  def post(self):
    global all_count
    all_count += 1
    uid = str(uuid4())
    yield self.time_demo((1, 100), uid) # Simulate business processing, using yield to achieve asynchronous blocking request
    r = {'status': 'True', 'thread id': '%s' % big_list[uid], "count": all_count}

    (.json_encode(r)) # Write return message to write response
    () # End of service

  def get(self):
    return ()

__init__.py

'''
@File    : __init__.py
@Copyright : rainbol
@Date    : 2020/8/31
@Desc    :
'''
import  # web framework
import  # http services
import  # Input-output event loop
import  # Configuration tools
from  import options, define
from  import configs
from  import urls
define('port', default=8000, type=int, help='Run port')


# Customized applications
class CustomApplication():
  def __init__(self): # Rewrite constructor methods
    # Specify routing rules
    handlers = urls
    # Specify the configuration file
    settings = configs
    super(CustomApplication, self).__init__(handlers=handlers, **settings)


# Define services
def create_server():
  # Allowed to be launched from the command line
  #.parse_command_line()
  # Create http service
  http_server = (
    CustomApplication() # Take care to instantiate
  )
  # Bind the listening port
  http_server.listen()
  # Start the input-output event loop
  ().start()
'''
@File    : 
@Copyright : rainbol
@Date    : 2020/8/31
@Desc    :
'''
from  import create_server



if __name__ == '__main__':
  create_server()

routing (in computer networks)

from .views_index import IndexHandler as index
from .views_service import ServiceHandler as service

# Configure routing and configure to mapping rules

urls = [
  (r"/index", index),
  (r"/demo", service),
]

Above is the use of python Tornado framework example of the details, more information about python Tornado framework please pay attention to my other related articles!