RabbitMQ廣播模式

廣播模式：1對多，produce發送一則消息多個consumer同時收到。
注意：廣播是實時的，produce只負責發出去，不會管對端是否收到，若發送的時刻沒有對端接收，那消息就沒了，所以在廣播模式下設置消息持久化是無效的。

三種廣播模式：

fanout: 全部bind到此exchange的queue均可以接收消息（純廣播，綁定到RabbitMQ的接受者都能收到消息）；
direct: 經過routingKey和exchange決定的那個惟一的queue能夠接收消息；
topic:全部符合routingKey(此時能夠是一個表達式)的routingKey所bind的queue能夠接收消息；

原理：

fanout：接受者會在RabbitMQ中建立一個queue用來接收消息，發送者往queue中發送消息。（發送給全員）

direct：隊列綁定關鍵字，發送者將數據根據關鍵字發送到消息exchange，exchange根據 關鍵字 斷定應該將數據發送至指定隊列。（發送給指定人）

topic：接受者能夠收取指定內容的消息

代碼飯粒1：fanout廣播模式

# publisher發送者

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='broadcast', exchange_type='fanout')
message = 'big bang!'
channel.basic_publish(
    exchange='broadcast',
    routing_key='',
    body=message,
)
print("[v] Send %r" % message)
connection.close()

 # subscriber接收者

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='broadcast', exchange_type='fanout')

result = channel.queue_declare(exclusive=True)
# 不指定queue名字，Rabbit會隨機分配一個名字，並在使用此queue的消費者斷開後,自動將queue刪除
queue_name = result.method.queue
channel.queue_bind(exchange='broadcast',queue=queue_name)
print(" [*] Waiting for broadcast. To exit press Ctrl+C")

def callback(ch, method, properties, body):
    print(" [v] Get broadcast:",body)

channel.basic_consume(
    callback,
    queue=queue_name,
)
channel.start_consuming()

 

代碼飯粒2：direct廣播模式

# publisher發送者

import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(
    exchange='direct_logs',
    routing_key=severity,
    body=message
)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()

# subscriber接收者

import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

severities = sys.argv[1:]
if not severities:
    sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
    sys.exit(1)

for severity in severities:
    channel.queue_bind(
        exchange='direct_logs',
        queue=queue_name,
        routing_key=severity
    )
print(' [*] Waiting for logs. To exit press CTRL+C')

def callback(ch, method, properties, body):
    print(" [x] %r:%r" % (method.routing_key, body))

channel.basic_consume(
    callback,
    queue=queue_name,
    no_ack=True
)
channel.start_consuming()

 

代碼飯粒3：topic廣播模式

# publisher發送者

import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs', exchange_type='topic')

routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(
    exchange='topic_logs',
    routing_key=routing_key,
    body=message
)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()

# subscriber接收者

import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs', exchange_type='topic')

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

binding_keys = sys.argv[1:]
if not binding_keys:
    sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
    sys.exit(1)

for binding_key in binding_keys:
    channel.queue_bind(
        exchange='topic_logs',
        queue=queue_name,
        routing_key=binding_key
    )
print(' [*] Waiting for logs. To exit press CTRL+C')

def callback(ch, method, properties, body):
    print(" [x] %r:%r" % (method.routing_key, body))

channel.basic_consume(
    callback,
    queue=queue_name,
    no_ack=True
)
channel.start_consuming()