【django】定制 django - 运行你的独立的不死线程_django中运行两个独立线程_FeatureOverload的博客

2019/May/14 更新 solution

最近开始了解 flask 框架，手写了一点点 flask 代码。用上了 “flask-script” 库之后，flask 的启动方式就极其像开发时的 django。
也发现其监听文件改动重启服务器的功能，这让我原本在 flask 上运行的一个不死线程启动了两次，于是又使用了本文的 solution。实际上本文原先的 solution 本身也是非常好用的，非常有效。
不过在我多写了一些后台代码之后，多理解了一些细节之后，我觉得可以使用现在的新版 solution，或许可能更简单一些，更少一些 trick，更直接一点点。
同样以 django 框架为例。
注，如果是上线的代码/后台服务，是不会启动两个进程的（一个“manage”，一个做服务器），可以根据本次更新的思路很容易就能启动一个后台线程。
django-admin startproject <project name> 后的原始 manage.py 代码：
#!/usr/bin/env python
"""Django's command-line utility for administrative tasks."""
import os
import sys
def main():
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'undeadthread.settings')
    try:
        from django.core.management import execute_from_command_line
    except ImportError as exc:
        raise ImportError(
            "Couldn't import Django. Are you sure it's installed and "
            "available on your PYTHONPATH environment variable? Did you "
            "forget to activate a virtual environment?"
        ) from exc
    execute_from_command_line(sys.argv)
if __name__ == '__main__':
    main()
因为在开发时是通过执行 python manage.py runserver，即 manage.py 是入口处，所以在 manage.py 文件中增加一点点代码： 
#!/usr/bin/env python
"""Django's command-line utility for administrative tasks."""
import os
import sys
import requests
from time import sleep
def enable_undeadthread():
    cnt = 10
    while cnt:
        try:
            requests.get("http://localhost:8000/undeadthread", )
        except Exception:
            cnt -= 1
            sleep(1)
        else:
            break
def main():
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'undeadthread.settings')
    try:
        from django.core.management import execute_from_command_line
    except ImportError as exc:
        raise ImportError(
            "...略..."
        ) from exc
    import threading
    threading.Thread(target=enable_undeadthread, daemon=True).start()
    execute_from_command_line(sys.argv)
if __name__ == '__main__':
    main()
 可以看到，这里需要 requests 库，使用 python 标准库的 urllib 之类的也可以实现相同的作用，但是使用 requests 库代码可以更简单一些。 
上面请求的 url 就是这个 manage.py runserver 启动的 server（你可以根据我旧版 solution 另一篇博客 ? 【django】（定制 django）启动服务器时调用默认浏览器打开项目首页 解析一下命令行，这样就能控制 domain 和 端口了） 
接下来添加一下请求的 url 中的 /undeadthread 部分（在最顶层 urls.py 中添加）： 
from django.contrib import admin
from django.urls import path
import sys
from django.http import HttpResponse
ONE_UNDEAD_THREAD_FLAG = False
def enable_undeadthread(request):
    import threading
    threading.Thread(target=undeadthread, daemon=True).start()
    return HttpResponse("it works")
def undeadthread():
    global ONE_UNDEAD_THREAD_FLAG
    from time import sleep, ctime
    if not ONE_UNDEAD_THREAD_FLAG:
        ONE_UNDEAD_THREAD_FLAG = True
    else:
        return
    while True:
        print(ctime(), file=sys.stderr)
        sleep(2)
urlpatterns = [
    path('admin/', admin.site.urls),
urlpatterns += [
    path('undeadthread', enable_undeadthread),
可以看到，def undeadthread() 就函数就是会从服务器从开始到结束一直运行的线程。 
 当然，除非它自己内部出错又没有处理，提前挂掉。 
在 undeadthread 函数内部的 while True 编写定制的代码即可。 
这次更新的原因主要是对“异步”理解稍微更深一些， 
所以想到请求一个 url 时，服务器内部开始运行一个函数（在 django 框架中写代码，这个函数一般在 views.py 中，这里为了简单起见直接写在 urls.py 中），然后函数可以启动任意的线程或进程，至于启动的线程或进程退不退出其实和这个处理 request 的函数已经没有关系了（除非它等待线程或进程结束，不过那样就不是异步了，最多是并发或着并行）。所以，该函数的任务可以是启动一个线程在后台运行，然后自己响应（response） request，即退出。 
但是被启动线程只要还没有 returrn 掉（和没有抛出异常），那么该线程就会继续运行。 
=== END 更新 
==== 旧版 solution 分割线 === 
说在前面的话
 
通过《轻量级 django》应该也可以定制 django，而且可能更“灵活”。
 并且可能会比本篇介绍的方法更加“健壮”。 
不过本篇的方法比较“简单”一点，并且暂时不用像《轻量级 django》中对 django 的框架机制了解地那么清楚。因此本篇还是值得一看的。 
Q: 实话说，你们为什么要如此热衷于使用 django，即使在它上面跑个 while 循环都不太那么“明白”地实现情况下，还要对它一改再改？ 
A: 因为 django 的 ORM 很棒呀！！！ 
判断当前进程是否是子进程
 
启动 django （manage.py runserver）最后会有两个进程在主机中运行。 
父进程
 父进程一个在开发阶段最重要的功能就是 “监听文件改动，如果有文件改动，则重启子进程”；当然，django 父进程的功能还有其它用途，但是本人没有对其进一步研究，所以就不多言了。 
子进程
 子进程提供 HTTP server 功能。 
我们将需要在 django 中使用的线程放在子进程中，因为这样可以和子进程共享内存空间（变量等）。 
所以需要判断当前进程是否为子进程，以用来启动我们的定制线程，在父进程中启动定制线程一般都没有意义。
 （当然，如果你需要在父进程中启动，那就做相反的操作） 
仅 LINUX 平台可用的 solution
 
(MAC OS 平台未测试) 
def Is_child_processing():
    import re
    re_result = re.findall(
        "{}".format(os.getpid()),
        os.popen("ps -ejf | grep {} | grep -v \"grep\"".format(os.getpid())).read()
    ret = True if len(re_result) == 1 else False
    return ret
在 Terminal 中输入 $ ps -ejf | grep {} | grep -v "grep" 查看输出，了解这段代码是如何实现的原理。 
 感觉这段代码应该没有必要做解释 ?。 
通用平台的 solution
 
def Is_child_processing():
    from multiprocessing.connection import Listener
    from queue import Queue
    q = Queue()
    def lock_system_port(_port):
        nonlocal q  # it's OK without this announce line 
        try:
            listener = Listener(("", _port))
            q.put(False)
        except Exception:  # port be used by parent
            # traceback.print_exc()
            q.put(True)
            return  # child don't listen
        while True:
            serv = listener.accept()  # just bind the port.
    t = threading.Thread(target=lock_system_port, args=(62771, ))
    t.setDaemon(True)
    t.start(); del t;
    return q.get()
当然，正如你所见到的，这种 solution 会占用一个系统端口。 
这个 solution 的机制就是通过占用一个系统端口，因为端口只能被一个进程占用（当然，一个进程内也不能重复打开端口），所以子进程运行时，会执行同样的代码，但是因为端口已经被父进程占用了，于是就可以判断当前进程是子进程。 
当然，这段代码还有不少线程的设计细节，希望你能明白我为什么要这么设计这段代码。 
 线程共享可以访问到的变量，但是线程之间不能直接通信；
并且父进程中的 lock_system_port 函数不会退出，知道 django server 停止运行，所以不能简单地通过获取这个函数的返回值来判断是否是子进程。 
  当然，我说的是不能“简单地”，通过一些复杂的超时机制设计，理论上总是可以实现这种方式的。 
 基于以上两点，使用一个 Queue 来悉知 lock_system_port 函数（该函数被用作独立的线程运行）指示的是否是子进程的状态是必要的。
 3. line 18, 19 lock_system_port 函数在父进程中使用 while True 死循环运行是必要的，否则该线程退出，端口就没有被绑定，子进程运行到这段代码，就会拿到这个端口，整个功能就没有起到作用。
 4. t.setDaemon(True) 是必要的，不是说这一行代码是必要写的，而是 “Daemon” 线程这个状态为 True 是必要的。否则 Ctrl + C 这类的方式可能无法（一次性）完全退出 server。这是因为多线程程序中，系统会等待所有“非 Daemon”线程退出之后，该进程才真正退出- 释放资源。而 “Daemon” 线程被视作“退出状态”是无关紧要的，所以当所有的 “非 Daemon” 线程退出之后，既是还有 “Daemon” 线程 正在运行，该进程也会退出。 
Note - 本段非常重要，避免 solution 的使用陷阱 
并不是所有一个程序有多个进程的地方都能使用这个 solution，因为父进程和子进程的启动顺序是不确定的，并不是说父进程一定比子进程先运行某一段共同的代码。
 但是在 django 的 case 中，这段代码我们总是能预期父进程先运行到的。因此可以使用这个 solution。 
直接放代码
 
默认 startproject 创建的 manage.py look like 
#!/usr/bin/env python
import os
import sys
if __name__ == '__main__':
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'web.settings')
    try:
        from django.core.management import execute_from_command_line
    except ImportError as exc:
        raise ImportError(
            "Couldn't import Django. Are you sure it's installed and "
            "available on your PYTHONPATH environment variable? Did you "
            "forget to activate a virtual environment?"
        ) from exc
    execute_from_command_line(sys.argv)
加入运行定制的独立运行的线程
 
 不死线程，会死的线程就没啥意义了，一个线程启动，运行，退出；直接在 views.py 加一个 function 一样能够实现这个功能。 
#!/usr/bin/env python
import os
import sys
def customization():
    import threading
    t = threading.Thread(target=_run)
    t.setDaemon(True)
    t.start()
    del t
def _run():
    from time import sleep
    from time import ctime
    while True:  # loop until universe collapses
        print("{} customization > _run is running".format(ctime()), file=sys.stderr)
        sleep(5)
def Is_child_processing():
    [...这个位置的代码参见上文，选择你喜欢的 solution...]
if __name__ == '__main__':
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'web.settings')
    try:
        from django.core.management import execute_from_command_line
    except ImportError as exc:
        raise ImportError(
            "Couldn't import Django. Are you sure it's installed and "
            "available on your PYTHONPATH environment variable? Did you "
            "forget to activate a virtual environment?"
        ) from exc
    if Is_child_processing():
        customization()
    execute_from_command_line(sys.argv)
打印输出混在一起？
 需要一个单独的输出？ 
为你献上一版更新，使用 logging module
 
#!/usr/bin/env python
import os
import sys
def customization():
    import threading
    t = threading.Thread(target=_run)
    t.setDaemon(True)
    t.start()
    del t
def _run():
    from time import sleep
    from time import ctime
    # logging
    import logging
    logging.basicConfig(
        filename=os.path.join(os.getcwd(), "customization_run.log"),
        level=logging.INFO,
        # level=logging.WARNING,
        format='[%(asctime)s]%(levelname)-9s%(message)s',
    while True:  # loop until universe collapses
        logging.info("customization > _run is running")
        sleep(5)
def Is_child_processing():
    [...这个位置的代码参见上文，选择你喜欢的 solution...]
if __name__ == '__main__':
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'web.settings')
    try:
        from django.core.management import execute_from_command_line
    except ImportError as exc:
        raise ImportError(
            "Couldn't import Django. Are you sure it's installed and "
            "available on your PYTHONPATH environment variable? Did you "
            "forget to activate a virtual environment?"
        ) from exc
    if Is_child_processing():
        customization()
    execute_from_command_line(sys.argv)
简单的运行说明： 
#### terminal 1 ###
<path>/<to>/<django>/<proj>/<dir> $ python manage.py runserver
...django information output...
Ctrl+C to quit
? 这样就和平常启动 django 无异。 
现在查看 customization 的 不死线程输出： 
#### terminal 2 ####
<path>/<to>/<django>/<proj>/<dir> $ tail -f customization_run.log
.........the log ........
you can use Ctrl+C to quit "tail -f"
                    【django】定制 django - 运行你的独立的不死线程 【django】定制 django - 运行你的独立的不死线程说在前面的话直接放代码说在前面的话通过《轻量级 django》应该也可以定制 django，而且可能更“灵活”。并且可能会比本篇介绍的方法更加“健壮”。不过本篇的方法比较“简单”一点，并且暂时不用像《轻量级 django》中对 django 的框架机制了解地那么清...
				这篇文章主要介绍了Django异步任务线程池实现原理,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
当数据库数据量很大时（百万级），许多批量数据修改请求的响应会非常慢，一些不需要即时响应的任务可以放到后台的异步线程中完成，发起异步任务的请求就可以立即响应
选择用线程池的原因是：线程比进程更为可控。不像子进程，子线程会在所属进程结束时立即结束。线程可共享内存。
请求任务异步处理的原理
使用python manage.py runserver模式启动的Django应用只有一个进程，对于每个请求，主线程会开启一个子线程来处理请求。请求子线程向主线




    

				django 的并发能力真的是令人担忧，django本身框架下只有一个线程在处理请求，任何一个请求阻塞，就会影响另一个情感求的响应，尤其是涉及到IO操作时，基于框架下开发的视图的响应并没有对应的开启多线程，再者Python的多线程对于多核CPU有效利用率非常低，参照
这里就使用 nginx + uwsgi 提供高并发
nginx 的并发能力超高，单台并发能力过万（这个也不是绝对），在纯静态的 ...
django 的并发能力真的是令人担忧，这里就使用 nginx + uwsgi 提供高并发
nginx 的并发能力超高，单台并发能力过万（这个也不是绝对），在纯静态的 web 服务中更是突出其优越的地方，由于其底层使用 epoll 异步IO模型进行处理，使其深受欢迎
做过运维的应该都知道，php 需要使用 nginx + fastcg...
				Django的ORM是非常好用的，哪怕不是做Web项目也值得一用，所以网上也可以找到不少使用 Django 开发非Web项目的资料，因为除了ORM之个，命令行、配置文件等组件也非常好用。
最近用这种方式开发了一个非Web项目，而且是多线程的。有N个工作线程从DB中获取jobs，并把结果写回DB。简单来说就是这样。
项目运行一段时间后，发现数据库连接耗尽了，幸好内存大，然后一直往上调，最后连接数都上九千多一万了。耗尽连接数的时候，PostgreSQL 会出现类似这样的错误：
FATAL: remaining connection slots are reserved for non-repli
				文章目录Django 多线程#1 环境#2 Django原生单线程#3 Django server 默认多线程#3.1 测试#4 使用 uWSGI 服务器启动 django#4.1 启动 uWSGI#4.2 给uWSGI加进程数#5 总结关于django的多线程问题
Django 多线程
#1 环境
Python3.7.3
Django==2.0.7
uWSGI==2.0.18
#2 Djang...
				当数据库数据量很大时（百万级），许多批量数据修改请求的响应会非常慢，一些不需要即时响应的任务可以放到后台的异步线程中完成，发起异步任务的请求就可以立即响应 
选择用线程池的原因是：线程比进程更为可控。不像子进程，子线程会在所属进程结束时立即结束。线程可共享内存。 
请求任务异步处理的原理 
使用python manage.py runserver模式启动的Django应用只有一个进程，对于每个请求...
				preload 为True的情况下，会将辅助线程或者进程开在master里，加重master的负担（master最好只是用来负责监听worker进程）
django应用的gunicorn示例：只在主线程里开启后台线程，worker里不启动后台线程
gunicorn -w 5 --preload -b 127.0.0.1:8088 application_name.wsg...
				在 Django 中，您可以使用 Python 的 `concurrent.futures` 模块来实现接口并行。这个模块提供了一些工具来简化多线程和多进程编程。
下面是一个简单的例子，展示了如何使用 `concurrent.futures` 模块来实现接口并行：
```python
import concurrent.futures
import requests
# 定义要并行调用的接口列表
api_urls = [
    'http://example.com/api1',
    'http://example.com/api2',
    'http://example.com/api3',
# 定义一个函数来调用单个接口
def call_api(url):
    response = requests.get(url)
    return response.json()
# 使用 concurrent.futures.ThreadPoolExecutor 来创建线程池
with concurrent.futures.ThreadPoolExecutor() as executor:
    # 使用 map() 方法来并行调用接口
    results = executor.map(call_api, api_urls)
# 处理调用的结果
for result in results:
    print(result)
在这个例子中，我们使用了 `concurrent.futures.ThreadPoolExecutor` 来创建线程池，然后使用 `map()` 方法来并行调用接口。您也可以使用 `concurrent.futures.ProcessPoolExecutor` 来创建进程池，并行调用接口。
希望这个回答对您有帮助。
				【Docker】【MySQL】(Ubuntu) Access denied for user '<user>'@'172.17.0.1'(using password: YES)