apt-get update
apt-get install erlang
apt-get install rabbitmq-server


#啟動(dòng)rabbitmq： service rabbitmq-server start
#停止rabbitmq： service rabbitmq-server stop
#重啟rabbitmq： service rabbitmq-server restart

#啟動(dòng)rabbitmq插件：rabbitmq-plugins enable rabbitmq_management

#添加新用戶(hù)
rabbitmqctl add_user 用戶(hù)名 密碼

#給指定用戶(hù)添加管理員權(quán)限
rabbitmqctl set_user_tags 用戶(hù)名 administrator

給用戶(hù)添加權(quán)限
rabbitmqctl set_permissions -p / 用戶(hù)名 ".*" ".*" ".*"

pip install pika
#皮卡皮卡，哈哈

# coding=utf-8
### 生產(chǎn)者

import pika
import time

user_info = pika.PlainCredentials('root', 'root')#用戶(hù)名和密碼
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))#連接服務(wù)器上的RabbitMQ服務(wù)

# 創(chuàng)建一個(gè)channel
channel = connection.channel()

# 如果指定的queue不存在，則會(huì)創(chuàng)建一個(gè)queue，如果已經(jīng)存在 則不會(huì)做其他動(dòng)作，官方推薦，每次使用時(shí)都可以加上這句
channel.queue_declare(queue='hello')

for i in range(0, 100):
    channel.basic_publish(exchange='',#當(dāng)前是一個(gè)簡(jiǎn)單模式，所以這里設(shè)置為空字符串就可以了
                          routing_key='hello',# 指定消息要發(fā)送到哪個(gè)queue
                          body='{}'.format(i)# 指定要發(fā)送的消息
                          )
    time.sleep(1)
    
# 關(guān)閉連接
# connection.close()

# coding=utf-8
### 消費(fèi)者

import pika

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
channel = connection.channel()

# 如果指定的queue不存在，則會(huì)創(chuàng)建一個(gè)queue，如果已經(jīng)存在 則不會(huì)做其他動(dòng)作，生產(chǎn)者和消費(fèi)者都做這一步的好處是
# 這樣生產(chǎn)者和消費(fèi)者就沒(méi)有必要的先后啟動(dòng)順序了
channel.queue_declare(queue='hello')


# 回調(diào)函數(shù)
def callback(ch, method, properties, body):
    print('消費(fèi)者收到:{}'.format(body))

# channel: 包含channel的一切屬性和方法
# method: 包含 consumer_tag, delivery_tag, exchange, redelivered, routing_key
# properties: basic_publish 通過(guò) properties 傳入的參數(shù)
# body: basic_publish發(fā)送的消息


channel.basic_consume(queue='hello',  # 接收指定queue的消息
                      auto_ack=True,  # 指定為T(mén)rue，表示消息接收到后自動(dòng)給消息發(fā)送方回復(fù)確認(rèn)，已收到消息
                      on_message_callback=callback  # 設(shè)置收到消息的回調(diào)函數(shù)
                      )

print('Waiting for messages. To exit press CTRL+C')

# 一直處于等待接收消息的狀態(tài)，如果沒(méi)收到消息就一直處于阻塞狀態(tài)，收到消息就調(diào)用上面的回調(diào)函數(shù)
channel.start_consuming()

# coding=utf-8
### 生產(chǎn)者
import pika
import time

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))

# 創(chuàng)建一個(gè)channel
channel = connection.channel()

# 如果指定的queue不存在，則會(huì)創(chuàng)建一個(gè)queue，如果已經(jīng)存在 則不會(huì)做其他動(dòng)作，官方推薦，每次使用時(shí)都可以加上這句
channel.queue_declare(queue='durable_queue',durable=True)
#PS：這里不同種隊(duì)列不允許名字相同

for i in range(0, 100):
    channel.basic_publish(exchange='',
                          routing_key='durable_queue',
                          body='{}'.format(i),
                          properties=pika.BasicProperties(delivery_mode=2)
                          )
# 關(guān)閉連接
# connection.close()

# coding=utf-8
### 消費(fèi)者

import pika
import time

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
channel = connection.channel()

# 如果指定的queue不存在，則會(huì)創(chuàng)建一個(gè)queue，如果已經(jīng)存在 則不會(huì)做其他動(dòng)作，生產(chǎn)者和消費(fèi)者都做這一步的好處是
# 這樣生產(chǎn)者和消費(fèi)者就沒(méi)有必要的先后啟動(dòng)順序了
channel.queue_declare(queue='queue')


# 回調(diào)函數(shù)
def callback(ch, method, properties, body):
    time.sleep(5)
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print('消費(fèi)者收到:{}'.format(body.decode('utf-8')))


# channel: 包含channel的一切屬性和方法
# method: 包含 consumer_tag, delivery_tag, exchange, redelivered, routing_key
# properties: basic_publish 通過(guò) properties 傳入的參數(shù)
# body: basic_publish發(fā)送的消息


channel.basic_consume(queue='queue',  # 接收指定queue的消息
                      auto_ack=False,  # 指定為False，表示取消自動(dòng)應(yīng)答，交由回調(diào)函數(shù)手動(dòng)應(yīng)答
                      on_message_callback=callback  # 設(shè)置收到消息的回調(diào)函數(shù)
                      )

# 應(yīng)答的本質(zhì)是告訴消息隊(duì)列可以將這條消息銷(xiāo)毀了

print('Waiting for messages. To exit press CTRL+C')

# 一直處于等待接收消息的狀態(tài)，如果沒(méi)收到消息就一直處于阻塞狀態(tài)，收到消息就調(diào)用上面的回調(diào)函數(shù)
channel.start_consuming()

# coding=utf-8
### 消費(fèi)者

import pika
import time

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
channel = connection.channel()

# 如果指定的queue不存在，則會(huì)創(chuàng)建一個(gè)queue，如果已經(jīng)存在 則不會(huì)做其他動(dòng)作，生產(chǎn)者和消費(fèi)者都做這一步的好處是
# 這樣生產(chǎn)者和消費(fèi)者就沒(méi)有必要的先后啟動(dòng)順序了
channel.queue_declare(queue='queue')


# 回調(diào)函數(shù)
def callback(ch, method, properties, body):
    time.sleep(0)#通過(guò)設(shè)置休眠時(shí)間來(lái)模擬不同消費(fèi)者的處理時(shí)間
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print('消費(fèi)者收到:{}'.format(body.decode('utf-8')))


# prefetch_count表示接收的消息數(shù)量，當(dāng)我接收的消息沒(méi)有處理完（用basic_ack標(biāo)記消息已處理完畢）之前不會(huì)再接收新的消息了
channel.basic_qos(prefetch_count=1)  # 還有就是這個(gè)設(shè)置必須在basic_consume之上，否則不生效

channel.basic_consume(queue='queue',  # 接收指定queue的消息
                      auto_ack=False,  # 指定為False，表示取消自動(dòng)應(yīng)答，交由回調(diào)函數(shù)手動(dòng)應(yīng)答
                      on_message_callback=callback  # 設(shè)置收到消息的回調(diào)函數(shù)
                      )

# 應(yīng)答的本質(zhì)是告訴消息隊(duì)列可以將這條消息銷(xiāo)毀了

print('Waiting for messages. To exit press CTRL+C')

# 一直處于等待接收消息的狀態(tài)，如果沒(méi)收到消息就一直處于阻塞狀態(tài)，收到消息就調(diào)用上面的回調(diào)函數(shù)
channel.start_consuming()

#生產(chǎn)者
import pika

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
channel = connection.channel()

# 創(chuàng)建一個(gè)指定名稱(chēng)的交換機(jī)，并指定類(lèi)型為fanout，用于將接收到的消息廣播到所有queue中
channel.exchange_declare(exchange='交換機(jī)', exchange_type='fanout')

# 將消息發(fā)送給指定的交換機(jī)，在fanout類(lèi)型中，routing_key=''表示不用發(fā)送到指定queue中，
# 而是將發(fā)送到綁定到此交換機(jī)的所有queue
channel.basic_publish(exchange='交換機(jī)', routing_key='', body='這是一條測(cè)試消息')

#消費(fèi)者
import pika

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
channel = connection.channel()

channel.exchange_declare(exchange='交換機(jī)', exchange_type='fanout')

# 使用RabbitMQ給自己生成一個(gè)專(zhuān)有的queue
result = channel.queue_declare(queue='333')
# result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
# 這里如果設(shè)置exclusive=True參數(shù)，那么該隊(duì)列就是一個(gè)只有隊(duì)列，在消費(fèi)者結(jié)束后，該專(zhuān)有隊(duì)列也會(huì)自動(dòng)清除，如果queue=''沒(méi)有設(shè)置名字的話(huà)，那么就會(huì)自動(dòng)生成一個(gè)
# 不會(huì)重復(fù)的隊(duì)列名

# 將queue綁定到指定交換機(jī)
channel.queue_bind(exchange='交換機(jī)', queue=queue_name)

print(' [*] Waiting for  message.')


def callback(ch, method, properties, body):
    print("消費(fèi)者收到:{}".format(body.decode('utf-8')))

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

channel.start_consuming()

import pika
import uuid
import json

class RPC(object):

    def __init__(self):
        self.call_id = None
        self.response = None
        user_info = pika.PlainCredentials('root', 'root')
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))
        self.channel = self.connection.channel()

        # 創(chuàng)建一個(gè)此客戶(hù)端專(zhuān)用的queue，用于接收服務(wù)端發(fā)過(guò)來(lái)的消息
        result = self.channel.queue_declare(queue='', exclusive=True)
        self.callback_queue = result.method.queue

        self.channel.basic_consume(
            queue=self.callback_queue,
            on_message_callback=self.on_response,
            auto_ack=True)

    def on_response(self, ch, method, props, body):
        # 判斷接收到的response是否屬于對(duì)應(yīng)request
        if self.call_id == props.correlation_id:
            self.response = json.loads(body.decode('utf-8')).get('result')

    def call(self, func, param):
        self.response = None
        self.call_id = str(uuid.uuid4())  # 為該消息指定uuid，類(lèi)似于請(qǐng)求id
        self.channel.queue_declare(queue='rpc_queue')
        self.channel.basic_publish(
            exchange='',
            routing_key='rpc_queue',  # 將消息發(fā)送到該queue
            properties=pika.BasicProperties(
                reply_to=self.callback_queue,  # 從該queue中取消息
                correlation_id=self.call_id,  # 為此次消息指定uuid
            ),
            body=json.dumps(
                {
                    'func': func,
                    'param': {'a': param[0], 'b': param[1]}
                }
            )
        )
        
        self.connection.process_data_events(time_limit=3)# 與start_consuming()相似，可以設(shè)置超時(shí)參數(shù)
        return self.response

rpc = RPC()

print("發(fā)送消息到消費(fèi)者,等待返回結(jié)果")

response = rpc.call(func='del', param=(1, 2))

print("收到來(lái)自消費(fèi)者返回的結(jié)果:{}".format(response))

import pika
import json

user_info = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(pika.ConnectionParameters('ip', 5672, '/', user_info))

channel = connection.channel()

# 指定接收消息的queue
channel.queue_declare(queue='rpc_queue')

def add_number(a, b):
    return a + b

def del_num(a, b):
    return a - b

execute_map = {
    'add': add_number,
    'del': del_num
}

def on_request(ch, method, props, body):
    body = json.loads(body.decode('utf-8'))
    func = body.get('func')
    param = body.get('param')
    result = execute_map.get(func)(param.get('a'), param.get('b'))
    print('進(jìn)行{}運(yùn)算，并將結(jié)果返回個(gè)消費(fèi)者'.format(func))

    ch.basic_publish(exchange='',  # 使用默認(rèn)交換機(jī)
                     routing_key=props.reply_to,  # response發(fā)送到該queue
                     properties=pika.BasicProperties(
                         correlation_id=props.correlation_id),  # 使用correlation_id讓此response與請(qǐng)求消息對(duì)應(yīng)起來(lái)
                     body=json.dumps({'result': result}))

    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)
# 從rpc_queue中取消息，然后使用on_request進(jìn)行處理
channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)

print(" [x] Awaiting RPC requests")
channel.start_consuming()