Python RabbitMQ消息队列

In |

WechatIMG91.jpeg

环境

  • Ubuntu 16.04.1

  • Python 3.5.2

安装RabbitMQServer

  • 更新软件源
1
$ apt-get update
  • 安装 rabbitmq-server
1
$ apt-get install -y rabbitmq-server
  • 安装 pika
1
$ pip3 install pika

用来实现 python 与 rabbitmq 的对接。

管理RabbitMQServer

  • 启动
1
$ service rabbitmq-server start
  • 关闭
1
$ service rabbitmq-server stop
  • 查看状态
1
$ service rabbitmq-server status
  • 调整系统限制
1
2
3
4
5
6
7
8
9
10
$ cat /etc/default/rabbitmq-server
#  This file is sourced by /etc/init.d/rabbitmq-server. Its primary
#  reason for existing is to allow adjustment of system limits for the
#  rabbitmq-server process.
# 
#  Maximum number of open file handles. This will need to be increased
#  to handle many simultaneous connections. Refer to the system
#  documentation for ulimit (in man bash) for more information.
# 
ulimit -n 1024

编辑 /etc/default/rabbitmq-server 中的 ulimit 参数，默认1024。

  • 日志
1
2
3
$ ls /var/log/rabbitmq
rabbit@VM-2-14-ubuntu.log       startup_err
rabbit@VM-2-14-ubuntu-sasl.log  startup_log

工作队列模式

61e677960a9dc_61e677960a9d6.png

  • 创建 new_task.py，代码如下
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# !/usr/bin/env python3
import pika
import sys

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

#  持久化一个队列，名为 task_queue
channel.queue_declare(queue='task_queue', durable=True)

message = ' '.join(sys.argv[1:]) or "No new task"
channel.basic_publish(exchange='',
                      routing_key='task_queue',
                      body=message,
                      properties=pika.BasicProperties(
                         #  消息持久化
                         delivery_mode = 2,
                      ))
print(" [x] Sent %r" % message)

#  关闭连接
connection.close()
  • 创建 worker.py，代码如下
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# !/usr/bin/env python3
import pika
import time

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

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

print(" [*] Waiting for messages. To exit press CTRL+C")

def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    time.sleep(body.count(b'.') )
    print(" [x] Done")
    ch.basic_ack(delivery_tag = method.delivery_tag)

channel.basic_qos(prefetch_count=1)

channel.basic_consume(queue='task_queue',
                      on_message_callback=callback)

channel.start_consuming()
  • 执行
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
#  终端1，发布新任务
$ python3 new_task.py eating
 [x] Sent 'eating'
$ python3 new_task.py drinking
 [x] Sent 'drinking'
$ python3 new_task.py playing
 [x] Sent 'playing'
 
#  终端2，添加工作者
$ python3 worker.py
 [*] Waiting for messages. To exit press CTRL+C
 [x] Received b'eating'
 [x] Done
 [x] Received b'drinking'
 [x] Done
 [x] Received b'playing'
 [x] Done

扇形交换机

61e686ab09d66_61e686ab09d5b.png

  • 创建 send_log.py，代码如下
1
2
3
4
5
6
7
8
9
10
11
12
13
14
# !/usr/bin/env python3
import pika
import sys

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

channel.exchange_declare(exchange='logs', exchange_type='fanout')

message = ' '.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange='logs', routing_key='', body=message)
print(" [x] Sent %r" % message)
connection.close()
  • 创建 receive_logs.py，代码如下
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
# !/usr/bin/env python3
import pika

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

channel.exchange_declare(exchange='logs', exchange_type='fanout')

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

channel.queue_bind(exchange='logs', queue=queue_name)

print(' [*] Waiting for logs. To exit press CTRL+C')


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


channel.basic_consume(
    queue=queue_name, on_message_callback=callback, auto_ack=True)

channel.start_consuming()
  • 执行
1
2
3
4
5
6
7
8
9
10
11
12
13
#  终端1，接收日志
$ python3 receive_logs.py
 [*] Waiting for logs. To exit press CTRL+C
 [x] b'eating'

#  终端2，接收日志
$ python3 receive_logs.py
 [*] Waiting for logs. To exit press CTRL+C
 [x] b'eating'

#  终端3，发送日志
$ python3 new_task.py eating
 [x] Sent 'eating'

直连交换机

61e68ad22bf39_61e68ad22bf33.png

  • 创建 send_log_direct.py，代码如下
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# !/usr/bin/env python3
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()
  • 创建 receive_logs_direct.py，代码如下
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
# !/usr/bin/env python3

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(queue='', 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(
    queue=queue_name, on_message_callback=callback, auto_ack=True)

channel.start_consuming()
  • 执行
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
#  终端1，接收日志
$ python3 receive_logs_direct.py error
 [*] Waiting for logs. To exit press CTRL+C
 [x] 'error':b'playing

#  终端2，接收日志
$ python3 receive_logs_direct.py info
 [*] Waiting for logs. To exit press CTRL+C
 [x] 'info':b'watching'

#  终端3，发送日志
$ python3 send_log_direct.py info watching
 [x] Sent 'info':'watching'
$ python3 send_log_direct.py error playing
 [x] Sent 'error':'playing'

主题交换机

61e690149f79f_61e690149f79a.png

  • 创建 send_log_topic.py，代码如下
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# !/usr/bin/env python3
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) > 2 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()
  • 创建 receive_logs_topic.py，代码如下
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
# !/usr/bin/env python3
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(
    queue=queue_name, on_message_callback=callback, auto_ack=True)

channel.start_consuming()
  • 执行
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
#  终端1，发送日志
$ python3 send_log_topic.py "quick.orange.rabbit"
 [x] Sent 'quick.orange.rabbit':"quick.orange.rabbit'
$ python3 send_log_topic.py "lazy.pink.rabbit"
 [x] Sent 'quick.orange.rabbit':'lazy.pink.rabbit'
 
#  终端2，接收日志
$ python3 receive_logs_topic.py "*.orange.*"
 [*] Waiting for logs. To exit press CTRL+C
 [x] 'quick.orange.rabbit':b'lazy.pink.rabbit'
 
#  终端3，接收日志
$ python3 receive_logs_topic.py "*.*.rabbit"
 [*] Waiting for logs. To exit press CTRL+C
 [x] 'quick.orange.rabbit':b'lazy.pink.rabbit'
-------------本文结束感谢您的阅读-------------
0%