celery worker 多线程执行完后卡住假死

celery 任务代码大致如下，此任务未加 soft_time_limit 或 time_limit 其中 Session 是根据以下四行代码得来：

SQLALCHEMY_DATABASE_URI = 'mysql://xx'
some_engine = create_engine(SQLALCHEMY_DATABASE_URI, echo=False, pool_pre_ping=True)
session_factory = sessionmaker(autocommit=False, bind=some_engine)  # autoflush=False,
Session = scoped_session(session_factory)

celery 任务大致代码：

'''
class NmapThread_(threading.Thread):
    def __init__(self, queue):
        threading.Thread.__init__(self)
        self.queue = queue

    def run(self):
        while True:
            try:
                ip = self.queue.get(block=False)
            except SoftTimeLimitExceeded as e:
                raise
            except Exception as e:
                logging.error(e, exc_info=True)  # 记录线程退出
                break
            try:
                ns = nmap.scan(ip)  # 使用 subprocess.Popen()调用外部 nmap 扫描该 ip 并将结果入库
                ns.run()
            except SoftTimeLimitExceeded as e:
                raise
            except Exception as e:
                logging.error(e, exc_info=True)
            finally:
                Session.remove()
                self.queue.task_done()


class IpScan:
    def __init__(self):
        self.queue = Queue()
        self.thread_count = 2

    def run(self):
        # 此处入队列的 IP 个数非常多，大概需要多线程运行 1 周
        self.queue.put('1.1.1.1')
        self.queue.put('2.2.2.2')
        self.queue.put('3.3.3.3')
        self.queue.put('4.4.4.4')
        self.queue.put('5.5.5.5')
        self.queue.put('6.6.6.6')
        self.scan_start()

    def scan_start(self):
        for i in range(self.thread_count):
            t = NmapThread_(self.queue)
            t.setDaemon(True)
            t.start()
        self.queue.join()

def web_fingerprint_discern(*args, **kwargs):
    print('web_fingerprint_discern begin!')
    print(args)
    print(kwargs)

@ce.task(name='default.test4queue', bind=True)
def test4queue(self):
    ips = IpScan()
    ips.run()
    web_fingerprint_discern()
'''

现在的问题是多线程运行期间一直没有问题，直到最后多线程执行完(2019-08-31)，抛出异常 break 跳出各线程之后，celry worker 就卡住假死了(2019-08-31)，具体表现在：

celery worker 卡住假死，没有再执行接下来的 web_fingerprint_discern()，这个函数里面其实只有 3 个 print()
celery flower 显示 celery worker 已离线
rabbitMQ 显示队列中的任务还在执行之中

然后我今天(2019-09-02)上班的时候强制 ctrl+c 之后，输出了一些日志，

大家可以看到， web_fingerprint_discern()的 3 条 print 语句，第 1 条发生 2019-08-31，后 2 条发生在 2019-09-02 我 ctrl+c 的时候，其他附加表现：

最后报错了 broken pipe
rabbitMQ 队列中的任务已经回滚到 Ready 阶段，也就是说如果重启 celery worker 任务会重新执行

一直不能解决这个 celery 问题，由于时间原因也不方便换其他类 celery 架构，而且调试发现如果不是扫描任务就不会假死（当然测试时间肯定没有一周那么长，只有几分钟，所以如果完整模拟测试非常耗时，想寻找可能出现的问题点修复后再行测试）

求各位大佬帮忙解决，若能解决，50 红包奉上以表谢意。

hanssx

2019-09-02 12:26:44 +08:00

celery 版本
```
(asset) [root@VM_9_196_centos asset]# celery --version
4.3.0 (rhubarb)
```

多线程应该是结束了，我在日志中收到了 break 线程的 while 循环之前，打印出的日志，一共 30 个日志，我也启了 30 个线程。
```
[2019-08-31 22:47:31] [ERROR] - (asscan.py:38)
Traceback (most recent call last):
File "/root/python/asset/scan/asscan.py", line 34, in run
task_host = self.queue.get(block=False)
File "/usr/local/python3/lib/python3.7/queue.py", line 167, in get
raise Empty
_queue.Empty
[2019-08-31 22:48:12] [ERROR] - (asscan.py:38)
Traceback (most recent call last):
File "/root/python/asset/scan/asscan.py", line 34, in run
task_host = self.queue.get(block=False)
File "/usr/local/python3/lib/python3.7/queue.py", line 167, in get
raise Empty
_queue.Empty
[2019-08-31 22:48:12] [WARNING] - web_fingerprint_discern begin! (log.py:235)
[2019-09-02 10:21:21] [WARNING] - () (log.py:235)
[2019-09-02 10:21:21] [WARNING] - {'task_uuid': '95c3dda4-cb04-11e9-a344-52540
```

neoblackcap

2019-09-02 12:38:10 +08:00

没记错的话，celery 自身实现是对 fork 之类有限制的，所以你不应该在任务里面进行类似 fork 之类的操作，线程 pthread_create 同理了。
而且线程的支持我记得已经被 celery 自身抛弃的，所以应该是有缺陷的，建议不使用线程。

根据我以前的做法，我一般都是将网络 IO 与逻辑处理分离。celery 对 gevent 跟进程支持都相当好，因此我会选用个 gevent 处理所有网络 IO （网络 IO，通过 IO 复用，几百万个任务都可以轻松搞定，前提是不能有任何 CPU 密集型处理）。然后通过跟进程型任务结合，组成流水线，在 celery 对应 chain 操作。那么就可以稳定地运行。

因为 gevent 是处理网络是不堵塞的，所以你还是可以继续发任务给该 worker

可以参考一下