线程中的socket实现

Question

我必须从数据库中读取一些数据并从 tcp 套接字发送 所以我从数据库中获取数据

    #main
    while True:
        cursor.execute("UPDATE `raw` SET `zombie`='"+zombieId+"' WHERE result='pending' AND protocol='tcp' AND zombie='0' LIMIT 1;")
#       time.sleep(0.2)
        cursor.execute("select * from raw WHERE `result`='pending' AND `protocol`='tcp' and `zombie`='"+zombieId+"' limit 1;")

            if cursor.rowcount>0 :
                    waitedSig = cursor.fetchone()
                    time.sleep(0.2)
                    t = threading.Thread(target=sendData , args=((waitedSig),))
                    t.start()
                    time.sleep(0.6)
            else:
                    time.sleep(1)

在线程上我将向目标发送数据

    def sendData(sig):
        timedata = datetime.datetime.fromtimestamp(int(sig[16]))
        devimei = sig[23]
        devdate = timedata.strftime("%d%m%y")
        devtime = timedata.strftime("%H%M%S")
        lat= format(sig[2])
        lon= format(sig[3])
        satcount = format(sig[5])
        speed = format(sig[4])
        batery = format(sig[7])
        if sig[9]>1000:
                band='00'
        elif sig[9]>850:
                band='02'
        else:
                band='01'
        hdop  = format(sig[10])
        gsmQ =  format(sig[6])
        lac =  format(sig[12])
        cid =  format(sig[13])
str='$MGV002,'+devimei+',12345,S,'+devdate+','+devtime+',A,'+lat+',N,'+lon+',E,0,'+satcount+',00,'+hdop+','+speed+',0,,,432,11,'+lac+','
            try:
                    clientsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                    result = clientsocket.connect(('ip',port))
                    clientsocket.send(str)
                    data = clientsocket.recv(1024)
                    print(str(datetime.datetime.now())+' -> send completed :'+format(sig[0]))
                    clientsocket.close()
            except:
                    print(str(datetime.datetime.now())+' -> connection to tcp server failed!!')

这会很好用，但有两个无聊的问题：

1) 如果我删除 0.2 和 0.6 睡眠延迟脚本崩溃由于重复的套接字使用，似乎系统试图打开另一个套接字，直到前一个还没有完成它的工作！

2) 如果sendData 函数出现问题，整个脚本将停止工作，直到我手动重新启动脚本

如此

1）我可以创建一个线程队列来一个接一个地运行并且不相互影响吗？！

2) 我如何处理线程函数中的错误以仅关闭该特定线程和脚本继续处理下一个数据库记录？！

Answer 1

这看起来像是线程池的一个很好的应用。在您的实现中，您为数据库表中的每个项目创建一个线程和套接字，这可能会对系统造成极大的负担。在这里，我创建了 20 个工人作为示例。当您开始对系统施加压力时，工人数量的回报就会递减。

import multiprocessing.pool

def sender():
    pool = multiprocessing.pool.ThreadPool(20) # pick your size...
    cursor.execute("select * from database")
    pool.map(sendData, cursor, chunksize=1)

def sendData(sig):
        try:
                clientsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                result = clientsocket.connect(('ip',port))
                clientsocket.sendall(sig)
                data = clientsocket.recv(1024)
                print(str(datetime.datetime.now())+' -> send completed :'+format(sig[0]))
                clientsocket.shutdown(socket.SOCK_RDWR)
                clientsocket.close()
        except:
                print(str(datetime.datetime.now())+' -> connection to tcp server fa