标签：info 回调 rmi 字符 通过 htm star 相同 api
一、RabbitMQ安装
1、简介
RabbitMQ是一个消息代理：它接受和转发消息。可以将其视为邮局，当你要发送的邮件放入邮局中时，你可以确认邮件是否安全的达到接收者的手中，这里RabbitMQ就相当于邮局的角色。
2、RabbitMQ的安装
因为RabbitMQ是erlang语言开发的，所以需要先安装erlang。
从EPEL源安装：
[root@localhost ~]# yum install epel-release
[root@localhost ~]# yum install erlang
下载：
[root@localhost ~]# wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.6/rabbitmq-server-3.6.6-1.el7.noarch.rpm
安装：
[root@localhost ~]# yum install rabbitmq-server-3.6.6-1.el7.noarch.rpm
启动服务：
[root@localhost bin]# rabbitmq-server start
查看用户以及权限：
查看用户：rabbitmqctl list_users 查看用户权限：rabbitmqctl list_user_permissions guest 新增用户： rabbitmqctl add_user admin 123 赋予管理员权限： rabbitmqctl set_user_tags admin administrator rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"
启动web监控：
[root@localhost ~]# rabbitmq-plugins enable rabbitmq_management
访问默认端口15672
另外，如果操作RabbitMQ，需要安装API，进行操作，在开发环境中安装pika模块
pip install pika
二、六种工作模式
1、生产者消费者模式
“P”是生产者，“C”是消费者。中间的框是一个队列 ，用于存放消息。
生产者是将任务放入到队列中：
import pika #创建用户名密码 credentials = pika.PlainCredentials("admin","123") #创建连接 connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() # 创建一个队列 channel.queue_declare(queue=‘hello‘) channel.basic_publish(exchange=‘‘, routing_key=‘hello‘, # 消息队列名称 body=‘helloworld‘) connection.close()
消费者是取出任务队列并且进行处理：
import pika credentials = pika.PlainCredentials("admin","123") connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() # 创建一个队列,如果已经存在就不会重新创建 channel.queue_declare(queue=‘hello‘) def callback(ch, method, properties, body): print("消费者接受到了任务: %r" % body) channel.basic_consume(‘hello‘,callback,auto_ack=True) channel.start_consuming()
详情参考：https://www.rabbitmq.com/tutorials/tutorial-one-python.html
2、 竞争消费者模式
（1）消息确认auto_ack
如果消费者（上图中的C1和C2）处理从队列取出的任务，但是没有完成时就已经挂掉了，那么如果使用之前的代码auto_ack=True，一旦RabbitMQ向消费者传递任务，它立即将其标记为删除。在这种情况下，如果挂掉一个消费者，将丢失它刚刚处理的任务。
为了确保任务永不丢失，RabbitMQ支持消息确认。消费者发回ack（nowledgement）告诉RabbitMQ已收到，处理了特定消息，RabbitMQ可以自由删除它。
如果消费者挂掉（其通道关闭，连接关闭或TCP连接丢失）而不发送确认，RabbitMQ将未完全处理的任务并重新排队。如果同时有其他在线消费者，则会迅速将其重新发送给其他消费者。
主要变动在于消费者的变动：
import pika credentials = pika.PlainCredentials("admin","123") connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() # 创建一个队列,如果已经存在就不会重新创建 channel.queue_declare(queue=‘hello‘) def callback(ch, method, properties, body): print("消费者接受到了任务: %r" % body) ch.basic_ack(delivery_tag=method.delivery_tag)#防止消费者挂掉任务丢失 channel.basic_consume(‘hello‘,callback,auto_ack=False)#默认情况auto_ack=False手动消息确认打开 channel.start_consuming()
生产者未变动：
import pika #创建用户名密码 credentials = pika.PlainCredentials("admin","123") #创建连接 connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() # 创建一个队列 channel.queue_declare(queue=‘hello‘) channel.basic_publish(exchange=‘‘, routing_key=‘hello‘, # 消息队列名称 body=‘你好‘) connection.close()
（2）消息持久性
如果RabbitMQ服务器停止，队列中任务仍然会丢失。此时需要使用durable 参数，声明队列是持久的，但是此时以前的队列就不可以使用，持久的队列需要重新声明，也就是说需要改一下队列名称。
主要变动在与生产者的变动：
import pika #创建用户名密码 credentials = pika.PlainCredentials("admin","123") #创建连接 connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() # 创建一个持久化队列 channel.queue_declare(queue = ‘task_queue‘,durable = True) channel.basic_publish(exchange=‘‘, routing_key=‘task_queue‘, # 消息队列名称 body=‘你好‘, properties=pika.BasicProperties( delivery_mode=2, # 将消息标记为持久性 )) connection.close()
消费者未变动：
import pika credentials = pika.PlainCredentials("admin","123") connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() # 声明一个队列，已经创建就不会再创建了 channel.queue_declare(queue=‘task_queue‘) def callback(ch, method, properties, body): print("消费者接受到了任务: %r" % body) channel.basic_consume(‘task_queue’,callback,auto_ack=True) channel.start_consuming()
（3）公平派遣
可以看到上面的图，一个生产者，两个消费者，生产者将任务不断的放入到队列中，消费者不断的取出任务，那么这两个消费者是如何取任务的呢？
RabbitMQ默认情况下是均匀地发送消息，也就是消费者一个接一个的取出任务。这样如果一个消费者处理任务的时间比较长，还是均匀的给任务，势必造成一个消费者将经常忙，而另一个会很闲。
此时可以通过basic.qos方法和 prefetch_count = 1设置，将任务发送给空闲的消费者。
变动主要在消费者：
import pika credentials = pika.PlainCredentials("admin","123") connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() # 声明一个队列(创建一个队列) channel.queue_declare(queue=‘hello‘) def callback(ch, method, properties, body): print("消费者接受到了任务: %r" % body) channel.basic_qos(prefetch_count=1) channel.basic_consume(‘hello‘,callback,auto_ack=True) channel.start_consuming()
生产者未变动：
import pika # 无密码 # connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘)) # 有密码 credentials = pika.PlainCredentials("admin","123") connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() # 声明一个队列(创建一个队列) channel.queue_declare(queue=‘hello‘) channel.basic_publish(exchange=‘‘, routing_key=‘hello‘, # 消息队列名称 body=‘jjfdk‘) connection.close()
详情参考：https://www.rabbitmq.com/tutorials/tutorial-two-python.html
3、发布/订阅
在之前的RabbittMQ中主要用于将任务放入一个队列钟，然后消费者分别取出任务进行处理，而发布订阅是每一个消费者都将拥有自己的一个队列，从而获取消息。这样每一个消费者都将拥有相同的消息。
在上述模型中，“P”是生产者，也就是消息制造者，“X”是exchange ，用于将生产出来的消息给每一个队列给一份，中间红色的就是队列，“C1”和“C2”是消息接受者。
发布者：
import pika credentials = pika.PlainCredentials("admin","123") connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() channel.exchange_declare(exchange=‘ex1‘,exchange_type=‘fanout‘)#fanout工作方式为ex1每一个队列添加消息 channel.basic_publish(exchange=‘ex1‘, routing_key=‘‘, body=‘abcd‘) connection.close()
订阅者：
import pika credentials = pika.PlainCredentials("admin","123") connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() # exchange=‘ex1‘,exchange的名称 # exchange_type=‘fanout‘ , 工作方式将消息发送给所有的队列 channel.exchange_declare(exchange=‘ex1‘,exchange_type=‘fanout‘) # 随机生成一个队列 result = channel.queue_declare(queue = ‘‘,exclusive = True) queue_name = result.method.queue # 让exchange和queque进行绑定. channel.queue_bind(exchange=‘ex1‘,queue=queue_name)
详情参考：https://www.rabbitmq.com/tutorials/tutorial-three-python.html
4、关键字发布/订阅
上述发布的工作方式是：
exchange_type=‘fanout‘
将消息发送给所有队列，而选择性的发布/订阅，需要使用exchange_type=‘direct‘和routing_key=""进行关键字发布订阅。
发布者
import pika credentials = pika.PlainCredentials("admin","123") connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() channel.exchange_declare(exchange=‘ex2‘,exchange_type=‘direct‘) channel.basic_publish(exchange=‘ex2‘, routing_key=‘gh‘, body=‘nhgjod‘) connection.close()
exchange名称为“ex2”，工作方式为“direct”，然后将exchange与关键字routing_key关键字进行绑定。
订阅者1
import pika credentials = pika.PlainCredentials("admin","123") connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() channel.exchange_declare(exchange=‘ex2‘,exchange_type=‘direct‘) # 随机生成一个队列 result = channel.queue_declare(queue = ‘‘,exclusive=True) queue_name = result.method.queue # 让exchange和queque进行绑定. channel.queue_bind(exchange=‘ex2‘,queue=queue_name,routing_key=‘bright‘) channel.queue_bind(exchange=‘ex2‘,queue=queue_name,routing_key=‘gh‘)
exchange与queue以及关键字进行绑定，routing_key=“gh”可以收到消息。
订阅者2
import pika credentials = pika.PlainCredentials("admin","123") connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() channel.exchange_declare(exchange=‘ex2‘,exchange_type=‘direct‘) # 随机生成一个队列 result = channel.queue_declare(queue = ‘‘,exclusive=True) queue_name = result.method.queue # 让exchange和queque进行绑定. channel.queue_bind(exchange=‘ex2‘,queue=queue_name,routing_key=‘bright‘) def callback(ch, method, properties, body): print("消费者接受到了任务: %r" % body) channel.basic_consume(queue_name,callback,auto_ack=True) channel.start_consuming()
routing_key=“bright”不能接收到关键字为“gh”的发布者发布的消息。
详情参考：https://www.rabbitmq.com/tutorials/tutorial-four-python.html
5、关键字模糊匹配发布
可以看到使用type=topic，以及使用“#”以及“*”：
当队列绑定“#”绑定routing_key时，匹配任意字符。
当特殊字符“*”绑定routing_key时，匹配一个单词。
消息发布：
import pika credentials = pika.PlainCredentials("admin","123") connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() channel.exchange_declare(exchange=‘ex3‘,exchange_type=‘topic‘) channel.basic_publish(exchange=‘ex3‘, routing_key=‘bright.gh.km‘, body=‘abcdefd‘) connection.close()
可以看到关键字routing_key=‘bright.gh.km‘，其余的与之前发布/订阅并没什么区别。
消息订阅1：
import pika credentials = pika.PlainCredentials("admin","123") connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() channel.exchange_declare(exchange=‘ex3‘,exchange_type=‘topic‘) # 随机生成一个队列 result = channel.queue_declare(queue="",exclusive=True) queue_name = result.method.queue # 让exchange和queque进行绑定. channel.queue_bind(exchange=‘ex3‘,queue=queue_name,routing_key=‘bright.*‘) def callback(ch, method, properties, body): print("消费者接受到了任务: %r" % body) channel.basic_consume(queue_name,callback,auto_ack=True) channel.start_consuming()
这时可以看到订阅者1的routing_key=‘bright.*‘，可以匹配发布消息以bright开头的后面跟一个单词，所以发布者的消息它接收不到。
订阅者2：
import pika credentials = pika.PlainCredentials("admin","123") connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() channel.exchange_declare(exchange=‘ex3‘,exchange_type=‘topic‘) # 随机生成一个队列 result = channel.queue_declare(queue="",exclusive=True) queue_name = result.method.queue # 让exchange和queque进行绑定. channel.queue_bind(exchange=‘ex3‘,queue=queue_name,routing_key=‘bright.#‘) def callback(ch, method, properties, body): print("消费者接受到了任务: %r" % body) channel.basic_consume(queue_name,callback,auto_ack=True) channel.start_consuming()
这时可以看到订阅者1的routing_key=‘bright.#‘，可以匹配发布消息以bright开头的后面多个单词、字符，所以发布者的消息它可以接收到，这也就说明“#”比“*”更强大。
详情参考：https://www.rabbitmq.com/tutorials/tutorial-five-python.html
6、远程过程调用（RPC）
RPC工作方式：
客户端：
import pika import uuid class FibonacciRpcClient(object): def __init__(self): credentials = pika.PlainCredentials("admin", "123") self.connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘, credentials=credentials)) self.channel = self.connection.channel() # 随机生成一个消息队列(用于接收结果) result = self.channel.queue_declare(queue="",exclusive=True) self.callback_queue = result.method.queue # 监听消息队列(返回结果的队列)中是否有值返回,如果有值则执行 on_response 函数(一旦有结果,则执行on_response) self.channel.basic_consume(self.callback_queue,self.on_response, auto_ack=True) def on_response(self, ch, method, props, body): if self.corr_id == props.correlation_id: self.response = body def call(self, n): self.response = None self.corr_id = str(uuid.uuid4()) # 发送一个任务包含的内容: 任务id = corr_id ；任务内容 = ‘10‘ ；用于接收结果的队列名称 self.channel.basic_publish(exchange=‘‘, routing_key=‘rpc_queue‘, # 接收任务的队列名称 properties=pika.BasicProperties( reply_to = self.callback_queue, # 用于接收结果的队列 correlation_id = self.corr_id, # 任务ID ), body=str(n)) while self.response is None: self.connection.process_data_events() return self.response fibonacci_rpc = FibonacciRpcClient() response = fibonacci_rpc.call(10) print(‘结果:‘,response)
在客户端需要做以下的事情：
服务端：
import pika credentials = pika.PlainCredentials("admin","123") connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.0.104‘,credentials=credentials)) channel = connection.channel() # 监听任务队列,是否有任务到来 channel.queue_declare(queue=‘rpc_queue‘) def on_request(ch, method, props, body): n = int(body) response = n*100 # props.reply_to 要放结果的队列. # props.correlation_id 任务id ch.basic_publish(exchange=‘‘, routing_key=props.reply_to, properties=pika.BasicProperties(correlation_id= props.correlation_id), body=str(response)) ch.basic_ack(delivery_tag=method.delivery_tag) channel.basic_qos(prefetch_count=1) channel.basic_consume(queue = ‘rpc_queue‘,on_message_callback = on_request) channel.start_consuming()
在服务端需要做以下的事情：
详情参考：https://www.rabbitmq.com/tutorials/tutorial-six-python.html
标签：info 回调 rmi 字符 通过 htm star 相同 api
原文地址：https://www.cnblogs.com/shenjianping/p/11017625.html
