python服务器消息分发_pythonRabbitMQ（1）简单队列使用,消息依次分发（一对一），消息持久化处理...

MQ全称为Message Queue, 消息队列(MQ)是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息(针对应用程序的数据)来通信&＃xff0c;而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信&＃xff0c;而不是通过直接调用彼此来通信&＃xff0c;直接调用通常是用于诸如远程过程调用的技术。

MQ是消费-生产者模型的一个典型的代表&＃xff0c;一端往消息队列中不断写入消息&＃xff0c;而另一端则可以读取或者订阅队列中的消息。

RabbitMQ是一个在AMQP(高级消息队列协议)基础上完成的&＃xff0c;可复用的企业消息系统。

推荐文章&＃xff1a;

4.windows下的安装(其中注意版本一致&＃xff0c;Erlang版本不要太低)

补充&＃xff1a;远程连接rabbitmq server需要配置权限

1.在服务端创建创建rabbitmq账号

sudo rabbitmqctl add_user 用户名 密码

2.为账号配置权限

sudo rabbitmqctl set_permissions -p / 账号 ".*" ".*" ".*"

set_permissions [-p vhost] {user} {conf} {write} {read}

该命令使用户具有/这个virtual host中所有资源的配置、写、读权限以便管理其中的资源

3.客户端连接时认证方式

#1.设置远程账号密码

credentials&＃61; pika.PlainCredentials(&＃39;账号&＃39;, &＃39;密码&＃39;)

#2.设置参数&＃xff0c;创建连接,这是链接一个socket

connection&＃61;pika.BlockingConnection(pika.ConnectionParameters(

host&＃61;&＃39;远程主机ip&＃39;,端口号,&＃39;/&＃39;,credentials 　　#/代表的是服务端rabbitmq的虚拟主机资源目录

))

#3.根据socket连接&＃xff0c;在其基础上创建Channel通道&＃xff0c;

channel&＃61; connection.channel()

python使用&＃xff1a;

简单队列模型&＃xff1a;(exchange交换机不工作模式)

生产者&＃xff1a;

import pika

#1.设置参数&＃xff0c;创建连接

connection&＃61;pika.BlockingConnection(pika.ConnectionParameters(

host&＃61;&＃39;localhost&＃39;))

#2.根据连接创建Channel通道

channel&＃61;connection.channel()

#3.创建queue(名为hello)和channel进行绑定

channel.queue_declare(queue&＃61;&＃39;hello&＃39;)

#4.生成消息&＃xff0c;发布到队列中

channel.basic_publish(exchange&＃61;&＃39;&＃39;,

routing_key&＃61;&＃39;hello&＃39;,#去相关队列中

body&＃61;&＃39;Hello world&＃39;)

print(&＃39;Send Hello World th queue&＃39;)

#5.关闭连接 connection.close()

持久化处理&＃xff1a;服务端注意&＃xff0c;当服务端断电或者其他情况&＃xff0c;导致消息原本存在内存中的消失&＃xff0c;更可靠的是将数据放到磁盘中保存&＃xff0c;更安全&＃xff0c;但是降低效率　。

决定权在于客户端是否设置应答&＃xff0c;若是无应答&＃xff0c;消息接收后立刻消失&＃xff0c;不需要持久化&＃xff0c;有应答时&＃xff0c;需要将消息持久化处理&＃xff0c;因为可能客户端在调用回调函数需要耗时太长&＃xff0c;中途挂了&＃xff0c;那么这个消息在服务器端需要再次恢复到消息队列中。

#3.创建queue(名为hello)和channel进行绑定

channel.queue_declare(queue&＃61;&＃39;hello&＃39;,durable&＃61;True)　　#durable持久化&＃xff0c;将这个队列持久化&＃xff0c;重启服务器&＃xff0c;这个队列依旧存在

#4.生成消息&＃xff0c;发布到队列中

channel.basic_publish(exchange&＃61;&＃39;&＃39;,

routing_key&＃61;&＃39;hello&＃39;,

body&＃61;&＃39;Hello world&＃39;,

properties&＃61;pika.BasicProperties(　　#将消息保存在本地磁盘持久化处理&＃xff0c;在上面durable基础上&＃xff0c;将持久化队列中的消息一块持久化了

delivery_mode&＃61;2,#默认是1

))

修改上面两处

消费者&＃xff1a;

import pika

#1.设置参数&＃xff0c;获取一个connection连接

connection&＃61;pika.BlockingConnection(pika.ConnectionParameters(

host&＃61;&＃39;localhost&＃39;))

#2.利用connection创建一个channel通道

channel&＃61;connection.channel()

#3.将channel通道和queue(之前生产者创建的队列)进行绑定

channel.queue_declare(queue&＃61;&＃39;hello&＃39;)#生产者和消费者只要有一个先声明&＃xff0c;就会去创建这个队列&＃xff0c;不会重复创建

#回调函数&＃xff0c;处理消息内容

def callback(ch,method,properties,body):

print("Recvived %s"%body)

# consumer_callback(channel, method, properties, body)

# channel: BlockingChannel&＃xff0c;通道实例

# method: spec.Basic.Deliver&＃xff0c;头部消息&＃xff0c;附带一些数据

# properties: spec.BasicProperties&＃xff0c;

# body: str or unicode&＃xff0c;数据内容&＃xff0c;byte格式

#4.创建消费者进行消息处理

channel.basic_consume(callback,

queue&＃61;&＃39;hello&＃39;,

no_ack&＃61;True)#无应答&＃xff0c;默认false

print("Waiting for message")

#5.开启消费者

channel.start_consuming()#这里开始堵塞&＃xff0c;等待获取信息&＃xff0c;开始消费&＃xff0c;这个是IO循环一致等待&＃xff0c;不会退出

客户端注意&＃xff1a;no_ack&＃61;True/False   #无应答/有应答

#回调函数&＃xff0c;处理消息内容

def callback(ch,method,properties,body):

import time

time.sleep(10)　　#在此处可以进行中断测试&＃xff0c;查看有无应答

print("Recvived %s"%body)

若是回调函数处理数据时&＃xff0c;需要耗费大量时间&＃xff0c;而客户端并未处理而退出&＃xff0c;那么这条消息是否应该继续存在在队列中&＃xff0c;当设置无应答no_ack&＃61;True时&＃xff0c;消息从队列中取出后即消失。消息的安全性不够&＃xff0c;但是效率高

当设置为有应答no_ack&＃61;False时&＃xff0c;消息经过客户端成功处理后会返回响应码进行应答操作&＃xff0c;确保消息被处理后&＃xff0c;才会从消息队列中删除。安全性够好&＃xff0c;但是等待过程效率低

在设置为有应答时&＃xff0c;需要在回调函数末尾中添加&＃xff1a;

ch.basic_ack(delivery_tag&＃61;method.delivery_tag)　　#唯一标识&＃xff0c;自动生成&＃xff0c;用于在回调函数中&＃xff0c;进行应答

进行回应

另外&＃xff1a;对于rabbitmq的消息分发&＃xff0c;默认是按照顺序&＃xff0c;有几个消费者&＃xff0c;大家一个一个来&＃xff0c;但是由于每个消费者的消费速度不一致&＃xff0c;会导致时间上&＃xff0c;某些消费者会产生空闲

所以需要修改这种状态&＃xff0c;在消费者这地方修改&＃xff0c;修改消费者取数据模式&＃xff0c;添加&＃xff1a;

channel.basic_qos(prefetch_count&＃61;1)

这样&＃xff0c;谁空闲&＃xff0c;谁就来获取数据