Python RabbitMQ 메시지 큐 rpc 구현
RabbitMQ 의 몇 가지 모델 과 메커니즘 은 ActiveMQ 보다 많 습 니 다.업무 수요 에 따라 RPC 를 사용 하여 기능 을 실현 합 니 다.그 중에서 밟 은 구덩이 들 은 기록 할 필요 가 있 습 니 다.
위의 코드,디 렉 터 리 구 조 는 c 로 나 뉜 다.server、c_client、c_hanlder:
c_server:
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pika
import time
import json
import io
import yaml
s_exchange = input(" ->>").decode('utf-8').strip()
s_queue = input(" ->>").decode('utf-8').strip()
credentials = pika.PlainCredentials('system', 'manager')
connection = pika.BlockingConnection(pika.ConnectionParameters(host='XXX.XXX.XXX.XXX',credentials=credentials))
#
channel = connection.channel()
channel.exchange_declare(exchange=s_exchange, exchange_type='direct')
channel.queue_declare(queue=s_queue, exclusive=True)
channel.queue_bind(queue=s_queue, exchange=s_exchange)
def s_manage(content):
# unicode json.JSONDecoder().decode(content)
str_content = yaml.safe_load(json.loads(content,encoding='utf-8'))
str_res = {
"errorid": 0,
"resp": str_content['cmd'],
"errorcont": " "
}
return json.dumps(str_res)
def on_request(ch, method, props, body):
response = s_manage(body)
ch.basic_publish(exchange='',
routing_key=props.reply_to,
properties=pika.BasicProperties(correlation_id = \
props.correlation_id),
body=response)
ch.basic_ack(delivery_tag = method.delivery_tag)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(on_request, queue=s_queue)
print(" [x] Awaiting RPC requests")
channel.start_consuming()
c_client:
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pika
import uuid
import json
import io
class RpcClient(object):
def __init__(self):
self.credentials = pika.PlainCredentials('guest', 'guest')
self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='XXX.XXX.XXX.XXX',
credentials=self.credentials))
self.channel = self.connection.channel()
def on_response(self, ch, method, props, body):
if self.callback_id == props.correlation_id:
self.response = body
ch.basic_ack(delivery_tag=method.delivery_tag)
def get_response(self, callback_queue, callback_id):
''' , callback_queued '''
self.callback_id = callback_id
self.response = None
self.channel.queue_declare('q_manager', durable=True)
self.channel.basic_consume(self.on_response, # on_response
queue=callback_queue)
while self.response is None:
self.connection.process_data_events() # start_consuming
return self.response
def call(self, queue_name, command, exchange,rout_key): #
''' '''
# result = self.channel.queue_declare(exclusive=False) #exclusive=False
self.callback_queue = 'q_manager' # result.method.queue
self.corr_id = str(uuid.uuid4())
self.channel.basic_publish(exchange=exchange,
routing_key=queue_name,
properties=pika.BasicProperties(
reply_to=self.callback_queue, # name
correlation_id=self.corr_id, # uuid
),
body=command)
return self.callback_queue,self.corr_id
client
c_handler:
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from c_client import *
import random, time
import threading
import json
import sys
class Handler(object):
def __init__(self):
self.information = {} #
def check_all(self, *args):
''' '''
time.sleep(2)
print(' ')
for key in self.information:
print("cid【%s】\t 【%s】\t 【%s】"%(key, self.information[key][0],
self.information[key][1]))
def check_task(self, cmd):
''' task_id '''
time.sleep(2)
try:
task_id = int(cmd)
print(task_id)
callback_queue= self.information[task_id][2]
callback_id= self.information[task_id][3]
client = RpcClient()
response = client.get_response(callback_queue, callback_id)
print(response)
# print(response.decode())
del self.information[task_id]
except KeyError as e :
print("error: [%s]" % e)
except IndexError as e:
print("error: [%s]" % e)
def run(self, user_cmd, host, exchange='', rout_key='',que=''):
try:
time.sleep(2)
command = user_cmd
task_id = random.randint(10000, 99999)
client = RpcClient()
response = client.call(queue_name=host, command=command,exchange=exchange,rout_key=que)
self.information[task_id] = [host, command, response[0], response[1]]
except IndexError as e:
print("[error]:%s"%e)
def reflect(self, str,cmd,host,exchange,que):
''' '''
if hasattr(self, str):
getattr(self, str)(cmd,host,exchange,que)
def start(self, m,cmd, host, exchange,que):
while True:
user_resp = input(" ID->>").decode('utf-8').strip()
self.check_task(user_resp)
str = m
print(self.information)
t1 = threading.Thread(target=self.reflect, args=(str,cmd,host,exchange,que)) #
t1.start()
s_exchange = input(" ->>").decode('utf-8').strip()
s_queue = input(" ->>").decode('utf-8').strip()
d_cmd_state =input(" json ->>").decode('utf-8').strip()
s_cmd = json.dumps(d_cmd_state)
handler = Handler()
handler.start('run',s_cmd, s_queue, s_exchange, s_queue)
handler
주의 포인트:1,cclient 가 rabbitmq 에 메 시 지 를 보 내 려 면 서버 가 되 돌아 오 는 대기 열 이름과 corrid2、c_handler 가 처 리 했 습 니 다.보 낼 때마다 task 목록 에 넣 고 ID 번호 가 표 시 될 때 까지 되 돌아 오 는 내용 을 조회 할 수 있 습 니 다.다음 과 같이 호출 할 수 있 습 니 다.
이상 이 바로 본 고의 모든 내용 입 니 다.여러분 의 학습 에 도움 이 되 고 저 희 를 많이 응원 해 주 셨 으 면 좋 겠 습 니 다.
이 내용에 흥미가 있습니까?
현재 기사가 여러분의 문제를 해결하지 못하는 경우 AI 엔진은 머신러닝 분석(스마트 모델이 방금 만들어져 부정확한 경우가 있을 수 있음)을 통해 가장 유사한 기사를 추천합니다:
로마 숫자를 정수로 또는 그 반대로 변환그 중 하나는 로마 숫자를 정수로 변환하는 함수를 만드는 것이었고 두 번째는 그 반대를 수행하는 함수를 만드는 것이었습니다. 문자만 포함합니다'I', 'V', 'X', 'L', 'C', 'D', 'M' ; 문자열이 ...
텍스트를 자유롭게 공유하거나 복사할 수 있습니다.하지만 이 문서의 URL은 참조 URL로 남겨 두십시오.
CC BY-SA 2.5, CC BY-SA 3.0 및 CC BY-SA 4.0에 따라 라이센스가 부여됩니다.