1
#!/bin/sh
2

3
sudo apt-get install curl gnupg apt-transport-https -y
4

5
## Team RabbitMQ's signing key
6
curl -1sLf "https://keys.openpgp.org/vks/v1/by-fingerprint/0A9AF2115F4687BD29803A206B73A36E6026DFCA" | sudo gpg --dearmor | sudo tee /usr/share/keyrings/com.rabbitmq.team.gpg > /dev/null
7

8
## Add apt repositories maintained by Team RabbitMQ
9
sudo tee /etc/apt/sources.list.d/rabbitmq.list <<EOF
10
## Modern Erlang/OTP releases
11
##
12
deb [arch=amd64 signed-by=/usr/share/keyrings/com.rabbitmq.team.gpg] https://deb1.rabbitmq.com/rabbitmq-erlang/ubuntu/noble noble main
13
deb [arch=amd64 signed-by=/usr/share/keyrings/com.rabbitmq.team.gpg] https://deb2.rabbitmq.com/rabbitmq-erlang/ubuntu/noble noble main
14

15
## Latest RabbitMQ releases
16
##
17
deb [arch=amd64 signed-by=/usr/share/keyrings/com.rabbitmq.team.gpg] https://deb1.rabbitmq.com/rabbitmq-server/ubuntu/noble noble main
18
deb [arch=amd64 signed-by=/usr/share/keyrings/com.rabbitmq.team.gpg] https://deb2.rabbitmq.com/rabbitmq-server/ubuntu/noble noble main
19
EOF
20

21
## Update package indices
22
sudo apt-get update -y
23

24
## Install Erlang packages
25
sudo apt-get install -y erlang-base \
26
                        erlang-asn1 erlang-crypto erlang-eldap erlang-ftp erlang-inets \
27
                        erlang-mnesia erlang-os-mon erlang-parsetools erlang-public-key \
28
                        erlang-runtime-tools erlang-snmp erlang-ssl \
29
                        erlang-syntax-tools erlang-tftp erlang-tools erlang-xmerl
30

31
## Install rabbitmq-server and its dependencies
32
sudo apt-get install rabbitmq-server -y --fix-missing

1
# 设置连接参数
2
connection_params = pika.ConnectionParameters(host="localhost")
3

4
# 连接服务器
5
connection = pika.BlockingConnection(connection_params)
6
channel = connection.channel()

1
# 将队列名称设置为 'hello'
2
queue_name = 'hello'
3

4
# 声明队列
5
channel.queue_declare(queue=queue_name)

1
# 新建队列，并使用 .method.queue 获取队列名
2
queue_created = channel.queue_declare("",exclusive=True)
3
queue_name = queue_created.method.queue

1
# 将待发送的消息体设置为 'Hello World!'
2
message = 'Hello World!'
3

4
# 发送消息
5
channel.basic_publish(exchange='',
6
                      routing_key=queue_name,
7
                      body=message)

1
connection.close()

1
# 定义callback函数，用于在收到消息后处理消息
2
# ch为channel对象，method为消息传递时用到的方法，properties为消息的属性，body为消息体
3
def callback(ch, method, properties, body):
4
    print(" [x] Received %r" % body)
5

6
# 配置消费者：1.监听'hello'队列 2.收到消息后执行callback函数
7
channel.basic_consume(queue='hello',
8
                      auto_ack=True,    # 自动确认消息
9
                      on_message_callback=callback)

1
channel.start_consuming()

1
# 定义callback函数，用于在收到消息后处理消息
2
def callback(ch, method, properties, body):
3
    print(" [x] Received %r" % body)        # 打印消息体
4
    ch.basic_ack(delivery_tag=method.delivery_tag)  # 手动确认消息已被处理
5

6
# 配置消费者：1.监听'hello2'队列 2.收到消息后执行callback函数
7
channel.basic_consume(queue='hello',
8
                      auto_ack=False,    # 手动确认消息
9
                      on_message_callback=callback)

1
# 在参数列表中添加 durable=True，以声明一个可持久化的'durable_queue'队列
2
channel.queue_declare(queue='durable_queue', durable=True)

1
# 在参数列表中添加 properties=pika.BasicProperties(delivery_mode=2)
2
# 发送持久化消息到'hello'队列
3
channel.basic_publish(exchange='',
4
                      routing_key='durable_queue',
5
                      body='Hello World!',
6
                      properties=pika.BasicProperties(delivery_mode=2))  # 标记消息为持久化

1
# 在参数列表中添加 'x-max-priority'，以声明一个'priority_queue'优先级队列
2
channel.queue_declare(
3
    queue='priority_queue',
4
    arguments={
5
        'x-max-priority': 10  #设置最大优先级
6
    }
7
)

1
ch.basic_publish(exchange='',
2
              routing_key='priority_queue',
3
              body='Hello World!',
4
              properties=pika.BasicProperties(priority=5))  # 标记消息的优先级为 5

1
import pika
2

3
# 连接至RabbitMQ服务器
4
connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))
5
channel = connection.channel()
6

7
# 声明一个'logs'交换机，设置类型为'fanout'
8
channel.exchange_declare(exchange = 'logs',
9
                         exchange_type = 'fanout')
10

11
# 发送消息到'logs'交换机
12
channel.basic_publish(exchange = 'logs',
13
                      routing_key = '',
14
                      body = 'Hello World!')
15

16
# 关闭连接
17
connection.close()

1
import pika
2

3
# 连接至RabbitMQ服务器
4
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
5
channel = connection.channel()
6

7
# 声明一个'logs'交换机，设置类型为'fanout'
8
channel.exchange_declare(exchange='logs',exchange_type='fanout')
9

10
# 创建一个队列
11
result = channel.queue_declare("",exclusive=True)
12
queue_name = result.method.queue
13

14
# 将队列绑定到交换机
15
channel.queue_bind(exchange='logs',queue=queue_name)
16

17
# 定义callback函数，用于在收到消息后处理消息
18
def callback(ch, method, properties, body):
19
    print(" [x] %r" % body)
20

21
# 配置消费者
22
channel.basic_consume(queue=queue_name,
23
                      auto_ack=True,
24
                      on_message_callback=callback)
25

26
# 正式开始监听队列
27
channel.start_consuming()

1
import pika
2

3
# 连接至RabbitMQ服务器
4
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
5
channel = connection.channel()
6

7
# 声明一个'logs'交换机，设置类型为'direct'
8
channel.exchange_declare(exchange='logs',exchange_type='direct')
9

10
# 发送消息到'logs'交换机
11
channel.basic_publish(exchange='logs',
12
                      routing_key='info',
13
                      body='Hello World!')
14

15
connection.close()  # 关闭连接

1
import pika
2

3
# 连接至RabbitMQ服务器
4
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
5
channel = connection.channel()
6

7
# 声明一个‘logs2’交换机，设置类型为‘direct’
8
channel.exchange_declare(exchange='logs',exchange_type='direct')
9

10
# 创建一个临时队列
11
result = channel.queue_declare("",exclusive=True)
12
queue_name = result.method.queue
13

14
# 将队列绑定到交换机，同时提供routing_key参数
15
channel.queue_bind(exchange='logs',queue=queue_name, routing_key='info')
16

17
# 定义callback函数，用于在收到消息后处理消息
18
def callback(ch, method, properties, body):
19
    print(" [x] %r" % body)
20

21
# 配置消费者
22
channel.basic_consume(queue=queue_name,
23
                      auto_ack=True,
24
                      on_message_callback=callback)
25

26
channel.start_consuming()  # 正式开始监听队列

1
import pika
2

3
# 连接至RabbitMQ服务器
4
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
5
channel = connection.channel()
6

7
# 声明一个'logs'交换机，设置类型为'topic'
8
channel.exchange_declare(exchange='logs',exchange_type='topic')
9

10
# 发送消息到'logs'交换机
11
for sender in ['system', 'user']:
12
    for level in ['error', 'info', 'warning']:
13
        key = f"{sender}.{level}"
14
        channel.basic_publish(exchange='logs',
15
                              routing_key=key,
16
                              body=f"[{sender}]: {level}")
17

18
connection.close()  # 关闭连接

1
import pika
2

3
# 连接至RabbitMQ服务器
4
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
5
channel = connection.channel()
6

7
# 声明一个'logs'交换机，设置类型为'topic'
8
channel.exchange_declare(exchange='logs3',exchange_type='topic')
9

10
# 创建一个临时队列
11
result = channel.queue_declare("",exclusive=True)
12
queue_name = result.method.queue
13

14
# 将队列绑定到交换机
15
channel.queue_bind(exchange='logs',
16
                   queue=queue_name,
17
                   routing_key="system.*") # 绑定所有以system.为开头且仅含两个关键字的路由键
18

19
# 定义callback函数，用于在收到消息后处理消息
20
def callback(ch, method, properties, body):
21
    print(" [x] %r" % body)
22

23
# 配置消费者
24
channel.basic_consume(queue=queue_name,
25
                      auto_ack=True,
26
                      on_message_callback=callback)
27

28
channel.start_consuming()  # 正式开始监听队列