1. 前言

1.1. 概述

基于Flask、Pika、Multiprocessing、Thread搭建一个架构，完成多线程、多进程工作。具体需求如下：

1. 并行计算任务：使用multiprocessing模块实现并行计算任务，提高计算效率、计算能力。
2. 消息侦听任务：使用threading模块完成RabbitMQ消息队列的侦听任务，将接收到的数据放入multiprocessing.Queue中，以便并行计算任务处理。
3. Web服务：使用Flask框架实现Web API服务，提供启停消息侦听任务、启停并行计算任务以及动态调整参数的功能。
4. 任务交互：通过multiprocessing.Queue实现消息侦听任务与并行计算任务之间的资源交互。
5. 非阻塞运行：使用threading模块非阻塞地运行Flask Web服务。

1.2. 续上一篇《python多线程与多进程开发实践及填坑记（1）》

python多线程与多进程开发实践及填坑记（1）

1.3. 重新启动侦听线程失败

程序没有报错，但是，没有启动侦听服务线程。

@app.route('/startlistening', methods=['GET'])
def start_listening():
    if stop_event.is_set():
        stop_event.clear()
    if not hasattr(app, 'pika_thread') or not app.pika_thread.is_alive():
        app.pika_thread = Thread(target=consume_from_rabbitmq_and_enqueue, args=(rabbitmq_params, rabbitmq_queue, data_queue))
        app.pika_thread.start()
        return jsonify({'status': 'listening'}), 200
    else:
        return jsonify({'status': 'already listening'}), 200

2. 原理解析

多线程下的Event（事件）原理与使用是并发编程中的重要概念，它主要用于线程间的同步与通信。下面我将从原理、类型、状态以及使用方法等方面详细介绍Event。

2.1. Event原理

Event（事件）是一个内核对象，用于线程间的同步与通信。它管理着一个全局的内部标志（Flag），这个标志通常有两种状态：已设置（True）和未设置（False）。线程可以通过Event对象来等待某个事件的发生，或者通知其他线程某个事件已经发生。

2.2. Event类型

Event主要分为两种类型：

• 自动重置事件：当事件被触发（即Flag被设置为True）时，会唤醒一个正在等待的线程。随后，事件的Flag会自动重置为False，直到再次被触发。
• 手动重置事件：当事件被触发时，会唤醒所有正在等待的线程。事件的Flag保持为True，直到程序显式地将其重置为False。

2.3. Event状态

Event对象的状态主要由其内部标志（Flag）决定，标志有两种可能的状态：

• True（已设置）：表示事件已发生，线程可以继续执行或不再被阻塞。
• False（未设置）：表示事件未发生，线程将处于阻塞状态，直到事件被触发。

2.4. Event使用方法

在多线程编程中，Event对象通常通过以下几个方法进行操作：

1. 创建Event对象：

import threading  
event = threading.Event()

2. 设置Event（触发事件）：
   使用set()方法将Event对象的内部标志设置为True，唤醒所有等待的线程。

event.set()

3. 清除Event（重置事件）：
   使用clear()方法将Event对象的内部标志重置为False，使等待的线程再次进入阻塞状态（对于自动重置事件，这一步是自动的）。

event.clear()

4. 等待Event（阻塞线程）：
   使用wait(timeout=None)方法使当前线程阻塞，直到Event对象的内部标志为True或调用超时。

event.wait()  # 可以指定超时时间，如 event.wait(1) 表示等待1秒

5. 检查Event状态：
   使用is_set()或isSet()（旧式Python中）方法检查Event对象的内部标志是否为True。

if event.is_set():  
    print("事件已发生")

2.5. 使用场景

Event在多线程编程中有广泛的应用场景，如：

• 生产者-消费者模式：生产者线程在生成数据后触发Event，消费者线程等待Event来接收数据。
• 线程间同步：当多个线程需要按照特定顺序执行时，可以使用Event来控制执行流程。
• 条件等待：在某些条件下，线程需要等待某个事件的发生才能继续执行，这时可以使用Event来实现。

3. 重启线程解决方案

在多线程和多进程编程中，Event 对象是一个重要的同步原语。它允许一个线程向其他线程发送信号，以通知某些事件已经发生。具体到 stop_event 的使用，其主要作用是控制线程的启动和停止。

3.1. 修改代码

最初，是在start_listening中增加延时sleep代码，但是，没有起作用。

def start_listening():
    if stop_event.is_set():
        stop_event.clear()
        time.sleep(3)

有效的是在stop_listening中增加event.clear()代码。

@app.route('/startlistening', methods=['GET'])
def start_listening():
    if not stop_event.is_set():
        stop_event.clear()
        #time.sleep(2)
        # 此处，可以增加sleep延时启动监听
    if not stop_event.is_set():
        #stop_event.clear()        
        if not hasattr(app,'pika_thread') or not app.pika_thread.is_alive():
            app.pika_thread = Thread(target = consume_from_rabbitmq_and_enqueue, args = (rabbitmq_params, rabbitmq_queue,  data_queue))
            app.pika_thread.start()
            return jsonify({'status':'listening'}), 200
        else:
            return jsonify({'status':'already listening'}), 200
    else:
        stop_event.clear()
        return jsonify({'status':'error','message':'Stop event is set.'}), 400

@app.route('/stoplistening', methods=['GET'])
def stop_listening():
    if hasattr(app,'pika_thread') and app.pika_thread.is_alive():
        stop_event.set()
        app.pika_thread.join()  # 等待线程结束
        del app.pika_thread     # 删除资源，确保重启成功
        
        time.sleep(2)  
        stop_event.clear()      # 新增起作用的代码！
      
        return jsonify({'status':'stopping'}), 200  
    else:
        return jsonify({'status':'not running'}), 400

3.2. 解释说明

1. stop_event.clear() 在 start_listening 中的作用
   在 start_listening 中调用 stop_event.clear() 确保了事件的内部标志被设置为 False。这意味着：

• 重新启动侦听线程时，如果事件的内部标志是 True，则不会启动新的线程。
• 通过 clear() 将标志设置为 False，确保新的线程可以启动，并开始侦听 RabbitMQ 消息。

2. stop_event.clear() 在 stop_listening 中的作用

• 在 stop_listening 中调用 stop_event.clear() 确保了事件的内部标志被重置为 False，这样下一次调用 start_listening 时，不会被之前的停止状态影响。
• clear() 的调用位置在 time.sleep(2) 之后，确保了有足够的时间让线程完全停止并释放资源。

3. 为什么只增加 sleep 不足以解决问题
   仅仅增加 sleep 只是延迟了操作，并没有改变 stop_event 的状态。stop_event 仍然是设置状态（True），这会导致 start_listening 无法通过 if not stop_event.is_set() 的检查，从而不会启动新的线程。

3.3. 小结

在多线程控制中，使用 Event 对象来管理线程的启动和停止非常有效。set() 和 clear() 方法可以控制线程的执行和等待状态。正确地使用 stop_event.clear() 确保了你的应用程序在停止线程后可以重新启动新的线程，而不受之前的停止状态影响。这是你代码中的关键步骤，确保了线程控制逻辑的正确性。