day11

時間  2019-11-13
標籤 day11 day 简体版
原文   原文鏈接

Day11 - 異步IO\數據庫\隊列\緩存

 

本節內容html

  1. Gevent協程
  2. Select\Poll\Epoll異步IO與事件驅動
  3. Python鏈接Mysql數據庫操做
  4. RabbitMQ隊列
  5. Redis\Memcached緩存
  6. Paramiko SSH
  7. Twsited網絡框架

 

 

引子

到目前爲止，咱們已經學了網絡併發編程的2個套路， 多進程，多線程，這哥倆的優點和劣勢都很是的明顯，咱們一塊兒來回顧下python

 

 

 

協程

協程，又稱微線程，纖程。英文名Coroutine。一句話說明什麼是線程：協程是一種用戶態的輕量級線程。git

協程擁有本身的寄存器上下文和棧。協程調度切換時，將寄存器上下文和棧保存到其餘地方，在切回來的時候，恢復先前保存的寄存器上下文和棧。所以：程序員

協程能保留上一次調用時的狀態（即全部局部狀態的一個特定組合），每次過程重入時，就至關於進入上一次調用的狀態，換種說法：進入上一次離開時所處邏輯流的位置。github

 

協程的好處：sql

  • 無需線程上下文切換的開銷
  • 無需原子操做鎖定及同步的開銷
    • 　　"原子操做(atomic operation)是不須要synchronized"，所謂原子操做是指不會被線程調度機制打斷的操做；這種操做一旦開始，就一直運行到結束，中間不會有任何 context switch （切換到另外一個線程）。原子操做能夠是一個步驟，也能夠是多個操做步驟，可是其順序是不能夠被打亂，或者切割掉只執行部分。視做總體是原子性的核心。
  • 方便切換控制流，簡化編程模型
  • 高併發+高擴展性+低成本：一個CPU支持上萬的協程都不是問題。因此很適合用於高併發處理。

 

缺點：數據庫

  • 沒法利用多核資源：協程的本質是個單線程,它不能同時將 單個CPU 的多個核用上,協程須要和進程配合才能運行在多CPU上.固然咱們平常所編寫的絕大部分應用都沒有這個必要，除非是cpu密集型應用。
  • 進行阻塞（Blocking）操做（如IO時）會阻塞掉整個程序

使用yield實現協程操做例子　　　　編程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import  time
import  queue
def  consumer(name):
     print ( "--->starting eating baozi..." )
     while  True :
         new_baozi  =  yield
         print ( "[%s] is eating baozi %s"  %  (name,new_baozi))
         #time.sleep(1)
 
def  producer():
 
     r  =  con.__next__()
     r  =  con2.__next__()
     n  =  0
     while  n <  5 :
         n  + = 1
         con.send(n)
         con2.send(n)
         print ( "\033[32;1m[producer]\033[0m is making baozi %s"  % n )
 
 
if  __name__  = =  '__main__' :
     con  =  consumer( "c1" )
     con2  =  consumer( "c2" )
     p  =  producer()

看樓上的例子，我問你這算不算作是協程呢？你說，我他媽哪知道呀，你前面說了一堆廢話，可是並沒告訴我協程的標準形態呀，我腚眼一想，以爲你說也對，那好，咱們先給協程一個標準定義，即符合什麼條件就能稱之爲協程：數組

  1. 必須在只有一個單線程裏實現併發
  2. 修改共享數據不需加鎖
  3. 用戶程序裏本身保存多個控制流的上下文棧
  4. 一個協程遇到IO操做自動切換到其它協程

基於上面這4點定義，咱們剛纔用yield實現的程並不能算是合格的線程，由於它有一點功能沒實現，哪一點呢？緩存

 

 

Gevent 

Gevent 是一個第三方庫，能夠輕鬆經過gevent實現併發同步或異步編程，在gevent中用到的主要模式是Greenlet, 它是以C擴展模塊形式接入Python的輕量級協程。 Greenlet所有運行在主程序操做系統進程的內部，但它們被協做式地調度。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import  gevent
 
def  func1():
     print ( '\033[31;1m小闖在跟小濤搞...\033[0m' )
     gevent.sleep( 2 )
     print ( '\033[31;1m小闖又回去跟繼續跟小濤搞...\033[0m' )
 
def  func2():
     print ( '\033[32;1m小闖切換到了跟小龍搞...\033[0m' )
     gevent.sleep( 1 )
     print ( '\033[32;1m小闖搞完了小濤，回來繼續跟小龍搞...\033[0m' )
 
 
gevent.joinall([
     gevent.spawn(func1),
     gevent.spawn(func2),
     #gevent.spawn(func3),
])

　　

 

輸出：

小闖在跟小濤搞...
小闖切換到了跟小龍搞...
小闖搞完了小濤，回來繼續跟小龍搞...
小闖又回去跟繼續跟小濤搞...

 

 

 

同步與異步的性能區別 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import  gevent
 
def  task(pid):
     """
     Some non-deterministic task
     """
     gevent.sleep( 0.5 )
     print ( 'Task %s done'  %  pid)
 
def  synchronous():
     for  i  in  range ( 1 , 10 ):
         task(i)
 
def  asynchronous():
     threads  =  [gevent.spawn(task, i)  for  i  in  range ( 10 )]
     gevent.joinall(threads)
 
print ( 'Synchronous:' )
synchronous()
 
print ( 'Asynchronous:' )
asynchronous()

上面程序的重要部分是將task函數封裝到Greenlet內部線程的gevent.spawn。 初始化的greenlet列表存放在數組threads中，此數組被傳給gevent.joinall 函數，後者阻塞當前流程，並執行全部給定的greenlet。執行流程只會在 全部greenlet執行完後纔會繼續向下走。　　

遇到IO阻塞時會自動切換任務

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from  gevent  import  monkey; monkey.patch_all()
import  gevent
from   urllib.request  import  urlopen
 
def  f(url):
     print ( 'GET: %s'  %  url)
     resp  =  urlopen(url)
     data  =  resp.read()
     print ( '%d bytes received from %s.'  %  ( len (data), url))
 
gevent.joinall([
         gevent.spawn(f,  'https://www.python.org/' ),
         gevent.spawn(f,  'https://www.yahoo.com/' ),
         gevent.spawn(f,  'https://github.com/' ),
])

 

經過gevent實現單線程下的多socket併發

server side 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import  sys
import  socket
import  time
import  gevent
 
from  gevent  import  socket,monkey
monkey.patch_all()
 
 
def  server(port):
     s  =  socket.socket()
     s.bind(( '0.0.0.0' , port))
     s.listen( 500 )
     while  True :
         cli, addr  =  s.accept()
         gevent.spawn(handle_request, cli)
 
 
 
def  handle_request(conn):
     try :
         while  True :
             data  =  conn.recv( 1024 )
             print ( "recv:" , data)
             conn.send(data)
             if  not  data:
                 conn.shutdown(socket.SHUT_WR)
 
     except  Exception as  ex:
         print (ex)
     finally :
         conn.close()
if  __name__  = =  '__main__' :
     server( 8001 )

　　

client side 　　

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import  socket
 
HOST  =  'localhost'     # The remote host
PORT  =  8001            # The same port as used by the server
s  =  socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((HOST, PORT))
while  True :
     msg  =  bytes( input ( ">>:" ),encoding = "utf8" )
     s.sendall(msg)
     data  =  s.recv( 1024 )
     #print(data)
 
     print ( 'Received' ,  repr (data))
s.close()
複製代碼 
import socket
import threading

def sock_conn():

    client = socket.socket()

    client.connect(("localhost",8001))
    count = 0
    while True:
        #msg = input(">>:").strip()
        #if len(msg) == 0:continue
        client.send( ("hello %s" %count).encode("utf-8"))

        data = client.recv(1024)

        print("[%s]recv from server:" % threading.get_ident(),data.decode()) #結果
        count +=1
    client.close()


for i in range(100):
    t = threading.Thread(target=sock_conn)
    t.start()
複製代碼

 

 

　　

論事件驅動與異步IO

一般，咱們寫服務器處理模型的程序時，有如下幾種模型：
（1）每收到一個請求，建立一個新的進程，來處理該請求；
（2）每收到一個請求，建立一個新的線程，來處理該請求；
（3）每收到一個請求，放入一個事件列表，讓主進程經過非阻塞I/O方式來處理請求
上面的幾種方式，各有千秋，
第（1）中方法，因爲建立新的進程的開銷比較大，因此，會致使服務器性能比較差,但實現比較簡單。
第（2）種方式，因爲要涉及到線程的同步，有可能會面臨 死鎖等問題。
第（3）種方式，在寫應用程序代碼時，邏輯比前面兩種都複雜。
綜合考慮各方面因素，通常廣泛認爲第（3）種方式是大多數 網絡服務器採用的方式
 

看圖說話講事件驅動模型

在UI編程中，經常要對鼠標點擊進行相應，首先如何得到鼠標點擊呢？
方式一：建立一個線程，該線程一直循環檢測是否有鼠標點擊，那麼這個方式有如下幾個缺點：
1. CPU資源浪費，可能鼠標點擊的頻率很是小，可是掃描線程仍是會一直循環檢測，這會形成不少的CPU資源浪費；若是掃描鼠標點擊的接口是阻塞的呢？
2. 若是是堵塞的，又會出現下面這樣的問題，若是咱們不但要掃描鼠標點擊，還要掃描鍵盤是否按下，因爲掃描鼠標時被堵塞了，那麼可能永遠不會去掃描鍵盤；
3. 若是一個循環須要掃描的設備很是多，這又會引來響應時間的問題；
因此，該方式是很是很差的。

方式二：就是事件驅動模型
目前大部分的UI編程都是事件驅動模型，如不少UI平臺都會提供onClick()事件，這個事件就表明鼠標按下事件。事件驅動模型大致思路以下：
1. 有一個事件（消息）隊列；
2. 鼠標按下時，往這個隊列中增長一個點擊事件（消息）；
3. 有個循環，不斷從隊列取出事件，根據不一樣的事件，調用不一樣的函數，如onClick()、onKeyDown()等；
4. 事件（消息）通常都各自保存各自的處理函數指針，這樣，每一個消息都有獨立的處理函數；

 

 

 

事件驅動編程是一種編程範式，這裏程序的執行流由外部事件來決定。它的特色是包含一個事件循環，當外部事件發生時使用回調機制來觸發相應的處理。另外兩種常見的編程範式是（單線程）同步以及多線程編程。

讓咱們用例子來比較和對比一下單線程、多線程以及事件驅動編程模型。下圖展現了隨着時間的推移，這三種模式下程序所作的工做。這個程序有3個任務須要完成，每一個任務都在等待I/O操做時阻塞自身。阻塞在I/O操做上所花費的時間已經用灰色框標示出來了。

 

在單線程同步模型中，任務按照順序執行。若是某個任務由於I/O而阻塞，其餘全部的任務都必須等待，直到它完成以後它們才能依次執行。這種明確的執行順序和串行化處理的行爲是很容易推斷得出的。若是任務之間並無互相依賴的關係，但仍然須要互相等待的話這就使得程序沒必要要的下降了運行速度。

在多線程版本中，這3個任務分別在獨立的線程中執行。這些線程由操做系統來管理，在多處理器系統上能夠並行處理，或者在單處理器系統上交錯執行。這使得當某個線程阻塞在某個資源的同時其餘線程得以繼續執行。與完成相似功能的同步程序相比，這種方式更有效率，但程序員必須寫代碼來保護共享資源，防止其被多個線程同時訪問。多線程程序更加難以推斷，由於這類程序不得不經過線程同步機制如鎖、可重入函數、線程局部存儲或者其餘機制來處理線程安全問題，若是實現不當就會致使出現微妙且使人痛不欲生的bug。

在事件驅動版本的程序中，3個任務交錯執行，但仍然在一個單獨的線程控制中。當處理I/O或者其餘昂貴的操做時，註冊一個回調到事件循環中，而後當I/O操做完成時繼續執行。回調描述了該如何處理某個事件。事件循環輪詢全部的事件，當事件到來時將它們分配給等待處理事件的回調函數。這種方式讓程序儘量的得以執行而不須要用到額外的線程。事件驅動型程序比多線程程序更容易推斷出行爲，由於程序員不須要關心線程安全問題。

當咱們面對以下的環境時，事件驅動模型一般是一個好的選擇：

  1. 程序中有許多任務，並且…
  2. 任務之間高度獨立（所以它們不須要互相通訊，或者等待彼此）並且…
  3. 在等待事件到來時，某些任務會阻塞。

當應用程序須要在任務間共享可變的數據時，這也是一個不錯的選擇，由於這裏不須要採用同步處理。

網絡應用程序一般都有上述這些特色，這使得它們可以很好的契合事件驅動編程模型。

 

此處要提出一個問題，就是，上面的事件驅動模型中，只要一遇到IO就註冊一個事件，而後主程序就能夠繼續幹其它的事情了，只到io處理完畢後，繼續恢復以前中斷的任務，這本質上是怎麼實現的呢？下面咱們就來一塊兒揭開這神祕的面紗。。。。

 

Select\Poll\Epoll異步IO　

http://www.cnblogs.com/alex3714/p/4372426.html　

番外篇 http://www.cnblogs.com/alex3714/articles/5876749.html 

select 多併發socket 例子

複製代碼 
#_*_coding:utf-8_*_
__author__ = 'Alex Li'

import select
import socket
import sys
import queue


server = socket.socket()
server.setblocking(0)

server_addr = ('localhost',10000)

print('starting up on %s port %s' % server_addr)
server.bind(server_addr)

server.listen(5)


inputs = [server, ] #本身也要監測呀,由於server自己也是個fd
outputs = []

message_queues = {}

while True:
    print("waiting for next event...")

    readable, writeable, exeptional = select.select(inputs,outputs,inputs) #若是沒有任何fd就緒,那程序就會一直阻塞在這裏

    for s in readable: #每一個s就是一個socket

        if s is server: #別忘記,上面咱們server本身也當作一個fd放在了inputs列表裏,傳給了select,若是這個s是server,表明server這個fd就緒了,
            #就是有活動了, 什麼狀況下它纔有活動? 固然 是有新鏈接進來的時候 呀
            #新鏈接進來了,接受這個鏈接
            conn, client_addr = s.accept()
            print("new connection from",client_addr)
            conn.setblocking(0)
            inputs.append(conn) #爲了避免阻塞整個程序,咱們不會馬上在這裏開始接收客戶端發來的數據, 把它放到inputs裏, 下一次loop時,這個新鏈接
            #就會被交給select去監聽,若是這個鏈接的客戶端發來了數據 ,那這個鏈接的fd在server端就會變成就續的,select就會把這個鏈接返回,返回到
            #readable 列表裏,而後你就能夠loop readable列表,取出這個鏈接,開始接收數據了, 下面就是這麼幹 的

            message_queues[conn] = queue.Queue() #接收到客戶端的數據後,不馬上返回 ,暫存在隊列裏,之後發送

        else: #s不是server的話,那就只能是一個 與客戶端創建的鏈接的fd了
            #客戶端的數據過來了,在這接收
            data = s.recv(1024)
            if data:
                print("收到來自[%s]的數據:" % s.getpeername()[0], data)
                message_queues[s].put(data) #收到的數據先放到queue裏,一會返回給客戶端
                if s not  in outputs:
                    outputs.append(s) #爲了避免影響處理與其它客戶端的鏈接 , 這裏不馬上返回數據給客戶端


            else:#若是收不到data表明什麼呢? 表明客戶端斷開了呀
                print("客戶端斷開了",s)

                if s in outputs:
                    outputs.remove(s) #清理已斷開的鏈接

                inputs.remove(s) #清理已斷開的鏈接

                del message_queues[s] ##清理已斷開的鏈接


    for s in writeable:
        try :
            next_msg = message_queues[s].get_nowait()

        except queue.Empty:
            print("client [%s]" %s.getpeername()[0], "queue is empty..")
            outputs.remove(s)

        else:
            print("sending msg to [%s]"%s.getpeername()[0], next_msg)
            s.send(next_msg.upper())


    for s in exeptional:
        print("handling exception for ",s.getpeername())
        inputs.remove(s)
        if s in outputs:
            outputs.remove(s)
        s.close()

        del message_queues[s]
複製代碼
複製代碼 
#_*_coding:utf-8_*_
__author__ = 'Alex Li'


import socket
import sys

messages = [ b'This is the message. ',
             b'It will be sent ',
             b'in parts.',
             ]
server_address = ('localhost', 10000)

# Create a TCP/IP socket
socks = [ socket.socket(socket.AF_INET, socket.SOCK_STREAM),
          socket.socket(socket.AF_INET, socket.SOCK_STREAM),
          ]

# Connect the socket to the port where the server is listening
print('connecting to %s port %s' % server_address)
for s in socks:
    s.connect(server_address)

for message in messages:

    # Send messages on both sockets
    for s in socks:
        print('%s: sending "%s"' % (s.getsockname(), message) )
        s.send(message)

    # Read responses on both sockets
    for s in socks:
        data = s.recv(1024)
        print( '%s: received "%s"' % (s.getsockname(), data) )
        if not data:
            print(sys.stderr, 'closing socket', s.getsockname() )
複製代碼

 

 

selectors模塊

This module allows high-level and efficient I/O multiplexing, built upon the select module primitives. Users are encouraged to use this module instead, unless they want precise control over the OS-level primitives used.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import  selectors
import  socket
 
sel  =  selectors.DefaultSelector()
 
def  accept(sock, mask):
     conn, addr  =  sock.accept()   # Should be ready
     print ( 'accepted' , conn,  'from' , addr)
     conn.setblocking( False )
     sel.register(conn, selectors.EVENT_READ, read)
 
def  read(conn, mask):
     data  =  conn.recv( 1000 )   # Should be ready
     if  data:
         print ( 'echoing' ,  repr (data),  'to' , conn)
         conn.send(data)   # Hope it won't block
     else :
         print ( 'closing' , conn)
         sel.unregister(conn)
         conn.close()
 
sock  =  socket.socket()
sock.bind(( 'localhost' ,  10000 ))
sock.listen( 100 )
sock.setblocking( False )
sel.register(sock, selectors.EVENT_READ, accept)
 
while  True :
     events  =  sel.select()
     for  key, mask  in  events:
         callback  =  key.data
         callback(key.fileobj, mask)

　　

數據庫操做與Paramiko模塊 

http://www.cnblogs.com/wupeiqi/articles/5095821.html 

 

 

RabbitMQ隊列　　

安裝 http://www.rabbitmq.com/install-standalone-mac.html

安裝python rabbitMQ module 

1
2
3
4
5
6
7
pip install pika
or
easy_install pika
or
源碼
  
https: / / pypi.python.org / pypi / pika

實現最簡單的隊列通訊

 

send端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
#!/usr/bin/env python
import  pika
 
connection  =  pika.BlockingConnection(pika.ConnectionParameters(
                'localhost' ))
channel  =  connection.channel()
 
#聲明queue
channel.queue_declare(queue = 'hello' )
 
#n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
channel.basic_publish(exchange = '',
                       routing_key = 'hello' ,
                       body = 'Hello World!' )
print ( " [x] Sent 'Hello World!'" )
connection.close()

receive端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
#_*_coding:utf-8_*_
__author__  =  'Alex Li'
import  pika
 
connection  =  pika.BlockingConnection(pika.ConnectionParameters(
                'localhost' ))
channel  =  connection.channel()
 
 
#You may ask why we declare the queue again ‒ we have already declared it in our previous code.
# We could avoid that if we were sure that the queue already exists. For example if send.py program
#was run before. But we're not yet sure which program to run first. In such cases it's a good
# practice to repeat declaring the queue in both programs.
channel.queue_declare(queue = 'hello' )
 
def  callback(ch, method, properties, body):
     print ( " [x] Received %r"  %  body)
 
channel.basic_consume(callback,
                       queue = 'hello' ,
                       no_ack = True )
 
print ( ' [*] Waiting for messages. To exit press CTRL+C' )
channel.start_consuming()

　　

Work Queues

在這種模式下，RabbitMQ會默認把p發的消息依次分發給各個消費者(c),跟負載均衡差很少

消息提供者代碼

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import  pika
import  time
connection  =  pika.BlockingConnection(pika.ConnectionParameters(
     'localhost' ))
channel  =  connection.channel()
 
# 聲明queue
channel.queue_declare(queue = 'task_queue' )
 
# n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
import  sys
 
message  =  ' ' .join(sys.argv[ 1 :])  or  "Hello World! %s"  %  time.time()
channel.basic_publish(exchange = '',
                       routing_key = 'task_queue' ,
                       body = message,
                       properties = pika.BasicProperties(
                           delivery_mode = 2 ,   # make message persistent
                       )
                       )
print ( " [x] Sent %r"  %  message)
connection.close()

　　

 

消費者代碼

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
#_*_coding:utf-8_*_
 
import  pika, time
 
connection  =  pika.BlockingConnection(pika.ConnectionParameters(
     'localhost' ))
channel  =  connection.channel()
 
 
def  callback(ch, method, properties, body):
     print ( " [x] Received %r"  %  body)
     time.sleep( 20 )
     print ( " [x] Done" )
     print ( "method.delivery_tag" ,method.delivery_tag)
     ch.basic_ack(delivery_tag = method.delivery_tag)
 
 
channel.basic_consume(callback,
                       queue = 'task_queue' ,
                       no_ack = True
                       )
 
print ( ' [*] Waiting for messages. To exit press CTRL+C' )
channel.start_consuming()

　　

 

此時，先啓動消息生產者，而後再分別啓動3個消費者，經過生產者多發送幾條消息，你會發現，這幾條消息會被依次分配到各個消費者身上　　

Doing a task can take a few seconds. You may wonder what happens if one of the consumers starts a long task and dies with it only partly done. With our current code once RabbitMQ delivers message to the customer it immediately removes it from memory. In this case, if you kill a worker we will lose the message it was just processing. We'll also lose all the messages that were dispatched to this particular worker but were not yet handled.

But we don't want to lose any tasks. If a worker dies, we'd like the task to be delivered to another worker.

In order to make sure a message is never lost, RabbitMQ supports message acknowledgments. An ack(nowledgement) is sent back from the consumer to tell RabbitMQ that a particular message had been received, processed and that RabbitMQ is free to delete it.

If a consumer dies (its channel is closed, connection is closed, or TCP connection is lost) without sending an ack, RabbitMQ will understand that a message wasn't processed fully and will re-queue it. If there are other consumers online at the same time, it will then quickly redeliver it to another consumer. That way you can be sure that no message is lost, even if the workers occasionally die.

There aren't any message timeouts; RabbitMQ will redeliver the message when the consumer dies. It's fine even if processing a message takes a very, very long time.

Message acknowledgments are turned on by default. In previous examples we explicitly turned them off via the no_ack=True flag. It's time to remove this flag and send a proper acknowledgment from the worker, once we're done with a task.

1
2
3
4
5
6
7
8
def  callback(ch, method, properties, body):
     print  " [x] Received %r"  %  (body,)
     time.sleep( body.count( '.' ) )
     print  " [x] Done"
     ch.basic_ack(delivery_tag  =  method.delivery_tag)
 
channel.basic_consume(callback,
                       queue = 'hello' )

　　Using this code we can be sure that even if you kill a worker using CTRL+C while it was processing a message, nothing will be lost. Soon after the worker dies all unacknowledged messages will be redelivered

　　　　

消息持久化　　

We have learned how to make sure that even if the consumer dies, the task isn't lost(by default, if wanna disable  use no_ack=True). But our tasks will still be lost if RabbitMQ server stops.

When RabbitMQ quits or crashes it will forget the queues and messages unless you tell it not to. Two things are required to make sure that messages aren't lost: we need to mark both the queue and messages as durable.

First, we need to make sure that RabbitMQ will never lose our queue. In order to do so, we need to declare it as durable:

1
channel.queue_declare(queue = 'hello' , durable = True )

　　

Although this command is correct by itself, it won't work in our setup. That's because we've already defined a queue called hello which is not durable. RabbitMQ doesn't allow you to redefine an existing queue with different parameters and will return an error to any program that tries to do that. But there is a quick workaround - let's declare a queue with different name, for exampletask_queue:

1
channel.queue_declare(queue = 'task_queue' , durable = True )

　　

This queue_declare change needs to be applied to both the producer and consumer code.

At that point we're sure that the task_queue queue won't be lost even if RabbitMQ restarts. Now we need to mark our messages as persistent - by supplying a delivery_mode property with a value 2.

1
2
3
4
5
6
channel.basic_publish(exchange = '',
                       routing_key = "task_queue" ,
                       body = message,
                       properties = pika.BasicProperties(
                          delivery_mode  =  2 ,  # make message persistent
                       ))

消息公平分發

若是Rabbit只管按順序把消息發到各個消費者身上，不考慮消費者負載的話，極可能出現，一個機器配置不高的消費者那裏堆積了不少消息處理不完，同時配置高的消費者卻一直很輕鬆。爲解決此問題，能夠在各個消費者端，配置perfetch=1,意思就是告訴RabbitMQ在我這個消費者當前消息還沒處理完的時候就不要再給我發新消息了。

 

1
channel.basic_qos(prefetch_count = 1 )

 

帶消息持久化+公平分發的完整代碼

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程