1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
安装配置epel源:(详见http:
/
/
www.cnblogs.com
/
ernest
-
zhang
/
p
/
5714434.html
)
安装erlang:
yum
-
y install erlang
注:安装erlang的时候碰到
Error: Package: erlang
-
erts
-
R14B
-
04.3
.el6.i686 (epel)
Requires: libz.so.
1
(ZLIB_1.
2.2
)
[root@localhost ~]
# yum whatprovides libz.so.1
Loaded plugins: rhnplugin
This system
is
not
registered with RHN.
RHN support will be disabled.
zlib
-
1.2
.
3
-
25.el6
.i686 : The zlib compression
and
decompression library
#提供压缩与解压缩库
Repo : local
Matched
from
:
Other : libz.so.
1
检查发现应该是zlib的版本太老了,从网上下载最新的zlib
-
1.2
.
8
-
10.fc24
.i686,然后使用RPM安装后解决。
下载地址:http:
/
/
www.zlib.net
/
#zlib官网
http:
/
/
rpmfind.net
/
linux
/
rpm2html
/
search.php?query
=
zlib
#zlib下载网站
安装rabbitMQ:
yum
-
y install rabbitmq
-
server
|
1
2
3
4
5
6
7
|
pip install pika
or
easy_install pika
or
源码
https:
/
/
pypi.python.org
/
pypi
/
pika
|
1
2
3
4
5
6
7
|
import
pika
connection
=
pika.BlockingConnection(pika.ConnectionParameters(host
=
'192.168.0.74'
))
#服务器地址
channel
=
connection.channel()
channel.queue_declare(queue
=
'Hi'
)
#如果有队列,略过;如果没有,创建队列
channel.basic_publish(exchange
=
'
',routing_key='
cc
',body='
hello!world!!!')
print
(
"[x] sent 'hello,world!'"
)
connection.close()
|
接收端:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
import
pika
#创建一个连接对象,绑定rabbitmq的IP
connection
=
pika.BlockingConnection(pika.ConnectionParameters(host
=
'192.168.0.74'
))
#创建一个频道对象
channel
=
connection.channel()
#频道中声明指定queue,如果MQ中没有指定queue就创建,如果有,则略过
channel.queue_declare(queue
=
'Hi'
)
#定义回调函数
def
callback(ch,method,properties,body):
print
(
'[x] Recieved %r'
%
body)
# channel.close()
#no_ack=Fales:表示消费完以后不主动把状态通知rabbitmq,callback:回调函数,queue:指定队列
channel.basic_consume(callback,queue
=
'Hi'
,no_ack
=
True
)
# channel.basic_consume(callback,queue='cc')
print
(
'[*] Waiting for msg'
)
channel.start_consuming()
|
1、acknowledgment 消息不丢失
no-ack = False,如果消费者遇到情况(its channel is closed, connection is closed, or TCP connection is lost)挂掉了,那么,RabbitMQ会重新将该任务添加到队列中。
- 回调函数中的
ch.basic_ack(delivery_tag=method.delivery_tag)
- basic_comsume中的
no_ack=False
import pika connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.0.74')) channel = connection.channel() channel.queue_declare(queue='Hi') # 定义回调函数 def callback(ch, method, properties, body): print('[x] Recieved %r' % body) # channel.close() ch.basic_ack(delivery_tag=method.delivery_tag) # no_ack=Fales:表示消费完以后不主动把状态通知rabbitmq channel.basic_consume(callback, queue='Hi', no_ack=False) print('[*] Waiting for msg') channel.start_consuming()
durable 消息不丢失
消息生产者端发送消息时挂掉了,消费者接消息时挂掉了,以下方法会让RabbitMQ重新将该消息添加到队列中:
- 回调函数中的
ch.basic_ack(delivery_tag=method.delivery_tag)
,消费端需要做的 - basic_comsume中的
no_ack=False
,消费端需要做的 - 发布消息端的basic_publish添加参数
properties=pika.BasicProperties(delivery_mode=2)
,生产者端需要做的
import pika connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.0.74')) channel = connection.channel() channel.queue_declare(queue='Hi') # 如果有,略过;如果没有,创建队列 channel.basic_publish(exchange='', routing_key='Hi', body='hello!world!!!', properties=pika.BasicProperties(delivery_mode=2)) #消息持久化 print("[x] sent 'hello,world!'") connection.close()
import pika connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.0.74')) channel = connection.channel() channel.queue_declare(queue='Hi') # 定义回调函数 def callback(ch, method, properties, body): print('[x] Recieved %r' % body) # channel.close() ch.basic_ack(delivery_tag=method.delivery_tag) # no_ack=Fales:表示消费完以后不主动把状态通知rabbitmq channel.basic_consume(callback, queue='Hi', no_ack=True) print('[*] Waiting for msg') channel.start_consuming()
消息获取顺序
默认消息队列里的数据是按照顺序被消费者拿走,例如:消费者1去队列中获取 奇数 序列的任务,消费者2去队列中获取 偶数 序列的任务。但有大部分情况下,消息队列后端的消费者服务器的处理能力是不相同的,这就会出现有的服务器闲置时间较长,资源浪费的情况,那么,我们就需要改变默认的消息队列获取顺序!
channel.basic_qos(prefetch_count=1)
表示谁来谁取,不再按照奇偶数排列,这是消费者端需要做的
import pika connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.0.74')) channel = connection.channel() channel.queue_declare(queue='Hi') # 定义回调函数 def callback(ch, method, properties, body): print('[x] Recieved %r' % body) # channel.close() ch.basic_ack(delivery_tag=method.delivery_tag) channel.basic_qos(prefetch_count=1) #改变默认获取顺序,谁来谁取 # no_ack=Fales:表示消费完以后不主动把状态通知rabbitmq channel.basic_consume(callback, queue='Hi', no_ack=True) print('[*] Waiting for msg') channel.start_consuming()
发布和订阅
发布订阅和简单的消息队列区别在于,发布订阅会将消息发送给所有的订阅者,而消息队列中的数据被消费一次便消失。所以,RabbitMQ实现发布和订阅时,会为每一个订阅者创建一个队列,而发布者发布消息时,会将消息放置在所有相关队列中。
exchange type = fanout
任何发送到Fanout Exchange的消息都会被转发到与该Exchange绑定(Binding)的所有Queue上。
1.可以理解为路由表的模式
2.这种模式不需要RouteKey
3.这种模式需要提前将Exchange与Queue进行绑定,一个Exchange可以绑定多个Queue,一个Queue可以同多个Exchange进行绑定。
4.如果接受到消息的Exchange没有与任何Queue绑定,则消息会被抛弃。
import pika connection=pika.BlockingConnection(pika.ConnectionParameters(host='192.168.0.74')) channel = connection.channel() channel.exchange_declare(exchange='logs_fanout',type='fanout') msg='456' channel.basic_publish(exchange='logs_fanout',routing_key='',body=msg) print('开始发送:%s'%msg) connection.close()
import pika connection=pika.BlockingConnection(pika.ConnectionParameters(host='192.168.0.74')) channel = connection.channel() channel.exchange_declare(exchange='logs_fanout',type='fanout') #随机创建队列 result=channel.queue_declare(exclusive=True) queue_name=result.method.queue #绑定相关队列名称 channel.queue_bind(exchange='logs_fanout',queue=queue_name) def callback(ch,method,properties,body): print('[x] %r'%body) channel.basic_consume(callback,queue=queue_name,no_ack=True) channel.start_consuming()
关键字
import pika connection=pika.BlockingConnection(pika.ConnectionParameters(host='192.168.0.74')) channel = connection.channel() channel.exchange_declare(exchange='logs_direct_test1',type='direct') serverity='error' msg='123' channel.basic_publish(exchange='logs_direct_test1',routing_key=serverity,body=msg) print('开始发送:%r:%r'%(serverity,msg)) connection.close()
import pika connection=pika.BlockingConnection(pika.ConnectionParameters(host='192.168.0.74')) channel = connection.channel() channel.exchange_declare(exchange='logs_direct_test1',type='direct') #随机创建队列 result=channel.queue_declare(exclusive=True) queue_name=result.method.queue serverities=['error','info','warning',] for serverity in serverities: channel.queue_bind(exchange='logs_direct_test1',queue=queue_name,routing_key=serverity) print('[***] 开始接受消息!') 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()
import pika connection=pika.BlockingConnection(pika.ConnectionParameters(host='192.168.0.74')) channel = connection.channel() channel.exchange_declare(exchange='logs_direct_test1',type='direct') #随机创建队列 result=channel.queue_declare(exclusive=True) queue_name=result.method.queue serverities=['error',] for serverity in serverities: channel.queue_bind(exchange='logs_direct_test1',queue=queue_name,routing_key=serverity) print('[***] 开始接受消息!') 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()
模糊订阅
#!/usr/bin/env python import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters( host='192.168.0.74')) channel = connection.channel() channel.exchange_declare(exchange='topic_logs', 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()
#!/usr/bin/env python import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters( host='192.168.0.74')) channel = connection.channel() channel.exchange_declare(exchange='topic_logs', 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()