Python与Rabbitmq

1

docker pull rabbitmq:management # 直接使用带web管理的

1
2

mkdir /home/rabbitmq_data
docker run -itd --name rabbitmq -p 5672:5672 -p 15672:15672 -v /home/rabbitmq_data:/var/lib/rabbitmq -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin rabbitmq:management

1

pip install kombu

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

from kombu import Connection


def kombu_production(key, body, key_timeout=None):
    """
    rabbitmq 生产者
    :param key: 队列名
    :param body: 队列数据
    :param key_timeout: type:int 单位:s 超时时间 默认不过期
    :return:
    """
    MQ_URL = "amqp://admin:admin@192.168.1.100:5672//"
    connection = Connection(MQ_URL)
    channel = connection.Producer(serializer='json')
    channel.publish(body=body, routing_key=key, expiration=key_timeout)
    connection.release()
    

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

from kombu import Connection, Queue


class Consumer(object):
    """任务队列消费者"""

    def __init__(self, queue_name):
        """
        :param queue_name: 队列名称
        """
        self.queue_name = queue_name
        self.connection = Connection('amqp://admin:admin@192.168.1.100:5672//')
        
    def run(self):
        """开始消费队列"""
        while True:
            try:
                with self.connection.Consumer(
                        queues=[Queue(self.queue_name, routing_key=self.queue_name)],
                        accept=['pickle', 'json'],
                        callbacks=[self.task_callbacks],
                        prefetch_count=1  # 每次取得一个消息
                ):
                    self.connection.drain_events(timeout=10)
            except socket.timeout as e:
                logger.error(f'运行时异常 {self.queue_name} {e}')
                break
            except Exception as e:
                logger.error(f'运行时异常 {self.queue_name} {e}')
                break

    @staticmethod
    def task_callbacks(body, message):
        """任务回调函数(重写)"""
        message.requeue()
        

class MyConsumer(Consumer):
    """
    自定义的一个消费者
    """
    
    def task_callbacks(body, message):
        if body:
            print(body)
        return message.ack
    
    
if __name__ == "__main__":
    my_consumer = MyConsumer('MyQueue')
    my_consumer.run()