9 消息队列和Celery

阅读量：2771 次

发布时间：2019-05-13

本文共 3844 字，大约阅读时间需要 12 分钟。

1、Rabbit MQLinux 安装Centos7

Yum安装socat

# yum -y install socat

下载erlang软件包，本文使用erlang-19.0.4版本，下面给出下载链接

# wget http://www.rabbitmq.com/releases/erlang/erlang-19.0.4-1.el7.centos.x86_64.rpm

安装erlang

# rpm -ivh erlang-19.0.4-1.el7.centos.x86_64.rpm

安装完成后执行erl命令，出现下图则代表成功

最后安装rabbitmq,下载地址如下

#  wget  http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.10/rabbitmq-server-3.6.10-1.el7.noarch.rpm

安装.....

# rpm -ivh rabbitmq-server-3.6.10-1.el7.noarch.rpm

注意：如果是重装请记得删除/var/lib/rabbitmq目录和/etc/rabbitmq目录，否则可能服务会起不来

2、Rabbit MQ使用Linux

2.1 启动rabbitmq
启动命令

# systemctl start rabbitmq-server

3、安装客户端

pip  install pika

4、先启动消费者

amqp_consumer.py

# coding=utf-8import pika# 处理接收到的消息的回调函数# method_frame携带了投递标记， header_frame表示AMQP信息头的对象# body为消息实体def on_message(channel, method_frame, header_frame, body):    # 消息确认， 确认之后才会删除消息并给消费者发送新的消息    channel.basic_ack(delivery_tag=method_frame.delivery_tag)    print(body)parameters = pika.URLParameters('amqp://guest:guest@localhost:5672/%2F')connection = pika.BlockingConnection(parameters)channel = connection.channel()channel.exchange_declare(exchange='web_develop', exchange_type='direct',                         passive=False, durable=True, auto_delete=False)# 声明队列， 如果没有就创建channel.queue_declare(queue='standard', auto_delete=True)# 通过路由键将队列和交换机绑定channel.queue_bind(queue='standard', exchange='web_develop',                   routing_key='xxx_routing_key')channel.basic_consume('standard', on_message)  # 订阅队列try:    channel.start_consuming()  # 开始消费except KeyboardInterrupt:    channel.stop_consuming()  # 退出消费connection.close()

5、在另一个终端，再启动发布者，发布消息

amqp_producer.py

# coding=utf-8import sysimport pika# %2F是被转义的/， 这里使用了默认的虚拟主机、 默认的guest这个账号和密码parameters = pika.URLParameters('amqp://guest:guest@localhost:5672/%2F')connection = pika.BlockingConnection(parameters)  # connection就是所谓的消息代理channel = connection.channel()  # 获得信道# 声明交换机，指定交换类型为直接交换。 最后2个参数表示想要持久化的交换机channel.exchange_declare(exchange='web_develop', exchange_type='direct',                         passive=False, durable=True, auto_delete=False)if len(sys.argv) != 1:    msg = sys.argv[1]  # 使用命令行参数作为消息体else:    msg = 'hah'# 创建一个消息， delivery_mode为2表示让这个消息持久化， 重启RabbitMQ也不会丢失props = pika.BasicProperties(content_type='text/plain', delivery_mode=2)# basic_publish表示发送路由键为xxx_routing_key，消息体为haha的消息给web_develop这个交换机channel.basic_publish('web_develop', 'xxx_routing_key', msg,                      properties=props)connection.close()  # 关闭连接

3、Rabbit MQ使用Linux

3.1 启动rabbitmq

启动命令需要到一个其他路径下运行哦

# systemctl start rabbitmq-server

或者使用：

rabbitmq-server -detached           //启动rabbitmq，-detached代表后台守护进程方式启动

查看rabbitmq 启动后的情况 ,如图：

# rabbitmqctl status

其他相关命令

启动服务：rabbitmq-server-detached【/usr/local/rabbitmq/sbin/rabbitmq-server -detached 】查看状态：rabbitmqctl status【/usr/local/rabbitmq/sbin/rabbitmqctl status 】关闭服务：rabbitmqctl stop【/usr/local/rabbitmq/sbin/rabbitmqctl stop 】列出角色：rabbitmqctl list_users

3.2 配置网页插件：

首先创建目录，否则可能报错：

# mkdir /etc/rabbitmq

然后启用插件：

# rabbitmq-plugins enable rabbitmq_management

3.3 配置防火墙

配置linux 端口 15672 网页管理 5672 AMQP端口：

# firewall-cmd --permanent --add-port=15672/tcp

# firewall-cmd --permanent --add-port=5672/tcp

# systemctl restart firewalld.service

Centos 7 的特性，需要注册开放的端口进防火墙，并重启防火墙。

3.4 web端管理界面

现在你在浏览器中输入服务器IP:15672就可以看到RabbitMQ的WEB管理页面了，是不是很兴奋，可是你没有账号密码，别急

http://168.168.88.129:15672/

3.5 配置web端访问账号密码和权限

默认网页是不允许访问的，需要增加一个用户修改一下权限，代码如下：

//添加用户，后面两个参数分别是用户名和密码，我这都用superrd了

# rabbitmqctl add_user liuxing liuxing

# rabbitmqctl set_permissions -p / liuxing ".*" ".*" ".*" //添加权限

# rabbitmqctl set_user_tags liuxing administrator //修改用户角色

如果是在vagrant中部署的rabbitmq-server需要开通端口：

Vagrant.configure("2") do |config|    ....  config.vm.network :forwarded_port, guest: 15672, host: 15672  config.vm.network :forwarded_port, guest: 5672, host: 5672

3.6 其他配置

其他参考：