激活回调功能 ThreadBasedSession 的回调实现原理是：创建一个单独的线程用于接收回调事件，进而调用相关的回调函数。 当用户Task中并没有使用到回调功能时，不必开启此线程，可以节省资源
(self)
| 222 | self._cleanup(nonblock=nonblock) |
| 223 | |
| 224 | def _activate_callback_env(self): |
| 225 | """激活回调功能 |
| 226 | |
| 227 | ThreadBasedSession 的回调实现原理是：创建一个单独的线程用于接收回调事件，进而调用相关的回调函数。 |
| 228 | 当用户Task中并没有使用到回调功能时，不必开启此线程，可以节省资源 |
| 229 | """ |
| 230 | |
| 231 | if self.callback_mq is not None: # 回调功能已经激活 |
| 232 | return |
| 233 | |
| 234 | self.callback_mq = queue.Queue(maxsize=self.callback_mq_maxsize) |
| 235 | self.callback_thread = threading.Thread(target=self._dispatch_callback_event, |
| 236 | daemon=True, name='callback-' + random_str(10)) |
| 237 | # self.register_thread(self.callback_thread) |
| 238 | self.thread2session[id(self.callback_thread)] = self # 用于在线程内获取会话 |
| 239 | event_mq = queue.Queue(maxsize=self.event_mq_maxsize) # 回调线程内的用户事件队列 |
| 240 | self.task_mqs[self._get_task_id(self.callback_thread)] = event_mq |
| 241 | |
| 242 | self.callback_thread.start() |
| 243 | logger.debug('Callback thread start') |
| 244 | |
| 245 | def _dispatch_callback_event(self): |
| 246 | while not self.closed(): |
no test coverage detected