RabbitMQ安装、配置、基本使用

时间 2019-11-11

原文原文链接

安装

ubuntu下参考： https://www.rabbitmq.com/install-debian.htmlhtml

启动/关闭：python

$ sudo service rabbitmq-server start|stop

修改文件打开数量的限制，提升并发性能

ulimit命令可查看限制信息：shell

Syntax
      ulimit [-acdfHlmnpsStuv] [limit]

Options

   -S   Change and report the soft limit associated with a resource. 
   -H   Change and report the hard limit associated with a resource. 

   -a   All current limits are reported. 
   -c   The maximum size of core files created. 
   -d   The maximum size of a process's data segment. 
   -f   The maximum size of files created by the shell(default option) 
   -l   The maximum size that can be locked into memory. 
   -m   The maximum resident set size. 
   -n   The maximum number of open file descriptors. 
   -p   The pipe buffer size. 
   -s   The maximum stack size. 
   -t   The maximum amount of cpu time in seconds. 
   -u   The maximum number of processes available to a single user. 
   -v   The maximum amount of virtual memory available to the process.

完成下面的两个步骤，电脑重启才能生效。ubuntu

一、引入pam_limits.so缓存

对于ubuntu，在/etc/pam.d/common-session中加入：session

session    required   pam_limits.so

若是/etc/pam.d/common-session-noninteractive存在，也要加入上面的代码。并发

二、修改/etc/security/limits.confapp

加入如下内容：dom

*               soft     nofile          65536
*               hard     nofile          65536

数值能够继续调大。soft的值不能大于hard的值。ide

hard限制智能root修改，soft的限制能够是进程本身修改。

The hard limit is the ceiling for the soft limit. The soft limit is what is actually enforced for a session or process. This allows the administrator (or user) to set the hard limit to the maximum usage they wish to allow. Other users and processes can then use the soft limit to self-limit their resource usage to even lower levels if they so desire.

重启电脑后：

$ ulimit -n
65536

另一个更加方便的方法是：在运行rabbitmq-server以前运行ulimit命令：

ulimit -n 32768

能够把这个命令放入/etc/default/rabbitmq-server文件中。

limits.conf文件自带说明：

$ cat /etc/security/limits.conf 
# /etc/security/limits.conf
#
#Each line describes a limit for a user in the form:
#
#<domain>        <type>  <item>  <value>
#
#Where:
#<domain> can be:
#        - a user name
#        - a group name, with @group syntax
#        - the wildcard *, for default entry
#        - the wildcard %, can be also used with %group syntax,
#                 for maxlogin limit
#        - NOTE: group and wildcard limits are not applied to root.
#          To apply a limit to the root user, <domain> must be
#          the literal username root.
#
#<type> can have the two values:
#        - "soft" for enforcing the soft limits
#        - "hard" for enforcing hard limits
#
#<item> can be one of the following:
#        - core - limits the core file size (KB)
#        - data - max data size (KB)
#        - fsize - maximum filesize (KB)
#        - memlock - max locked-in-memory address space (KB)
#        - nofile - max number of open files
#        - rss - max resident set size (KB)
#        - stack - max stack size (KB)
#        - cpu - max CPU time (MIN)
#        - nproc - max number of processes
#        - as - address space limit (KB)
#        - maxlogins - max number of logins for this user
#        - maxsyslogins - max number of logins on the system
#        - priority - the priority to run user process with
#        - locks - max number of file locks the user can hold
#        - sigpending - max number of pending signals
#        - msgqueue - max memory used by POSIX message queues (bytes)
#        - nice - max nice priority allowed to raise to values: [-20, 19]
#        - rtprio - max realtime priority
#        - chroot - change root to directory (Debian-specific)
#
#<domain>      <type>  <item>         <value>
#

#*               soft    core            0
#root            hard    core            100000
#*               hard    rss             10000
#@student        hard    nproc           20
#@faculty        soft    nproc           20
#@faculty        hard    nproc           50
#ftp             hard    nproc           0
#ftp             -       chroot          /ftp
#@student        -       maxlogins       4

# End of file

配置

端口占用：

4369 (epmd), 25672 (Erlang distribution)
5672, 5671 (AMQP 0-9-1 without and with TLS)
15672 (if management plugin is enabled)
61613, 61614 (if STOMP is enabled)
1883, 8883 (if MQTT is enabled)

配置文件：

Configuration - 官网
 RabbitMQ的安装，配置，监控

中文入门教程阅读笔记

教程在http://rabbitmq-into-chinese.readthedocs.org/zh_CN/latest/，是官方教程的翻译。

安装python库

sudo pip install pika==0.9.5

一些概念

使用“交换机”（exchange）进行路由。

发消息时必须指定一个存在的队列，或者建立一个队列，而后向该队列发送消息。

在RabbitMQ中，消息是不能直接发送到队列，它须要发送到交换机（exchange）中。

RabbitMq不容许你使用不一样的参数从新定义一个队列，它会返回一个错误。

队列能够设置为持久化，重启以后依然存在设置为持久化的队列。不过队列持久化，不表明着消息会持久化，因此消息也要设置持久化参数。RabbitMQ中消息的持久化有些问题：由于RabbitMq并非全部的消息都使用fsync(2)——它有可能只是保存到缓存中，并不必定会立刻写到硬盘中。

python pika中一个channel对应一个消息队列。

消息响应默认是开启的。pika中须要显式编写代码来响应。

一个消息能够只发送给一个消费者，也能够发送给多个消费者。发送给多个消费者的模式叫作“订阅/发布模式”。

一个交换机上能够绑定若干队列。消息从生产者发送给交换机，交换机根据指定的策略，将消息发送给它绑定的队列。

命令

查看有哪些队列？队列中有多少消息？

$ sudo rabbitmqctl list_queues
Listing queues ...
hello    0
...done.

查看全部的交换器及其类型：

$ sudo rabbitmqctl list_exchanges
Listing exchanges ...
logs      fanout
amq.direct      direct
amq.topic       topic
amq.fanout      fanout
amq.headers     headers
...done.

amq.*的交换器是默认建立的。

查看交换机和队列的绑定关系：

$ sudo rabbitmqctl list_bindings

代码示例1

来自http://rabbitmq-into-chinese.readthedocs.org/zh_CN/latest/tutorials_with_python/%5B1%5DHello_World/

send.py:

#!/usr/bin/env python
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()

channel.queue_declare(queue='hello')

channel.basic_publish(exchange='',
                      routing_key='hello',
                      body='Hello World!')
print " [x] Sent 'Hello World!'"
connection.close()

receive.py:

#!/usr/bin/env python
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()

channel.queue_declare(queue='hello')

print ' [*] Waiting for messages. To exit press CTRL+C'

def callback(ch, method, properties, body):
    print " [x] Received %r" % (body,)

channel.basic_consume(callback,
                      queue='hello',
                      no_ack=True)

channel.start_consuming()

代码示例2

来自http://rabbitmq-into-chinese.readthedocs.org/zh_CN/latest/tutorials_with_python/%5B2%5DWork_Queues/

new_task.py:

#!/usr/bin/env python
import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()

channel.queue_declare(queue='task_queue', durable=True)

message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(exchange='',
                      routing_key='task_queue',
                      body=message,
                      properties=pika.BasicProperties(
                         delivery_mode = 2, # make message persistent
                      ))
print " [x] Sent %r" % (message,)
connection.close()

worker.py:

#!/usr/bin/env python
import pika
import time

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()

channel.queue_declare(queue='task_queue', durable=True)
print ' [*] Waiting for messages. To exit press CTRL+C'

def callback(ch, method, properties, body):
    print " [x] Received %r" % (body,)
    time.sleep( body.count('.') )
    print " [x] Done"
    ch.basic_ack(delivery_tag = method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
                      queue='task_queue')

channel.start_consuming()

代码示例3

来自http://rabbitmq-into-chinese.readthedocs.org/zh_CN/latest/tutorials_with_python/%5B3%5DPublish_Subscribe/

emit_log.py ：

#!/usr/bin/env python
import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='logs',
                         type='fanout')

message = ' '.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange='logs',
                      routing_key='',
                      body=message)
print " [x] Sent %r" % (message,)
connection.close()

receive_logs.py ：

#!/usr/bin/env python
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='logs',
                         type='fanout')

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

channel.queue_bind(exchange='logs',
                   queue=queue_name)

print ' [*] Waiting for logs. To exit press CTRL+C'

def callback(ch, method, properties, body):
    print " [x] %r" % (body,)

channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)

channel.start_consuming()

代码示例4

来自http://rabbitmq-into-chinese.readthedocs.org/zh_CN/latest/tutorials_with_python/%5B4%5DRouting/。

emit_log_direct.py：

#!/usr/bin/env python
import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs',
                         type='direct')

severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='direct_logs',
                      routing_key=severity,
                      body=message)
print " [x] Sent %r:%r" % (severity, message)
connection.close()

receive_logs_direct.py：

#!/usr/bin/env python
import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs',
                         type='direct')

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

severities = sys.argv[1:]
if not severities:
    print >> sys.stderr, "Usage: %s [info] [warning] [error]" % \
                         (sys.argv[0],)
    sys.exit(1)

for severity in severities:
    channel.queue_bind(exchange='direct_logs',
                       queue=queue_name,
                       routing_key=severity)

print ' [*] Waiting for logs. To exit press CTRL+C'

def callback(ch, method, properties, body):
    print " [x] %r:%r" % (method.routing_key, body,)

channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)

channel.start_consuming()

代码示例5

来自http://rabbitmq-into-chinese.readthedocs.org/zh_CN/latest/tutorials_with_python/%5B5%5DTopics/。

emit_log_topic.py：

#!/usr/bin/env python
import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='topic_logs',
                         type='topic')

routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='topic_logs',
                      routing_key=routing_key,
                      body=message)
print " [x] Sent %r:%r" % (routing_key, message)
connection.close()

receive_logs_topic.py：

#!/usr/bin/env python
import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='topic_logs',
                         type='topic')

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

binding_keys = sys.argv[1:]
if not binding_keys:
    print >> sys.stderr, "Usage: %s [binding_key]..." % (sys.argv[0],)
    sys.exit(1)

for binding_key in binding_keys:
    channel.queue_bind(exchange='topic_logs',
                       queue=queue_name,
                       routing_key=binding_key)

print ' [*] Waiting for logs. To exit press CTRL+C'

def callback(ch, method, properties, body):
    print " [x] %r:%r" % (method.routing_key, body,)

channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)

channel.start_consuming()

参考：

Installing RabitMQ on Debian / Ubuntu

Permanently set process limit

深刻Linux PAM体系结构 - 红黑联盟

Pluggable authentication module - wikipedia

Soft limit vs Hard limit?

ulimit: difference between hard and soft limits

官方英文入门教程