rabbitMq超详解

Posted 2020-10-14

在此向前辈们致敬：http://blog.csdn.net/shatty/article/details/9529463

为什么要学rabbitMQ

在此之前，我们想来进行一个概念区分

threading queue ：只能用于线程之间的消息传发

进程queue:可以用于进程（父进程与子进程或者同属于同一父进程之间的子进程交互）之间的消息传发

那么不同的语言之间，不同的机器之间怎么实现相互通信呢，这是一个问题吧

因此，我们的rabbitMq就起了很大的作用

接下来，我们对函数进行一一的相关介绍

connection = pika.BlockingConnection(pika.ConnectionParameters(

   host=‘localhost‘))#固定格式，创建一个类似于socket连接，，因为是在本地进行，所以可以直接用localhost

如果与其他电脑连接

pika.BlockingConnection(pika.ConnectionParameters(‘127.0.0.1‘,5672,‘simple‘,credentials))

这样

我们来看一看关于这个函数的介绍

def get_connection_parameters （self ，host ，port ，vhost ，username ，password ，
                                  heartbeat_interval ）：“”“返回一个pika连接的连接参数。
        
        ：参数str主机：连接到的RabbitMQ主机
        ：param int port：连接的端口
        ：param str vhost：虚拟主机
        ：参数str用户名：使用的用户名
        ：参数str密码：使用的密码
        ：param int heartbeat_interval：AMQP心跳间隔
        ：rtype：pika。ConnectionParameters
        “””

第三步：channel = connection.channel() #在连接上创建一个频道

channel = connection.channel() #进行一个管道的声明

channel.queue_declare(queue=‘hello‘) #声明一个队列，进行消息的传送#客户端与服务端都需要这样

#注意以上都是套路，固定格式

接下来就是消息的发送呢

channel.basic_publish(exchange=‘‘,
                     routing_key=‘hello‘,#消息队列的名字

                     body=‘Hello World!‘)#消息的内容

connection.close() #当生产者发送完消息后，可选择关闭连接

我们再来看看消费者的写法

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
   ‘localhost‘))

channel = connection.channel()

channel.queue_declare(queue=‘hello‘)#这个是防止不知道谁先运行而造成的错误

#上面的几行代码都是套路，服务端的时候已介绍，此处不做过解释

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

   print(" [x] Received %r" % body)

#下面代码是对消息的处理

channel.basic_consume(callback,#一旦接收到消息，就调用callback函数
                     queue=‘hello‘,

                     no_ack=True)

原理代码如下：生产者：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(‘localhost‘))
channel = connection.channel()

# 声明queue
channel.queue_declare(queue=‘hello‘)

# n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
channel.basic_publish(exchange=‘‘,
                     routing_key=‘hello‘,
                     body=‘Hello World!‘)
print(" [x] Sent ‘Hello World!‘")
connection.close()

消费者：

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
   ‘localhost‘))
channel = connection.channel()
channel.queue_declare(queue=‘hello‘)
def callback(ch, method, properties, body):
   print(‘--->‘,ch,method,properties,body)
   print(" [x] Received %r" % body)
channel.basic_consume(callback,
                     queue=‘hello‘,
                     no_ack=True)
print(‘ [*] Waiting for messages. To exit press CTRL+C‘)

channel.start_consuming()

最后结果为：

C:\Python\Python36\python.exe C:/Users/Administrator/PycharmProjects/untitled3/python/day9/消费者.py

 [*] Waiting for messages. To exit press CTRL+C

---> <BlockingChannel impl=<Channel number=1 OPEN conn=<SelectConnection OPEN socket=(‘::1‘, 58661, 0, 0)->(‘::1‘, 5672, 0, 0) params=<ConnectionParameters host=localhost port=5672 virtual_host=/ ssl=False>>>> #ch是我们刚声明的内存对象的地址<Basic.Deliver([‘consumer_tag=ctag1.3ee0d6275e9f43288f95fe2ba2c83e1a‘, ‘delivery_tag=1‘, ‘exchange=‘, ‘redelivered=False‘, ‘routing_key=hello‘])> #这个包含你要把消息发给哪个queue的信息<BasicProperties> b‘Hello World!‘

 [x] Received b‘Hello World!‘

好了，我们可以同时开三个消费者，不断地接收消息，

那么生产者没有收到接收消息的确认怎么办呢

消费者：

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
   ‘localhost‘))
channel = connection.channel()
channel.queue_declare(queue=‘hello‘)
def callback(ch, method, properties, body):
   print(‘--->‘,ch,method,properties,body)
   print(" [x] Received %r" % body)
channel.basic_consume(callback,
                     queue=‘hello‘,
                     #no_ack=True
                     )
print(‘ [*] Waiting for messages. To exit press CTRL+C‘)

channel.start_consuming()

生产者：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(‘localhost‘))
channel = connection.channel()

# 声明queue
channel.queue_declare(queue=‘hello‘)

# n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
channel.basic_publish(exchange=‘‘,
                     routing_key=‘hello‘,
                     body=‘Hello World!‘)
print(" [x] Sent ‘Hello World!‘")

connection.close()

结果是发现，生产者发送给一个消费者的消息传递给生产者了