2. 程式人生 > >Python佇列及在微信機器人中的應用

Python佇列及在微信機器人中的應用

阿新 • • 發佈：2019-01-28

本文來源於i春秋學院，未經允許嚴禁轉載。

最近打算更新微信機器人，發現機器人的作者將程式碼改進了很多，但去掉了sqlite資料庫，需要自己根據需求設計資料庫，跟作者溝通得到的建議是為了防止訊息併發導致資料庫死鎖，建議另開一個程序讀寫資料庫，將訊息加入一個佇列中，因為對Python瞭解有限，佇列和多執行緒更不是我擅長的內容，於是最近瘋狂Google、百度，探索著實現了此功能。寫此文記錄下基本概念和實現方法

0x00 Python佇列
佇列是執行緒中交換資料的形式。
建立一個佇列物件

import Queue

q = Queue.Queue(maxsize = 10) #maxsize是佇列長度，不限制長度可以不不賦值


將一個值放入佇列

q.put(10)


將一個值從佇列取出

q.get()


三種佇列及建構函式

  • FIFO佇列先進先出: class Queue.Queue(maxsize)
  • LIFO類似於堆，即先進後出: class Queue.LifoQueue(maxsize)
  • 優先順序佇列: class Queue.PriorityQueue(maxsize)
     

佇列的常用方法

q.qsize() #返回佇列的大小

q.empty() #如果佇列為空，返回True,反之False

q.full() #如果佇列滿了，返回True,反之False

q.full #與 maxsize 大小對應

q.get([block[, timeout]
 
]) #獲取佇列，block:是否阻塞等待，timeout等待時間

q.get_nowait() #相當q.get(False)

q.put(item[, block[, timeout]) #非阻塞寫入佇列，block:是否阻塞等待，timeout等待時間

q.put_nowait(item) #相當q.put(item, False)

q.task_done() #在完成一項工作之後，q.task_done() 函式向任務已經完成的佇列傳送一個訊號

q.join() #等到佇列為空，再執行別的操作


幾個例子

一個執行緒往佇列裡寫入隨機數，另一個執行緒從佇列裡取數字(阻塞等待)

#!/usr/bin/env python
 


#coding:utf8

import random,threading,time

from Queue import Queue

#Producer thread

classProducer(threading.Thread):

    def__init__(self, t_name, queue):

        threading.Thread.__init__(self,name=t_name)

        self.data=queue

    defrun(self):

        for i in range(10):  #隨機產生10個數字 ，可以修改為任意大小

            randomnum=random.randint(1,99)

            print "%s: %s 生成了一個數字 %d 並把它扔進了佇列!" % (time.ctime(), self.getName(), randomnum)

            self.data.put(randomnum)  #將資料依次存入佇列

            time.sleep(1)

        print "%s: %s finished!" %(time.ctime(), self.getName())



#Consumer thread

classConsumer_all(threading.Thread):

    def__init__(self, t_name, queue):

        threading.Thread.__init__(self, name=t_name)

        self.data = queue



    defrun(self):

        while 1:

            try:

                # print self.data

                val_even = self.data.get() 

                print "%s: %s 從佇列裡取出了 %d !" % (time.ctime(), self.getName(), val_even)

                time.sleep(1)

            except  Exception, e: 

                print '取資料失敗'

                continue



#Main thread

defmain():

    queue = Queue()

    producer = Producer('Pro.', queue)

    consumer_all = Consumer_all('Con_all.', queue)

    producer.start()

    consumer_all.start()



if __name__ == '__main__':

    main()

執行結果

Wed Aug  3 00:04:18 2016: Pro. 生成了一個數字 74 並把它扔進了佇列!

Wed Aug  3 00:04:18 2016: Con_all. 從佇列裡取出了 74 !

Wed Aug  3 00:04:19 2016: Pro. 生成了一個數字 77 並把它扔進了佇列!

Wed Aug  3 00:04:19 2016: Con_all. 從佇列裡取出了 77 !

Wed Aug  3 00:04:20 2016: Pro. 生成了一個數字 63 並把它扔進了佇列!

Wed Aug  3 00:04:20 2016: Con_all. 從佇列裡取出了 63 !

Wed Aug  3 00:04:21 2016: Pro. 生成了一個數字 96 並把它扔進了佇列!

Wed Aug  3 00:04:21 2016: Con_all. 從佇列裡取出了 96 !

Wed Aug  3 00:04:22 2016: Pro. 生成了一個數字 82 並把它扔進了佇列!

Wed Aug  3 00:04:22 2016: Con_all. 從佇列裡取出了 82 !

Wed Aug  3 00:04:23 2016: Pro. 生成了一個數字 19 並把它扔進了佇列!

Wed Aug  3 00:04:23 2016: Con_all. 從佇列裡取出了 19 !

Wed Aug  3 00:04:24 2016: Pro. 生成了一個數字 56 並把它扔進了佇列!

Wed Aug  3 00:04:24 2016: Con_all. 從佇列裡取出了 56 !

Wed Aug  3 00:04:25 2016: Pro. 生成了一個數字 57 並把它扔進了佇列!

Wed Aug  3 00:04:25 2016: Con_all. 從佇列裡取出了 57 !

Wed Aug  3 00:04:26 2016: Pro. 生成了一個數字 42 並把它扔進了佇列!

Wed Aug  3 00:04:26 2016: Con_all. 從佇列裡取出了 42 !

Wed Aug  3 00:04:27 2016: Pro. 生成了一個數字 7 並把它扔進了佇列!

Wed Aug  3 00:04:27 2016: Con_all. 從佇列裡取出了 7 !

Wed Aug  3 00:04:28 2016: Pro. finished!

如果將入隊的時間間隔修改，出隊的程式將阻塞執行，將“ time.sleep(1)”改為“ time.sleep(10)”，再次執行

Wed Aug  3 00:10:46 2016: Pro. 生成了一個數字 45 並把它扔進了佇列!

Wed Aug  3 00:10:46 2016: Con_all. 從佇列裡取出了 45 !

Wed Aug  3 00:10:56 2016: Pro. 生成了一個數字 75 並把它扔進了佇列!

Wed Aug  3 00:10:56 2016: Con_all. 從佇列裡取出了 75 !

Wed Aug  3 00:11:06 2016: Pro. 生成了一個數字 26 並把它扔進了佇列!

Wed Aug  3 00:11:06 2016: Con_all. 從佇列裡取出了 26 !

Wed Aug  3 00:11:16 2016: Pro. 生成了一個數字 67 並把它扔進了佇列!

Wed Aug  3 00:11:16 2016: Con_all. 從佇列裡取出了 67 !

Wed Aug  3 00:11:26 2016: Pro. 生成了一個數字 60 並把它扔進了佇列!

Wed Aug  3 00:11:26 2016: Con_all. 從佇列裡取出了 60 !

Wed Aug  3 00:11:36 2016: Pro. 生成了一個數字 83 並把它扔進了佇列!

Wed Aug  3 00:11:36 2016: Con_all. 從佇列裡取出了 83 !

Wed Aug  3 00:11:46 2016: Pro. 生成了一個數字 33 並把它扔進了佇列!

Wed Aug  3 00:11:46 2016: Con_all. 從佇列裡取出了 33 !

Wed Aug  3 00:11:56 2016: Pro. 生成了一個數字 65 並把它扔進了佇列!

Wed Aug  3 00:11:56 2016: Con_all. 從佇列裡取出了 65 !

Wed Aug  3 00:12:06 2016: Pro. 生成了一個數字 44 並把它扔進了佇列!

Wed Aug  3 00:12:06 2016: Con_all. 從佇列裡取出了 44 !

Wed Aug  3 00:12:16 2016: Pro. 生成了一個數字 28 並把它扔進了佇列!

Wed Aug  3 00:12:16 2016: Con_all. 從佇列裡取出了 28 !

Wed Aug  3 00:12:26 2016: Pro. finished!

會發現當佇列沒有值的時候程式會阻塞等待佇列有值才繼續執行。
稍微修改程式，一個執行緒往佇列裡寫入隨機數，另一個執行緒從佇列裡取數字(非阻塞等待)

#!/usr/bin/env python

#coding:utf8

import random,threading,time

from Queue import Queue

#Producer thread

classProducer(threading.Thread):

    def__init__(self, t_name, queue):

        threading.Thread.__init__(self,name=t_name)

        self.data=queue

    defrun(self):

        for i in range(10):  #隨機產生10個數字 ，可以修改為任意大小

            randomnum=random.randint(1,99)

            print "%s: %s 生成了一個數字 %d 並把它扔進了佇列!" % (time.ctime(), self.getName(), randomnum)

            self.data.put(randomnum)  #將資料依次存入佇列

            time.sleep(10)

        print "%s: %s finished!" %(time.ctime(), self.getName())



#Consumer thread

classConsumer_all(threading.Thread):

    def__init__(self, t_name, queue):

        threading.Thread.__init__(self, name=t_name)

        self.data = queue



    defrun(self):

        while 1:

            try:

                # print self.data

                val_even = self.data.get(1,5)  # get(self, block=True, timeout=None) ,1就是阻塞等待,5是超時5秒

                print "%s: %s 從佇列裡取出了 %d !" % (time.ctime(), self.getName(), val_even)

                time.sleep(1)

            except  Exception, e:  # 等待輸入，超過5秒  就報異常

                print '取資料失敗'

                continue



#Main thread

defmain():

    queue = Queue()

    producer = Producer('Pro.', queue)

    consumer_all = Consumer_all('Con_all.', queue)

    producer.start()

    consumer_all.start()



if __name__ == '__main__':

    main()
Wed Aug  3 00:20:58 2016: Pro. 生成了一個數字 57 並把它扔進了佇列!

Wed Aug  3 00:20:58 2016: Con_all. 從佇列裡取出了 57 !

取資料失敗

Wed Aug  3 00:21:08 2016: Pro. 生成了一個數字 83 並把它扔進了佇列!

Wed Aug  3 00:21:08 2016: Con_all. 從佇列裡取出了 83 !

取資料失敗

Wed Aug  3 00:21:18 2016: Pro. 生成了一個數字 85 並把它扔進了佇列!

Wed Aug  3 00:21:18 2016: Con_all. 從佇列裡取出了 85 !

取資料失敗

Wed Aug  3 00:21:28 2016: Pro. 生成了一個數字 64 並把它扔進了佇列!

Wed Aug  3 00:21:28 2016: Con_all. 從佇列裡取出了 64 !

取資料失敗

Wed Aug  3 00:21:38 2016: Pro. 生成了一個數字 23 並把它扔進了佇列!

Wed Aug  3 00:21:38 2016: Con_all. 從佇列裡取出了 23 !

取資料失敗

Wed Aug  3 00:21:48 2016: Pro. 生成了一個數字 98 並把它扔進了佇列!

Wed Aug  3 00:21:48 2016: Con_all. 從佇列裡取出了 98 !

取資料失敗

Wed Aug  3 00:21:58 2016: Pro. 生成了一個數字 12 並把它扔進了佇列!

Wed Aug  3 00:21:58 2016: Con_all. 從佇列裡取出了 12 !

取資料失敗

Wed Aug  3 00:22:08 2016: Pro. 生成了一個數字 27 並把它扔進了佇列!

Wed Aug  3 00:22:08 2016: Con_all. 從佇列裡取出了 27 !

取資料失敗

Wed Aug  3 00:22:18 2016: Pro. 生成了一個數字 60 並把它扔進了佇列!

Wed Aug  3 00:22:18 2016: Con_all. 從佇列裡取出了 60 !

取資料失敗

Wed Aug  3 00:22:28 2016: Pro. 生成了一個數字 69 並把它扔進了佇列!

Wed Aug  3 00:22:28 2016: Con_all. 從佇列裡取出了 69 !

取資料失敗

Wed Aug  3 00:22:38 2016: Pro. finished!

取資料失敗

取資料失敗

取資料失敗

可以看出，取資料的執行緒在超過設定的等待時間後會丟擲異常並繼續往下執行。

0x01 使用佇列將微信機器人訊息存入MongoDB

使用上面的例子，稍作修改，將接收到的訊息扔進佇列，開啟另一個執行緒取資料，取到後將訊息格式化並存入資料庫

#!/usr/bin/env python

#coding:utf8

import threading,time,pymongo

from pymongo import MongoClient

from Queue import Queue

#訊息入隊

classMsgInQueue():

    def__init__(self, queue):

        threading.Thread.__init__(self)

        self.data=queue

    defputmsgqueue(self,msg):

        self.data.put(msg)

        print 'put msg in queue success:'+msg['Content']



###訊息出隊並存入資料庫

classMsgOutQueue2db(threading.Thread):

    def__init__(self, queue):

        threading.Thread.__init__(self)

        self.data = queue

        #建立MongoDB連線

        self.conn = MongoClient()

        #資料庫

        self.db = self.conn.wechatRobot

        #資料表

        self.messages = self.db.messages

    defrun(self):

        while 1:

            try:

                # print self.data

                #從佇列裡取訊息

                msg = self.data.get(1, 5)  # get(self, block=True, timeout=None) ,1就是阻塞等待,5是超時5秒

                print "%s: %s get %s from queue !" % (time.ctime(), self.getName(), msg['Content'].encode('utf-8'))

                try:

                    #格式化訊息資料

                    m = dict(groupname=msg['FromUserName'].encode('utf-8'),

                             time=msg['CreateTime'],

                             username=msg['ActualUserName'],

                             usernickname=msg['ActualNickName'].encode('utf-8'),

                             message=msg['Content'].encode('utf-8'),

                             messagetype=msg['MsgType']

                             )

                    print m

                    #存入資料庫

                    self.db.messages.insert(m)

                    time.sleep(1)

                except  Exception, e:

                    print e

                    continue

            except  Exception, e:

                continue

主執行緒

defcomplex_reply():

    queue = Queue()

    outqueue = MsgOutQueue2db(queue)#例項化出隊入庫類

    outqueue.start()#開啟執行緒



    @itchat.msg_register('Text', isGroupChat = True)

    deftext_reply(msg):

        # print itchat.__client.storageClass.groupDict

        print itchat.__client.storageClass.chatroomList

        print msg

        inqueue=MsgInQueue(queue)#例項化入隊類

        inqueue.putmsgqueue(msg)#訊息入隊

        if msg['isAt']:

            print msg

            itchat.send(u'@%s\u2005I received: %s'%(msg['ActualNickName'], msg['Content']), msg['FromUserName'])

訊息存入資料庫:

通過多執行緒和MongoDB的結合，有效防止訊息過多導致資料庫死鎖的問題，也更加模組化，可以根據真實需求更換其他資料庫。後面我將結合MongoDB插入、更新資料快的特點寫一下我如何設計群聊統計功能，在Python方面和MongoDB方面我都是小白，如有更好的建議請多指教，我們共同學習！有關Python多執行緒的課程請參考《python安全程式設計入門》

相關推薦

Python佇列及在微信機器人中的應用

本文來源於i春秋學院，未經允許嚴禁轉載。 最近打算更新微信機器人，發現機器人的作者將程式碼改進了很多，但去掉了sqlite資料庫，需要自己根據需求設計資料庫，跟作者溝通得到的建議是為了防止訊息併發導致資料庫死鎖，建議另開一個程序讀寫資料庫，將訊息加入一個佇列中，因為對Pyt

VueJs單頁應用實現微信網頁授權及微信分享功能示例

在實際開發中，無論是做PC端、WebApp端還是微信公眾號等型別的專案的時候，或多或少都會涉及到微信相關的開發，最近公司專案要求實現微信網頁授權，並獲取微信使用者基本資訊的功能及微信分享的功能，現在總算完成了，但開發過程中遇到好幾個坑。廢話不多說了，開始正題。 描述點 微信相關開發知識

Web應用多賬號系統設計及微信掃碼登入實現

class WeChatAuthCallback(MyBaseHandler): """ 微信第三方認證之後，開始將此使用者在本系統沉澱下來 - 用於微信伺服器傳回code的值 - 此處要再請求獲得access_token """ asy

Android整合微信分享功能應用簽名生成方法及分享不生效的問題

通過友盟sdk整合微博、微信、qq等分享功能時，微博和qq很順利，但在做微信整合時一直不成功。主要問題還是之前在微信開放平臺申請建立移動應用時，對應用簽名沒有填寫對，走了很多彎路現總結出來，加深記憶避

微信小程式/應用號開發IDE破解版（免APPID）下載及教程

微信應用號/小程式開發IDE 交流Q群：294630305 注意：先下0.7版用於登入，再下0.9版用於開發，目前只能本地編譯，不能上傳 下載宣告 相關程式碼純屬技術研究與愛好，請在學習使用後刪除 相關下載資源均來自於QQ群，與作者及貢獻者本人無關 當微信官方

分享到QQ及微信成功後無法返回應用

分享到QQ及微信無法返回應用主要檢查下面兩點： 1. URL scheme是否按照要求配置 在你的工程設定項,targets 一欄下,選中自己的 target,在 Info->URL Types 中新增 URL Schemes 配置QQ URL scheme 格式： “QQ”+騰訊QQ互聯應用appId

[Python微信開發] 一.itchat入門知識及微信自動回覆、微信簽名詞雲分析

轉自： https://blog.csdn.net/Eastmount/article/details/79618039 最近準備學習微信小程式開發，偶然間看到了python與微信互動的介面itchat，簡單學習了下，感覺還挺有意思的，故寫了篇基礎文章供大家學習。it

Ultimus BPM微信版本移動應用解決方案

微信版本移動應用解決方案特點 ?提供手機審批配置平臺，用戶可選擇任意流程的任意節點進行手機審批。 ?移動化辦公，智能手機與平板體驗，隨時隨地進行審批，不受終端的局限。 ?移動終端可以支持iOS、Android、Window

Python 小程序-微信AI自動回復

ges images 微信 ccf try 消息 .post style register 一簡介 　　使用itcha實現微信登錄及消息發送接收，使用圖靈AI實現智能回復。 參考API： http://www.tuling123.com/ https://itchat.re

python利用企業微信api來進行發送自定義報警的類實現

python 微信報警python利用企業微信api來進行發送自定義報警的類實現企業微信註冊打開http://work.weixin.qq.com/企業微信主頁；點擊企業註冊；填寫相關信息，營業執照和註冊號可以不用填，直接下一步，按照提示操作即可；註冊完成後，登陸，就顯示如下界面：點擊我的企業標簽：看到如上界

微信jsapi開發應用實例並記錄下錯誤信息

var appid json gbk 信息 pip rand 常常 數據 雖然大家已經有實例，但還是要根據自己的項目更改 在這裏需要註意的幾點 也是常常報錯的 1、頁面必須是UTF8（已經json不認gbk，所以會提醒 -1調用支付JSAPI缺少參數APPID 這是編碼引起

Python 開發個人微信號在運維開發中的使用

模塊開發 padding 效果 用處 微信網頁版 企業 微信 技術分享 body 一、主題：Python 開發個人微信號在運維開發中的使用 二、內容： 企業公眾號 介紹開發微信公眾號的後臺邏輯，包括服務器驗證邏輯、用戶認證邏輯 個人微信號 面對企業微信的種種限制，可

用python寫一個微信聊天機器人

python wechat 聊天機器人 # -*- coding: utf-8 -*- """ package.module ~~~~~~~~~~~~~~ 一個微信機器人程序 微信客戶端itchat: http://itchat.readthed

我用 Python 爬取微信好友，最後發現一個大秘密

代碼 我們 同學 strong 分享 簽名 ast ron tps 前言 你身處的環境是什麽樣，你就會成為什麽樣的人。現在人們日常生活基本上離不開微信，但微信不單單是一個即時通訊軟件，微信更像是虛擬的現實世界。你所處的朋友圈是怎麽樣，慢慢你的思想也會變的怎麽樣。最近在學習

用python寫一個微信跳一跳外掛,瞬間稱霸朋友圈

python 微信 跳一跳 爬蟲12月28日，微信宣布，小程序增加了新的類目：小遊戲，同時上線小遊戲 你們跳的再好，在毫無心理波動的程序面前都是渣渣。 剛剛會python的小白想玩怎麽辦？ 下有詳細的教程，哈哈，包教會不收任何的費用。 感受一下被支配的恐懼吧： 使用工具1.python3.6 2.adb 3

python 打造一個微信跳一跳輔助！（手機本地運行）

.com simple pre 註釋 ima 獲取 IE 分享圖片 mask 先上成果效果圖： 用opencv 做識圖識別出棋子的坐標並把它框出來 終點位置的坐標是： 先觀察圖像發現棋子每跳過後的下一個目標點總是在棋子的上面 這樣就可以先獲取一個感興趣的區域，用num

博客站點選擇SqlServer數據庫版本技巧及微信打賞插件分析[圖]

sqlserver 博客 微信 插件 打賞 首先和大家分享一個有趣的社會現象，也是關於IT行業的，對於技術方面來說，就不用多說了，這些年的進步和變化大家都有目共睹，雖然我只是一個技術小白，但是最近也開始不斷的學習中。 自媒體平臺其實就是博客的集中營： 這兩年出現了火爆的自媒體平臺，卷走了大

Python學習---模擬微信網頁登錄180410

link 符號 byod 分析 字符 uid order ons tro WEB微信 網頁登錄的猜想： a. 訪問頁面出現二維碼 b. 長輪詢監聽是否已經掃碼並且點擊確認 c. 如何進行會話保持 d.

微信PK10平臺開發與用python爬取微信公眾號文章

網址 谷歌瀏覽器 pytho google http 開發 微信 安裝python rom 本文通過微信提供微信PK10平臺開發[q-21528-76294] 網址diguaym.com 的公眾號文章調用接口，實現爬取公眾號文章的功能。註意事項 1.需要安裝python s

微信小程式(應用號)資源彙總整理

原文：http://www.cnblogs.com/txw1958/p/wechat-app-summarizing.html 微信小應用資源彙總整理 開源專案 WeApp - 微信小程式版的微信 wechat-weapp-redux-todos - 微信