安装
ubuntu下参考：
启动/关闭：
$ sudo service rabbitmq-server start|stop
修改文件打开数量的限制，提高并发性能
ulimit命令可查看限制信息：
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.
完成下面的两个步骤，电脑重启才能生效。
1、 引入pam_limits.so
对于ubuntu，在/etc/pam.d/common-session中加入：
session required pam_limits.so
如果/etc/pam.d/common-session-noninteractive存在，也要加入上面的代码。
2、修改/etc/security/limits.conf
加入以下内容：
* soft nofile 65536* hard nofile 65536
数值可以继续调大。soft的值不能大于hard的值。
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 -n65536
另外一个更加方便的方法是：在运行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:####Where:# 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, must be# the literal username root.## can have the two values:# - "soft" for enforcing the soft limits# - "hard" for enforcing hard limits## - 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)##
##* 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)
配置文件：
中文入门教程阅读笔记
教程在，是官方教程的翻译。
安装python库
sudo pip install pika==0.9.5
一些概念
使用“交换机”（exchange）进行路由。
发消息时必须指定一个存在的队列，或者创建一个队列，然后向该队列发送消息。
在RabbitMQ中，消息是不能直接发送到队列，它需要发送到交换机（exchange）中。
RabbitMq不允许你使用不同的参数重新定义一个队列，它会返回一个错误。
队列可以设置为持久化，重启之后依然存在设置为持久化的队列。不过队列持久化，不代表着消息会持久化，所以消息也要设置持久化参数。RabbitMQ中消息的持久化有些问题：因为RabbitMq并不是所有的消息都使用fsync(2)——它有可能只是保存到缓存中，并不一定会马上写到硬盘中。
python pika中一个channel对应一个消息队列。
消息响应默认是开启的。pika中需要显式编写代码来响应。
一个消息可以只发送给一个消费者，也可以发送给多个消费者。发送给多个消费者的模式叫做“订阅/发布模式”。
一个交换机上可以绑定若干队列。消息从生产者发送给交换机，交换机根据指定的策略，将消息发送给它绑定的队列。
命令
查看有哪些队列？队列中有多少消息？
$ sudo rabbitmqctl list_queuesListing queues ...hello 0...done.
查看所有的交换器及其类型：
$ sudo rabbitmqctl list_exchangesListing exchanges ...logs fanoutamq.direct directamq.topic topicamq.fanout fanoutamq.headers headers...done.
amq.*的交换器是默认创建的。
查看交换机和队列的绑定关系：
$ sudo rabbitmqctl list_bindings
代码示例1
来自
send.py:
#!/usr/bin/env pythonimport pikaconnection = 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 pythonimport pikaconnection = 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
来自
new_task.py:
#!/usr/bin/env pythonimport pikaimport sysconnection = 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 pythonimport pikaimport timeconnection = 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
来自
emit_log.py ：
#!/usr/bin/env pythonimport pikaimport sysconnection = 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 pythonimport pikaconnection = 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.queuechannel.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
来自。
emit_log_direct.py：
#!/usr/bin/env pythonimport pikaimport sysconnection = 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 pythonimport pikaimport sysconnection = 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.queueseverities = 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
来自。
emit_log_topic.py：
#!/usr/bin/env pythonimport pikaimport sysconnection = 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 pythonimport pikaimport sysconnection = 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.queuebinding_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()