RabbitMQ簡單生產者消費者(一對一)發送消息

概述

　　RabbitMQ是一種消息隊列,它接收並轉發消息。

　　官方例子：可以把RabbitMQ視為一個郵局,將要發布的郵件放在郵箱中，通過郵遞員傳遞給收件人。

　　但是又有區別二者：RabbitMQ不處理，只做接收，存儲和轉發數據消息的中間介質

實現最簡單的隊列通信

producer_sender.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import pika

credentials = pika.PlainCredentials('admin', 'admin123456') connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.1.13', credentials=credentials))  8 # 建立通道
channel = connection.channel()
# 如果將消息發送到不存在的位置,RabbitMQ會刪除該消息,創建消息隊列叫hello
channel.queue_declare(queue='hello')
# 發消息了,在RabbitMQ中永遠無法將消息直接發送到隊列中,它始終需要進行交換(不理解這是為毛線?)
# 在這里使用空字符串標識的默認交換,准確的將指定消息放入隊列中routing_key來指定
channel.basic_publish(exchange='',
                      routing_key='hello',
                      body='Hello World!')

print('[x]發送hello word')
# 在關閉連接之前,需要確保緩沖區已刷新消息是否已經傳到消息隊列中
connection.close()

consumer_recv.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import pika

credentials = pika.PlainCredentials('admin', 'admin123456') connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.1.13', credentials=credentials))  8 channel = connection.channel()
#  在此再次申明一次是因為當我們不確定運行哪個程序,在這種情況下,最好在兩個程序中重復聲明隊列
channel.queue_declare('hello')


def callback(ch, method, properties, body):
    """收消息原理:向隊列定義一個回調函數,無論何時接收消息,都由Pika庫調用"""
    print('[x] Received %r' % body)
    print(ch)
    print(method)
    print(properties)


# 告訴rabbitmq這個特定的回調函數應該接收來自hello隊列的消息
channel.basic_consume(queue='hello',
                      auto_ack=True,
                      on_message_callback=callback)
print(' [*] Waiting for messages.')
# 這里只要消費者一直存在通道之上，就一直死循環,源碼當中有說明
channel.start_consuming()

這里注意首次運行要報錯,報錯內容如下（因為我在這里是遠程連接需要認證證書）

在此時就要看下源碼參數設置

def __init__( # pylint: disable=R0913,R0914
        self,
        host=_DEFAULT,  # 默認'localhost'
        port=_DEFAULT,  # 5672
        virtual_host=_DEFAULT, # 使用rabbitmq虛擬主機,源碼中還做了一次判斷
            if virtual_host is not self._DEFAULT:
                self.virtual_host = virtual_host
        credentials=_DEFAULT, # auth憑證
        channel_max=_DEFAULT, # 允許的最大的通道數
        frame_max=_DEFAULT,   # AMQP幀的最大字節大小 數據鏈路層的最小傳輸單位稱為幀 為什么要有幀？就是為了保證數據的可靠傳輸把數據封裝成幀了
        heartbeat=_DEFAULT,   # 默認是None 
        ssl_options=_DEFAULT, # 默認是None
        connection_attempts=_DEFAULT, #最大重試連接次數默認1次
        retry_delay=_DEFAULT, # 在幾秒鍾內就要等待，在之后的那一刻就不等待了 默認2.0秒
        socket_timeout=_DEFAULT, # socket連接超時 默認10s
        stack_timeout=_DEFAULT,  # (TCP/[SSL]/AMQP)協議棧超時 建議要比socket_timeout大 默認15s
        locale=_DEFAULT,         # 默認'en_US'
        blocked_connection_timeout=_DEFAULT, #默認None 如果不是None,連接保持阻塞(由連接觸發)。阻止代理);如果超時在連接解除阻塞之前過期，連接將被斷開
        client_properties=_DEFAULT,  # 默認None 或dict的客戶端屬性使用覆蓋報告給的默認客戶端屬性中的字段RabbitMQ通過“Connection.StartOk”方法。
        tcp_options=_DEFAULT,        # 默認None 為套接字設置一個或沒有TCP選項的dict
        **kwargs):
    """Create a new ConnectionParameters instance. See `Parameters` for
    default values.

    :param str host: Hostname or IP Address to connect to
    :param int port: TCP port to connect to
    :param str virtual_host: RabbitMQ virtual host to use
    :param pika.credentials.Credentials credentials: auth credentials
    :param int channel_max: Maximum number of channels to allow
    :param int frame_max: The maximum byte size for an AMQP frame
    :param int|None|callable heartbeat: Controls AMQP heartbeat timeout negotiation
        during connection tuning. An integer value always overrides the value
        proposed by broker. Use 0 to deactivate heartbeats and None to always
        accept the broker's proposal. If a callable is given, it will be called
        with the connection instance and the heartbeat timeout proposed by broker
        as its arguments. The callback should return a non-negative integer that
        will be used to override the broker's proposal.
    :param `pika.SSLOptions`|None ssl_options: None for plaintext or
        `pika.SSLOptions` instance for SSL/TLS. Defaults to None.
    :param int connection_attempts: Maximum number of retry attempts
    :param int|float retry_delay: Time to wait in seconds, before the next
    :param int|float socket_timeout: Positive socket connect timeout in
        seconds.
    :param int|float stack_timeout: Positive full protocol stack
        (TCP/[SSL]/AMQP) bring-up timeout in seconds. It's recommended to
        set this value higher than `socket_timeout`.
    :param str locale: Set the locale value
    :param int|float|None blocked_connection_timeout: If not None,
        the value is a non-negative timeout, in seconds, for the
        connection to remain blocked (triggered by Connection.Blocked from
        broker); if the timeout expires before connection becomes unblocked,
        the connection will be torn down, triggering the adapter-specific
        mechanism for informing client app about the closed connection:
        passing `ConnectionBlockedTimeout` exception to on_close_callback
        in asynchronous adapters or raising it in `BlockingConnection`.
    :param client_properties: None or dict of client properties used to
        override the fields in the default client properties reported to
        RabbitMQ via `Connection.StartOk` method.
    :param tcp_options: None or a dict of TCP options to set for socket
    """

默認是連接的localhost主機,需要從新增加用戶並設置權限即可

運行結果

D:\python\python.exe F:/abc/messagequeue/consumer_recv.py
 [*] Waiting for messages.
[x] Received b'Hello World!'
<BlockingChannel impl=<Channel number=1 OPEN conn=<SelectConnection OPEN transport=<pika.adapters.utils.io_services_utils._AsyncPlaintextTransport object at 0x00000171F0AF17B8> params=<ConnectionParameters host=192.168.1.13 port=5672 virtual_host=/ ssl=False>>>>
<Basic.Deliver(['consumer_tag=ctag1.1830cf91aad94871b0ed054af1f71e3d', 'delivery_tag=1', 'exchange=', 'redelivered=False', 'routing_key=hello'])>
<BasicProperties>