码迷,mamicode.com
首页 > 其他好文 > 详细

RabbitMQ 消息中间件的部署

时间:2015-06-05 12:12:32      阅读:213      评论:0      收藏:0      [点我收藏+]

标签:

一、RabbitMQ安装

安装步骤:

1、  安装Erlang  

2、  安装RabbitMQ server

3、  配置环境

4、  开启服务

安装Erlang:

       安装最新版本Erlang

       配置环境变量:ERLANG_HOME=D:\RabbitMQ\erl5.9.2,Erlang的安装目录。

      技术分享

安装RabbitMQ服务器:

       下载rabbitmq-server-windows-2.8.7.zip,解压到D:\RabbitMQ\rabbitmq_server-2.8.7。

       sbin目录下存放RabbitMQ server的控制命令,RabbitMQ server可以以服务和应用程序两种方式运行。

       rabbitmq-server.bat 是应用程序方式运行。

       rabbitmq-service.bat 是服务方式运行。

       rabbitmqctl.bat 管理插件,管理日志、输出、通过命令访问等。

      

       配置系统路径:RABBITMQ_SERVER = D:\RabbitMQ\rabbitmq_server-2.8.7。

       在path中加入“;% RABBITMQ_SERVER%\sbin”。

      

RabbitMQ日志等数据在“C:\Documents and Settings\gzcheng\Application Data\RabbitMQ”目录下。

 

运行RabbitMQ:

在Dos窗口中输入rabbitmq-server.bat运行RabbitMQ。

关闭RabbitMQ应用程序,可以直接关闭RabbitMQ,也可以通过rabbitmqctl进行命令管理。在另外一个DOS窗口运行:rabbitmqctl stop。用rabbitmqctl status,可以查看RabbitMQ状态。

RabbitMQ默认端口是:5672。

 

二、编译RabbitMQ-C

1、安装python,要使用2.*版python,如:python-2.7.3.msi。

2、安装cmake,cmake-2.8.10.1-win32-x86.zip。

3、下载rabbitmq-c-master.zip,解压rabbitmq-c-master.zip。

4、下载rabbitmq-codegen-master.zip,解压rabbitmq-codegen-master.zip,将解压内容拷贝到rabbitmq-c-master 中的 codegen 目录下。

5、运行cmake-gui.exe,设置source code路径设置rabbitmq-c-master目录,输出设置自定义的目录。

6、单击配置,选择编译器类型,VS2005或者VS2008。

 技术分享

 

三、RabbitMQ-C的API说明

1、amqp_connection_state_t amqp_new_connection(void)

接口说明:声明一个新的amqp connection

2、int amqp_open_socket(char const *hostname, int portnumber)

接口说明:获取socket

参数说明:hostname RabbitMQ server所在主机;portnumber RabbitMQ server监听端口

3、void amqp_set_sockfd(amqp_connection_state_t state,int sockfd)

接口说明:将amqp connection和sockfd进行绑定

4、amqp_rpc_reply_t amqp_login(amqp_connection_state_t state, char const *vhost,int channel_max,int frame_max,int heartbeat,amqp_sasl_method_enum sasl_method, ...)

接口说明:用于登录RabbitMQ server,主要目的为了进行权限管理;

    参数说明:state    amqp connection

           vhost   rabbit-mq的虚机主机,是rabbit-mq进行权限管理的最小单位

channel_max  最大链接数,此处设成0即可

frame_max  和客户端通信时所允许的最大的frame size.默认值为131072,增大这个值有助于提高吞吐,降低这个值有利于降低时延

heartbeat 含义未知,默认值填0

sasl_method  用于SSL鉴权,默认值参考后文demo

5、amqp_channel_open_ok_t *amqp_channel_open(amqp_connection_state_t state, amqp_channel_t channel)

接口说明:用于关联conn和channel

6、amqp_exchange_declare_ok_t *amqp_exchange_declare(amqp_connection_state_t state, amqp_channel_t channel, amqp_bytes_t exchange, amqp_bytes_t type, amqp_boolean_t passive, amqp_boolean_t durable, amqp_table_t arguments)

接口说明:声明declare

参数说明:state

channel

exchange

type  "fanout"  "direct" "topic"三选一

passive

curable

arguments

7、amqp_queue_declare_ok_t *amqp_queue_declare(amqp_connection_state_t state, amqp_channel_t channel, amqp_bytes_t queue, amqp_boolean_t passive, amqp_boolean_t durable, amqp_boolean_t exclusive, amqp_boolean_t auto_delete, amqp_table_t arguments)

接口说明:声明queue

参数说明:state   amqp connection

channel

queue  queue name

passive

durable  队列是否持久化

exclusive  当前连接不在时,队列是否自动删除

aoto_delete 没有consumer时,队列是否自动删除

arguments 用于拓展参数,比如x-ha-policy用于mirrored queue

8、amqp_queue_bind_ok_t *amqp_queue_bind(amqp_connection_state_t state, amqp_channel_t channel, amqp_bytes_t queue, amqp_bytes_t exchange, amqp_bytes_t routing_key, amqp_tab le_t arguments)

接口说明:声明binding

9、amqp_basic_qos_ok_t *amqp_basic_qos(amqp_connection_state_t state, amqp_channel_t channel, uint32_t prefetch_size, uint16_t prefetch_count, amqp_boolean_t global)

接口说明:qos是 quality of service,我们这里使用主要用于控制预取消息数,避免消息按条数均匀分配,需要和no_ack配合使用

参数说明:state

channel

prefetch_size 以bytes为单位,0为unlimited

prefetch_count 预取的消息条数

global

10、amqp_basic_consume_ok_t *amqp_basic_consume(amqp_connection_state_t state, amqp_channel_t channel, amqp_bytes_t queue, amqp_bytes_t consumer_tag, amqp_boolean_t no_local, amqp_boolean_t no_ack, amqp_boolean_t exclusive, amqp_table_t arguments)

接口说明:开始一个queue consumer

参数说明:state

channel

queue

consumer_tag

no_local

no_ack    是否需要确认消息后再从队列中删除消息

exclusive

arguments

11、int amqp_basic_publish(amqp_connection_state_t state,amqp_channel_t channel,amqp_bytes_t exchange,amqp_bytes_t routing_key,amqp_boolean_t mandatory,amqp_boolean_t immediate,struct amqp_basic_properties_t_ const *properties,amqp_bytes_t body)

接口说明:发布消息

参数说明:state

channel

exchange

routing_key  当exchange为默认“”时,此处填写queue_name,当exchange为direct,此处为binding_key

mandatory

immediate

properties 更多属性,如何设置消息持久化

body 消息体

12、amqp_rpc_reply_t amqp_channel_close(amqp_connection_state_t state,amqp_channel_t channel,int code)

       接口说明:关闭channel

13、amqp_rpc_reply_t amqp_connection_close(amqp_connection_state_t state,int code)

       接口说明:关闭connection

14、int amqp_destroy_connection(amqp_connection_state_t state)

       接口说明:销毁connection

 

四、RabbitMQ服务客户端的的业务逻辑

    1,打开 socket:

     amqp_new_connection();

     amqp_open_socket(hostname, port);

    2,用户登陆:

    amqp_set_sockfd(conn, sockfd);

    amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, userName, password);

    3,打开channel

    amqp_channel_open(conn, 1);

    amqp_get_rpc_reply(conn);

    4,声明 Exchange

amqp_exchange_declare(conn, 1, amqp_cstring_bytes(exchange), amqp_cstring_bytes("fanout"), 0, 0, amqp_empty_table);

    5,声明Queue

    amqp_queue_declare_ok_t *r = amqp_queue_declare(conn, 1, amqp_cstring_bytes(queue), 0, 0, 0, 1,amqp_empty_table);

    6,将Queue和Exchange进行binding

    amqp_queue_bind(conn, 1,

                    queuename,

                    amqp_cstring_bytes(exchange),

                    amqp_cstring_bytes(bindingkey),

                    amqp_empty_table);

    注意:binding动作在程序中只能执行一次,如果第二次再执行,程序会crash!

    7,操作,包括 发送(publish),接收(consume)等。

        发送:

            amqp_basic_publish(conn,

                           1,

                           amqp_cstring_bytes(exchange),

                           amqp_cstring_bytes(routingkey),

                           0,

                           0,

                           NULL,

                           amqp_cstring_bytes(messagebody));

        接收:

            amqp_basic_consume(conn, 1,queuename, amqp_empty_bytes, 0, 1, 0, amqp_empty_table);

            amqp_maybe_release_buffers(conn);

    此处应该注意:接受的过程是一个阻塞的异步过程,所以必须在子线程中进行操作,这样就不会影响主线程中的UI操作,所以在以前使用的block编程就用了很大的用武之地,一方面,通过block多核编程提高程序的运行效率,第二方面,异步的dispatch能够完美的解决阻塞的问题,并且可以使处理后返回的数据直接在OC类中直接使用,从而规避了在C函数中传递OC的指针来对OC的对象进行的操作,真是一劳永逸的好方法,推荐大家使用。

    8,进行unbinding

           amqp_queue_unbind(conn, 1,

                      queuename,

                      amqp_cstring_bytes(exchange),

                      amqp_cstring_bytes(bindingkey),

                      amqp_empty_table);

    9,关闭channel

            amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS);

    10,关闭 connection连接。

    // Closing connection

    amqp_connection_close(conn, AMQP_REPLY_SUCCESS);

    //Ending connection

amqp_destroy_connection(conn);

 

五、RabbitMQ虚拟主机

虚拟主机virtual host只起一个命名空间的作用,多个用户user可以共用一个虚拟主机。系统默认虚拟主机是‘/’。要注意不同命名空间之间的资源是不能访问的,比如exchange、queue、binging等。

可以通过rabbitmqctl创建自己虚拟主机。

1、创建用户:rabbitmqctl add_user test 123456;新建的用户名是test,密码是123456,通过rabbitmqctl list_users查看有多少用户。

2、新建虚拟主机:rabbitmqctl add_vhosts test_host;新建一个test_host虚拟主机,通过rabbitmqctl list_vhosts查看有几个虚拟主机。

3、设置访问权限:rabbitmqctl set_permissions –p test_host test “.*” “.*” “.*”;test用户就有读写test_host虚拟主机的权限。

4、删除用户:rabbitmqctl delete_user guest。

 

六、RabbitMQ主要概念

1、Exchange(交换机)

       有三个类型:“Direct”点对点直接传输,Publisher和Consumer直接连接,不处理路由键;“Fanout”不处理路由键,消息发送到与Exchange绑定的所有队列上;“Topic”主题订阅,Consumer可以使用统配符来定义routingKey,使用相同的topic类型,并且符合该统配符的Publisher发布的信息,Consumer都可以收到。

       程序指定类型字符串:“direct”、“topic”、“fanout”、“headers”。要修改系统默认类型,需要用amqp_exchange_declare声明,声明后才能用。

       交换机可以是持久的、临时的和自动删除的。持久的就是一直存在于服务器;临时的会工作到RabbitMQ被关闭为止;自动删除是指当没程序使用就自动删除。

2、Queue(消息队列)

       消息队列可以是持久的、临时的和自动删除的。消息队列是用来替消费者保存消息,采用先进先出规则。消息队列会跟踪消息的获取情况,消息要出队就必须被获取(acquire和consume是两个动作,先执行acquire,相当于对消息加锁),这阻止多个客户端同时获取和消费同一条消息,也可以被用于做单队列多消费者之间的负载均衡。

3、持久化

Queue、消息和Exchange都要持久。

4、绑定

      

七、数据发送和接收规则

1、一个Connection只能对应一个接收线程,不能用多线程结果一个Connection。

2、通过一个connection上可以创建多个通道,从不同队列上接收数据。

 

八、RabbitMQ管理

1、  开启Management Plugin

在DOS窗口中输入:rabbitmq-plugins enable rabbitmq_management

管理页面地址:http://server-name:15672/,早期端口是:55672。

Management Plugin Web地址: http://www.rabbitmq.com/management.html

2、  RabbitMQ配置

一般情况下,RabbitMQ的默认配置就足够了,如果希望特殊设置的话,有两个途径:

一个是环境变量的配置文件rabbitmq-env.conf,只适用非Windows环境,Windows环境需要通过Start > Settings > Control Panel > System > Advanced > Environment Variables来设置环境变量。设置好后,还要重新安装RabbitMQ Service。

一个是配置信息的配置文件rabbitmq.config,这个文件内容必须符合erlang配置文件的标准。它既有默认的目录,也可以通过RABBITMQ_CONFIG_FILE环境变量进行设置。

详细说明见http://www.rabbitmq.com/configure.html

 

环境变量配置(配置端口,文件路径,名称等)

RABBITMQ_NODE_IP_ADDRESS 默认为空 为空时表示允许所有IP连接,如果你添加约束配置这一个选项

RABBITMQ_NODE_PORT 默认为5672 端口 端口

HOSTNAME 默认为HOST名称 当前机器HOST名称

COMPUTERNAME windows默认:localhost 当前机器名称

RABBITMQ_BASE Windows默认: %APPDATA%\RabbitMQ 安装目录

RABBITMQ_NODENAME Unix*:rabbit@$HOSTNAME Windows: rabbit@%COMPUTERNAME% 节点名称

RABBITMQ_CONFIG_FILE Unix*:/etc/rabbitmq/rabbitmq Windows: %RABBITMQ_BASE%\rabbitmq 配置文件

RABBITMQ_MNESIA_BASE Unix*:/var/lib/rabbitmq/mnesia Windows: %RABBITMQ_BASE%\db mq数据库目录

RABBITMQ_LOG_BASE Unix*:/var/log/rabbitmq Windows: %RABBITMQ_BASE%\log LOG文件目录

RABBITMQ_PLUGINS_DIR 默认为空 插件目录

RABBITMQ_ENABLED_PLUGINS_FILE Unix*:/etc/rabbitmq/enabled_plugins Windows: %RABBITMQ_BASE%\enabled_plugins 记录激活的插件

ERLANG_SERVICE_MANAGER_PATH WindowsService: %ERLANG_HOME%\erts-x.x.x\bin erlang安装目录

RABBITMQ_SERVICENAME Windows Service: RabbitMQ rabbitmq服务名称

RABBITMQ_CONSOLE_LOG Windows Service: 控制台日志

 

RabbitMQ 消息中间件的部署

标签:

原文地址:http://www.cnblogs.com/SamRichard/p/4554185.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!