RabbitMQ研究安装与入门

Posted 2022-12-02 wu6660563

简介

MQ

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

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

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

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

作用

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

安装

windows

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

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

• 安装RabbitMQ
  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，出现如下

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

安装成功

添加用户并授权

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 登录

• 简单管理命令
  查看状态：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();