RabbitMQ研究（一）安装与入门

叁歲伎倆 2022-03-26 07:39 264阅读 0赞

# 简介 #

## MQ ##

MQ是指利用高校可靠的消息传递机制与平台无关的数据交流，并基于数据通信来进行的分布式系统的集成。

主流的MQ最常用的有两种模式：点对点P2P、订阅/发布模式。  
点对点主要是基于队列，消息生产者发送到队列里面，消息消费者从队列里面获取，小型项目，可以直接使用内存Queue，或者直接用Redis的List作为队列。发布订阅定义了一个内容节点，这个内容在MQ里面成为Topic（主题），主题可以认为是消息传递的中介，消息发布者将消息发布到某个主题，而消息订阅者则从主题中订阅消息，主要应用场景是广播一对多的时候。

主流的开源的MQ：RabbitMQ、Kafka、ActiveMQ、RocketMQ、ZeroMQ（现在用的较少）  
以下指标是我根据网上的自己一个一个去找了资料验证补充过后的

<table> 
 <thead> 
 <tr> 
 <th>特性</th> 
 <th>ActiveMQ</th> 
 <th>RabbitMQ</th> 
 <th>RocketMQ</th> 
 <th>Kafka</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td>提供者-消费者</td> 
 <td>支持</td> 
 <td>支持</td> 
 <td>支持</td> 
 <td>支持</td> 
 </tr> 
 <tr> 
 <td>订阅-发布</td> 
 <td>支持</td> 
 <td>支持</td> 
 <td>支持</td> 
 <td>支持</td> 
 </tr> 
 <tr> 
 <td>request-replay(请求回应)</td> 
 <td>支持</td> 
 <td>支持</td> 
 <td>不支持</td> 
 <td>不支持</td> 
 </tr> 
 <tr> 
 <td>API和文档完备性</td> 
 <td>高质量</td> 
 <td>高质量</td> 
 <td>高质量</td> 
 <td>高质量</td> 
 </tr> 
 <tr> 
 <td>多语言支持</td> 
 <td>支持，Java优先</td> 
 <td>支持多语言</td> 
 <td>只支持Java</td> 
 <td>支持，Java优先</td> 
 </tr> 
 <tr> 
 <td>单机吞吐量</td> 
 <td>万级</td> 
 <td>万级</td> 
 <td>万级</td> 
 <td>十万级</td> 
 </tr> 
 <tr> 
 <td>消息延迟</td> 
 <td></td> 
 <td>微秒级</td> 
 <td>毫秒级</td> 
 <td>毫秒级</td> 
 </tr> 
 <tr> 
 <td>可用性</td> 
 <td>高（主从）</td> 
 <td>高（主从）</td> 
 <td>非常高（分布式）</td> 
 <td>非常高（分布式）</td> 
 </tr> 
 <tr> 
 <td>消息丢失</td> 
 <td>低</td> 
 <td>低</td> 
 <td>理论上不会丢失</td> 
 <td>理论上不会丢失</td> 
 </tr> 
 <tr> 
 <td>社区活跃度</td> 
 <td>高</td> 
 <td>高</td> 
 <td>中</td> 
 <td>高</td> 
 </tr> 
 <tr> 
 <td>成熟度</td> 
 <td>成熟</td> 
 <td>成熟</td> 
 <td>比较成熟</td> 
 <td>成熟日志领域</td> 
 </tr> 
 <tr> 
 <td>特点</td> 
 <td>功能齐全</td> 
 <td>Erlang语言，性能好</td> 
 <td>各个环节分布式</td> 
 <td></td> 
 </tr> 
 <tr> 
 <td>支持协议</td> 
 <td>STOMP、REST、XMPP、AMQP</td> 
 <td>AMQP</td> 
 <td>自己定义的</td> 
 <td></td> 
 </tr> 
 <tr> 
 <td>持久化</td> 
 <td>内存、文件、数据库</td> 
 <td>内容、文件</td> 
 <td>磁盘文件</td> 
 <td>内存、磁盘</td> 
 </tr> 
 <tr> 
 <td>事务</td> 
 <td>支持</td> 
 <td>支持</td> 
 <td>支持</td> 
 <td></td> 
 </tr> 
 <tr> 
 <td>负载均衡</td> 
 <td>支持</td> 
 <td>支持</td> 
 <td>支持</td> 
 <td></td> 
 </tr> 
 <tr> 
 <td>优先级队列</td> 
 <td>支持</td> 
 <td>支持</td> 
 <td>支持（分区顺序）</td> 
 <td>不支持</td> 
 </tr> 
 <tr> 
 <td>延迟队列</td> 
 <td>支持</td> 
 <td>支持</td> 
 <td>支持</td> 
 <td>不支持</td> 
 </tr> 
 <tr> 
 <td>死信队列</td> 
 <td>支持</td> 
 <td>支持</td> 
 <td>支持</td> 
 <td>不支持</td> 
 </tr> 
 <tr> 
 <td>评价</td> 
 <td>优点：成熟的产品，大规模使用 缺点：会丢失消息，不适用于上千个队列（Queue的数量少）的应用场景</td> 
 <td>优点：管理界面友好 缺点：不支持动态扩展</td> 
 <td>优点：接口模型接单，性能非常好，可以大量堆积消息在Broker 缺点：新文档缺乏</td> 
 <td></td> 
 </tr> 
 </tbody> 
</table>

## 作用 ##

**异步解耦**：各个模块之间异步解耦  
**冗余（存储）**：比如A系统调用B系统，如果调用一次，失败了，在代码中不进行二次调用，都不会再次成功。而MQ则可以持久化，除非删掉这个消息  
**扩展性**：因为消息本身已经解耦了，入队和处理效率很高，扩展性也很高，代码修改起来较容易  
**削峰**：访问量巨大的时候，有削峰效果  
**可恢复性**：一部分组件失效，不会影响到整个系统，直接恢复即可

# 安装 #

## windows ##

*  安装erlang语言  
    [http://www.erlang.org/downloads][http_www.erlang.org_downloads]  
    下载，然后安装，之后再配置`PATH`环境变量到bin目录  
    验证erlang语言，进入CMD里面输入`erl`  
    如果出现如下则表示环境安装成功

Eshell V10.2  (abort with ^G)
    1>

*  安装RabbitMQ  
    [http://www.rabbitmq.com/download.html][http_www.rabbitmq.com_download.html]

找到 `Windows: Installer (recommended) | Binary build`，选择`Binary`点击即可，解压，下载放到合适的位置，DOS进入到目录下面的sbin，输入`rabbitmq-plugins enable rabbitmq_management`安装，如果出现如下，表示安装成功

F:\software\rabbitmq-server-windows-3.7.10\rabbitmq_server-3.7.10\sbin>rabbitmq-plugins enable rabbitmq_management
    Enabling plugins on node rabbit@DESKTOP-2VOJNE8:
    rabbitmq_management
    The following plugins have been configured:
      rabbitmq_management
      rabbitmq_management_agent
      rabbitmq_web_dispatch
    Applying plugin configuration to rabbit@DESKTOP-2VOJNE8...
    The following plugins have been enabled:
      rabbitmq_management
      rabbitmq_management_agent
      rabbitmq_web_dispatch
    
    set 3 plugins.
    Offline change; changes will take effect at broker restart.

启动RabbitMQ：打开sbin目录，双击rabbitmq-server.bat，出现如下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d1NjY2MDU2Mw_size_16_color_FFFFFF_t_70]

访问管理界面，打开浏览器：[http://localhost:15672/][http_localhost_15672] 输入用户密码:guest/guest

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d1NjY2MDU2Mw_size_16_color_FFFFFF_t_70 1]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d1NjY2MDU2Mw_size_16_color_FFFFFF_t_70 2]

安装成功

添加用户并授权

F:\software\rabbitmq-server-windows-3.7.10\rabbitmq_server-3.7.10\sbin>rabbitmqctl add_user nick nick
    Adding user "nick" ...
    
    F:\software\rabbitmq-server-windows-3.7.10\rabbitmq_server-3.7.10\sbin>rabbitmqctl list_users
    Listing users ...
    user    tags
    nick    []
    guest   [administrator]
    
    F:\software\rabbitmq-server-windows-3.7.10\rabbitmq_server-3.7.10\sbin>rabbitmqctl set_user_tags nick administrator
    Setting tags for user "nick" to [administrator] ...
    
    F:\software\rabbitmq-server-windows-3.7.10\rabbitmq_server-3.7.10\sbin>rabbitmqctl list_users
    Listing users ...
    user    tags
    nick    [administrator]
    guest   [administrator]
    
    F:\software\rabbitmq-server-windows-3.7.10\rabbitmq_server-3.7.10\sbin>rabbitmqctl set_permissions -p / nick '.*' '.*' '.*'
    Setting permissions for user "nick" in vhost "/" ...

## Centos7 ##

*  安装erlang

检查依赖  
`yum -y install make gcc gcc-c++ kernel-devel m4 ncurses-devel openssl-devel unixODBC-devel`

下载解压：  
`wget http://erlang.org/download/otp_src_19.3.tar.gz`(当然使用wget,ifconfig命令需要先安装插件yum install wget ,yum install net-tools)

下载完成后解压：`tar -xvzf otp_src_19.3.tar.gz`

编译安装  
`./configure --prefix=/usr/local/erlang --enable-smp-support --enable-threads --enable-sctp --enable-kernel-poll --enable-hipe --with-ssl --without-javac`

`make && make install`  
配置变量：

vim /etc/profile
    export PATH=$PATH:/usr/local/erlang/bin 
    source /etc/profile

验证：`./bin/erl`

*  安装RabbitMQ

cd /usr/local  //切换到计划安装RabbitMQ的目录，我这里放在/usr/local
    wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.1/rabbitmq-server-generic-unix-3.6.1.tar.xz  //下载RabbitMQ安装包
    xz -d rabbitmq-server-generic-unix-3.6.1.tar.xz
    tar -xvf rabbitmq-server-generic-unix-3.6.1.tar

解压，配置环境变量，同上erl语言方式

同样修改/etc/profile 文件， 添加下面的环境变量:
    export PATH=$PATH : /opt/rabbitmq/sbin
    export RABBITMQ HOME=/opt/rabbitmq

启动

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

查看状态：  
`rabbitmqctl status`

配置网页管理，启动插件  
`mkdir /etc/rabbitmq`  
`rabbitmq-plugins enable rabbitmq_management`

开放管理端口和服务器端口

firewall-cmd --permanent --add-port=15672/tcp
    firewall-cmd --permanent --add-port=5672/tcp
    systemctl restart firewalld.service

增加用户

rabbitmqctl add_user nick nick  //添加用户，后面两个参数分别是用户名和密码，我这都用nick了。
    rabbitmqctl set_permissions -p / nick ".*" ".*" ".*"  //添加权限
    rabbitmqctl set_user_tags nick administrator  //修改用户角色

以守护线程启动：  
`rabbitmq-server -detached`

> 在rabbitmq -s erver 命令后面添加一个"-detached" 参数是为了能够让RabbitMQ  
> 服务以守护进程的方式在后台运行，这样就不会因为当前Shell 窗口的关闭而影响服务。

然后就可以远程访问了，然后可直接配置用户权限等信息。  
登录：[http://ip:15672][http_ip_15672] 登录

*  简单管理命令  
    查看状态：`rabbitmqctl status`

F:\software\rabbitmq-server-windows-3.7.10\rabbitmq_server-3.7.10\sbin>rabbitmqctl status
    Status of node rabbit@DESKTOP-2VOJNE8 ...
    [{pid,7068},
     {running_applications,
         [{rabbitmq_management,"RabbitMQ Management Console","3.7.10"},
          {rabbitmq_management_agent,"RabbitMQ Management Agent","3.7.10"},
          {rabbitmq_web_dispatch,"RabbitMQ Web Dispatcher","3.7.10"},
          {rabbit,"RabbitMQ","3.7.10"},
          {mnesia,"MNESIA  CXC 138 12","4.15.5"},
          {amqp_client,"RabbitMQ AMQP Client","3.7.10"},
          {rabbit_common,
              "Modules shared by rabbitmq-server and rabbitmq-erlang-client",
              "3.7.10"},
          {cowboy,"Small, fast, modern HTTP server.","2.6.1"},
          {ranch,"Socket acceptor pool for TCP protocols.","1.7.1"},
          {ssl,"Erlang/OTP SSL application","9.1"},
          {public_key,"Public key infrastructure","1.6.4"},
          {asn1,"The Erlang ASN1 compiler version 5.0.8","5.0.8"},
          {jsx,"a streaming, evented json parsing toolkit","2.9.0"},
          {os_mon,"CPO  CXC 138 46","2.4.7"},
          {cowlib,"Support library for manipulating Web protocols.","2.7.0"},
          {xmerl,"XML parser","1.3.18"},
          {crypto,"CRYPTO","4.4"},
          {inets,"INETS  CXC 138 49","7.0.3"},
          {recon,"Diagnostic tools for production use","2.3.6"},
          {lager,"Erlang logging framework","3.6.5"},
          {goldrush,"Erlang event stream processor","0.1.9"},
          {compiler,"ERTS  CXC 138 10","7.3"},
          {syntax_tools,"Syntax tools","2.1.6"},
          {sasl,"SASL  CXC 138 11","3.3"},
          {stdlib,"ERTS  CXC 138 10","3.7"},
          {kernel,"ERTS  CXC 138 10","6.2"}]},
     {os,{win32,nt}},
     {erlang_version,
         "Erlang/OTP 21 [erts-10.2] [64-bit] [smp:8:8] [ds:8:8:10] [async-threads:64]\n"},
     {memory,
         [{connection_readers,0},
          {connection_writers,0},
          {connection_channels,0},
          {connection_other,24324},
          {queue_procs,0},
          {queue_slave_procs,0},
          {plugins,1395392},
          {other_proc,24664368},
          {metrics,206196},
          {mgmt_db,377832},
          {mnesia,75096},
          {other_ets,2993424},
          {binary,221904},
          {msg_index,29568},
          {code,27675747},
          {atom,1172689},
          {other_system,12906612},
          {allocated_unused,17844560},
          {reserved_unallocated,1048576},
          {strategy,rss},
          {total,[{erlang,71743152},{rss,90636288},{allocated,89587712}]}]},
     {alarms,[]},
     {listeners,
         [{clustering,25672,"::"},
          {amqp,5672,"::"},
          {amqp,5672,"0.0.0.0"},
          {http,15672,"::"},
          {http,15672,"0.0.0.0"}]},
     {vm_memory_calculation_strategy,rss},
     {vm_memory_high_watermark,0.4},
     {vm_memory_limit,6740297318},
     {disk_free_limit,50000000},
     {disk_free,151823843328},
     {file_descriptors,
         [{total_limit,8092},
          {total_used,2},
          {sockets_limit,7280},
          {sockets_used,0}]},
     {processes,[{limit,1048576},{used,397}]},
     {run_queue,1},
     {uptime,1881},
     {kernel,{net_ticktime,60}}]
    
    F:\software\rabbitmq-server-windows-3.7.10\rabbitmq_server-3.7.10\sbin>

如果RabbitMQ 正常启动， 会输出如上所示的信息。当然也可以通过`rabbitmqctlcluster status`命令来查看集群信息,目前只有一个RabbitMQ 服务节点，可以看作单节点的集群

F:\software\rabbitmq-server-windows-3.7.10\rabbitmq_server-3.7.10\sbin>rabbitmqctl cluster_status
 Cluster status of node rabbit@DESKTOP-2VOJNE8 ...
 [{nodes,[{disc,['rabbit@DESKTOP-2VOJNE8']}]},
 {running_nodes,['rabbit@DESKTOP-2VOJNE8']},
 {cluster_name,<<"rabbit@DESKTOP-2VOJNE8">>},
 {partitions,[]},
 {alarms,[{'rabbit@DESKTOP-2VOJNE8',[]}]}]

## Java简单连接RabbitMQ ##

Maven：

<dependency>
 <groupId>com.rabbitmq</groupId>
 <artifactId>amqp-client</artifactId>
 <version>5.5.0</version>
 </dependency>

创建Producer

/** * @Description * @Date 2019/1/16 11:20 * @Since v1.7 * @Autor Nick */
    public class RabbitProducer { 
    
    	private static final String EXCHANGE_NAME = "exchange demo ";
    	private static final String ROUTING_KEY = " routingkey demo";
    	private static final String QUEUE_NAME = "queue demo ";
    	private static final String IP_ADDRESS = "localhost";
    	private static final int PORT = 5672;//RabbitMQ 服务端默认端口号为5672
    
    	public static void main(String[] args) throws IOException, TimeoutException { 
    		ConnectionFactory factory = new ConnectionFactory();
    		factory.setHost(IP_ADDRESS);
    		factory.setPort(PORT);
    		factory.setUsername("guest");
    		factory.setPassword("guest");
    
    		Connection connection = factory.newConnection();	//创建链接
    		Channel channel = connection.createChannel();	//创建信道
    		//创建一个 type = direct 持久化、非自动删除的交换器
    		channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
    		//创建一个持久化、非排他的、非自动删除的队列
    		channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    		//将交换器和队列绑定
    		channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
    
    		//发送一条持久化消息
    		String message = "Hello Nick";
    		channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
    
    		//关闭资源
    		channel.close();
    		connection.close();
    
    	}
    
    }

创建Consumer

/** * @Description * @Date 2019/1/16 11:35 * @Since v1.7 * @Autor Nick */
    public class RabbitConsumer { 
    
    	private static final String QUEUE_NAME = "queue demo ";
    	private static final String IP_ADDRESS = "localhost";
    	private static final int PORT = 5672;//RabbitMQ 服务端默认端口号为5672
    
    	public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { 
    		Address[] addresses = new Address[]{ new Address(IP_ADDRESS, PORT)};
    		ConnectionFactory factory = new ConnectionFactory();
    		factory.setUsername("guest");
    		factory.setPassword("guest");
    
    		Connection connection = factory.newConnection(addresses); // 创建连接
    
    		final Channel channel = connection.createChannel(); //创建信道
    		channel.basicQos(64); // 设置客户端最多接收未被ack 的消息的个数
    		Consumer consumer = new DefaultConsumer(channel) { 
    			@Override
    			public void handleDelivery(String consumerTag,
    									   Envelope envelope,
    									   AMQP.BasicProperties properties,
    									   byte[] body)
    					throws IOException { 
    				System.out.println("recv message: " + new String(body));
    				try { 
    					TimeUnit.SECONDS.sleep(1);
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				}
    				channel.basicAck(envelope.getDeliveryTag(), false);
    			}
    		};
    
    		channel.basicConsume(QUEUE_NAME, consumer);
    		//等待回调函数执行完毕之后， 关闭资源
    		TimeUnit.SECONDS.sleep(5);
    		channel.close();
    		connection.close();
    	}
    }

[http_www.erlang.org_downloads]: http://www.erlang.org/downloads
[http_www.rabbitmq.com_download.html]: http://www.rabbitmq.com/download.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d1NjY2MDU2Mw_size_16_color_FFFFFF_t_70]: /images/20220326/5a64e422d1164f86b2e2e9cfe0129bfc.png
[http_localhost_15672]: http://localhost:15672/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d1NjY2MDU2Mw_size_16_color_FFFFFF_t_70 1]: /images/20220326/1269a862c7e64e57a19521de034db234.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d1NjY2MDU2Mw_size_16_color_FFFFFF_t_70 2]: /images/20220326/0b92ff017c7c4f6eb1fa946b0b9e26dd.png
[http_ip_15672]: http://ip:15672