Linux学习之HIS学习-RabbitMQ所使用文件

# ！/usr/bin/env python3 import pika import json import threading import time class RabbitMQConfig(object): host = "192.168.88.50" port = 5672 username = 'admin' password = 'hisadmin' vhost = "/his" exchange = "demo" routing_key = "routing_demo" queue_name = "test" def callback(ch, method, properties, body): print('callback body ', body.decode()) ch.basic_ack(delivery_tag=method.delivery_tag) class RabbitMQ(object): def __init__(self, username, password, host, port, virtual_host): self.username = username self.password = password self.host = host self.port = port self.virtual_host = virtual_host self.get_connect() def get_connect(self): credentials = pika.PlainCredentials( username=self.username, password=self.password ) self.connect = pika.BlockingConnection( pika.ConnectionParameters( host=self.host, port=self.port, virtual_host=self.virtual_host, credentials=credentials ) ) self.channel = self.connect.channel() def set(self, exchange, routing_key, queue, body): # 创建交换机 self.channel.exchange_declare( exchange=exchange, exchange_type='direct', durable=False ) # 声明队列并持久化 self.channel.queue_declare( queue=queue, durable=True ) # 通过routing_key绑定消息交换机和队列 self.channel.queue_bind( queue=queue, exchange=exchange, routing_key=routing_key ) # 发消息 self.channel.basic_publish( exchange=exchange, routing_key=routing_key, body=body, properties=pika.BasicProperties( delivery_mode=2 ) ) def get(self, exchange, queue, callback): print('消费任务启动') self.channel.exchange_declare( exchange=exchange, exchange_type='direct', durable=False ) self.channel.queue_declare( queue=queue, durable=True ) self.channel.basic_consume( queue=queue, on_message_callback=callback, auto_ack=False ) self.channel.start_consuming() if __name__ == '__main__': rabbit_obj = RabbitMQ( username=RabbitMQConfig.username, password=RabbitMQConfig.password, host=RabbitMQConfig.host, port=RabbitMQConfig.port, virtual_host=RabbitMQConfig.vhost ) for i in range(10): msg = { "Content": "message " + time.ctime(time.time()) } rabbit_obj.set( RabbitMQConfig.exchange, RabbitMQConfig.routing_key, RabbitMQConfig.queue_name, json.dumps(msg) ) time.sleep(1) print('生产者发送第{}个消息成功 '.format(i)) p = threading.Thread( target=rabbit_obj.get, args=( RabbitMQConfig.exchange, RabbitMQConfig.queue_name, callback, ) ) p.start() print('\n --主程序不被阻塞，执行其他操作 --\n')