后台程序处理（二） python threading - queue 模块使用

时间：2023-12-17 08:33:02

由于协程没办法完成（一）中所说的任务模式

接下来就尝试一下使用线程和队列来实现一下这个功能

在实现之前，我们先明确一个问题——python的线程是伪并发的。同一时间只能有一个线程在运行。具体怎样的运作方式由解释器决定

然后回顾一下上一章遇到的问题——return以后，需要另外一个线程去检测之前的操作是否执行成功

因此程序的流程设计应该是这样的：

 # 大致流程步骤如下

 # 1.获取参数（接口被访问时触发）

 request_data = request.form

 # 2.根据参数查询内容

 target = Target.query.filter_by(id=request_data).first()

 # 3.将结果插入队列

 ans_queue.put(target)

 # 4.激活线程

 thread.set()

 # 5.将结果从队列中取出

 ans_queue.get()

 # 6.处理结果

 check()

 # 7.将线程休眠（阻塞）

 thread.event.clear()

这样设计的考虑主要是以下几点：

1.简单

2.入队可以保证消息按时间顺序被处理

3.出队可以保证当队列不为空时，检查线程会执行到队列为空为止。免去不必要的唤醒检查。然后在有消息入队时被重新激活

4.其实我们的设计正常来说不会出现3中的检查情况。基本上队列一旦有消息入队，线程就会启动并清空队列

5.入队可以保证消息的完整和独立性，每次请求得到的数据入队后，队列中都是一列数组。处理逻辑更清晰

6.队列中的数据不出栈是不可见的

7.我就是宁愿用全局队列也不想用全局变量

实际接口代码和线程代码如下：

A.队列和线程代码

 # 消息队列

 lock_queue = Queue()

 def check_kill(event):

     while True:

         # check queue

         if lock_queue.empty() is True:

             event.clear()

         # wait event

         if event.is_set() is not True:

             event.wait()

         # do some work

         sids, serials, minutes, hosts, insts, opasses, ospasses = [], [], [], [], [], [], []

         # get data until queue empty or datas more than 10

         if lock_queue.empty() is not True:

             data = lock_queue.get()

             for i in data:

                 sid, serial, minute, host, inst, opass, ospass = i.split(',')

                 sids.append(sid)

                 serials.append(serial)

                 minutes.append(minute)

                 hosts.append(host)

                 insts.append(inst)

                 opasses.append(opass)

                 ospasses.append(ospass)

         # init the command

         kill_command = 'kill -9'

         # each time we deal less or equal 10 check

         for i in range(len(minutes)):

             current = datetime.datetime.now().minute

             if current >= int(minutes[i]):

                 passtime = current - int(minutes[i])

             else:

                 passtime = current + 60 - int(minutes[i])

             print("passtime is", passtime)

             if (5 - passtime) >= 0:

                 time.sleep((5 - passtime)*60)

             # split piece of list

             sql_sids, sids = sids[0], sids[1:]

             sql_serials, serials = serials[0], serials[1:]

             sql_hosts, hosts = hosts[0], hosts[1:]

             sql_insts, insts = insts[0], insts[1:]

             sql_opass, opasses = opasses[0], opasses[1:]

             sql_ospass, ospasses = ospasses[0], ospasses[1:]

             print("data", sql_hosts, sql_insts, sql_serials, sql_sids)

             # create cursor

             try:

                 conn = sqlite3.connect('data-dev.sqlite')

                 c = conn.cursor()

                 cu = c.execute("select ouser,oport,osport,osuser from tool_target where host='%s' and inst='%s'" % (sql_hosts, sql_insts))

                 result = cu.fetchall()

                 ouser = result[0][0]

                 opass = sql_opass

                 str_conn = (sql_hosts

                             + ':'

                             + str(result[0][1])

                             + '/'

                             + sql_insts)

                 odb = cx_Oracle.connect(ouser, opass, str_conn)

                 cursor = odb.cursor()

                 # select to find if lock exist

                 sql = '''select b.spid, a.sid, a.serial#, a.event from v$session a, v$process b

                         where a.sid = %s and a.serial# = %s ''' % (sql_sids, sql_serials)

                 cursor.execute(sql)

                 answer = cursor.fetchall()

                 print("answer is", answer)

                 kill_command += ' ' + answer[0][0]

                 s = paramiko.SSHClient()

                 s.load_system_host_keys()

                 s.set_missing_host_key_policy(paramiko.AutoAddPolicy())

                 s.connect(sql_hosts, result[0][2], result[0][3], sql_ospass)

                 stdin, stdout, stderr = s.exec_command(kill_command)

                 stdout.read()

                 print('------------------------')

                 s.close()

                 cursor.close()

                 odb.close()

                 c.close()

                 conn.close()

             except:

                 pass

 txkill_ready = threading.Event()

 t1 = threading.Thread(target=check_kill, args=(txkill_ready,), name='t1')

 t1.start()

 # txkill_ready.set()

B.接口代码

 @main.route('/txlock/startkillurl', methods=['POST'])

 def start_kill_url():

     if request.method == 'POST':

         cmd = request.form.getlist('list')[0]

         host = request.form.getlist('host')[0]

         inst = request.form.getlist('inst')[0]

         # print(len(cmd))

         # cmd.replace("\n", "")

         # cmd.replace("\t", "")

         # print(len(cmd))

         tooltarget = ToolTarget.query.filter_by(host=host, inst=inst).first()

         ouser = tooltarget.ouser

         opass = ToolTarget.de_rsa(pwd=tooltarget.opass)

         ospass = ToolTarget.de_rsa(pwd=tooltarget.ospass)

         str_conn = (tooltarget.host

                     + ':'

                     + str(tooltarget.oport)

                     + '/'

                     + tooltarget.inst)

         odb = cx_Oracle.connect(ouser, opass, str_conn)

         cursor = odb.cursor()

         # add into queue

         c = re.findall('\d*,\d*', cmd)

         d = [i+','+str(datetime.datetime.now().minute)+','+host+','+inst+','+opass+','+ospass for i in c]

         # data example : ['15,5,17', '16,23,17', '14,5,17', '142,1,17']

         lock_queue.put(d)

         txkill_ready.set()

         try:

             cursor.execute(cmd)

             # pass

         except:

             return "执行失败,关闭弹窗后会自动刷新列表"

     return "执行成功,关闭弹窗后会自动刷新列表"

相关文章