由于换 Mac 了,所以一些环境就直接在 Mac 搞,但是像安装 RabbitMQ 这些又会把自己电脑系统给搞的太乱,所以能在 Docker 里面安装就安装在 Docker,这次 RabbitMQ 我也直接在 Docker 里安装。
启动 Docker for Mac
当然你也可以在自己的 Linux 服务器或者虚拟机里启动安装 RabbitMQ 。
Docker 安装的话很简单,因为 RabbitMQ 官方已经提供了自己的 Docker 容器,只需要一行命令:
docker run -d -p 15672:15672 -p 5672:5672 -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin --name rabbitmq rabbitmq:3-management
该镜像拥有一个基于 web 的控制台和 Http API。Http API 可以在地址看到如何使用:http://localhost:15672/api/
讲解下上面命令行:
容器启动成功后,可以在浏览器输入地址:http://localhost:15672/ 访问控制台
登陆后:
简单描述下上图中中控制台的列表的作用:
SpringBoot 框架中已经内置了对 RabbitMQ 的支持,如果你看过官方文档的话,就可以看到的,我们需要把依赖 spring-boot-starter-amqp 引入就行。
1、 pom.xml 引入依赖后如下:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.zhisheng</groupId>
<artifactId>rabbitmq</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>rabbitmq</name>
<description>Demo project for Spring Boot RabbitMQ</description>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.5.9.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
2、application.properties 配置修改如下:
spring.rabbitmq.addresses=localhost:5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
3、消息发送类 RabbitMQClient.java
package com.zhisheng.rabbitmq.client;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* Created by zhisheng_tian on 2018/1/23
*/
@Component
public class RabbitMQClient {
@Autowired
private RabbitTemplate rabbitTemplate;
public void send(String message) {
rabbitTemplate.convertAndSend("zhisheng", message);
}
}
就这样,发送消息代码就实现了。
这里关键的代码为 rabbitTemplate.convertAndSend() 方法,zhisheng
这个是路由规则(routingKey),它的值表明将消息发送到指定的队列 zhisheng
中去,这里跟了下源码,发现 convertAndSend() 方法最后调用的方法其实是一个 doSend() 方法。
4、消息接收类
package com.zhisheng.rabbitmq.server;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* Created by zhisheng_tian on 2018/1/23
*/
@Component
public class RabbitMQServer {
@RabbitListener(queues = "zhisheng")
public void receive(String message) {
System.out.println("收到的 message 是:" + message);
}
}
你看,这里就有个 RabbitListener
一直在监听着队列 zhisheng
。
当然这个队列是必须要我们自己在应用程序中创建好,它不会像 Kafka 一样,Kafka 它会在用到队列的时候动态的创建,不需要我们提前创建好。
那么在 RabbitMQ 中该如何创建队列呢?
如上图所示:这样我们就创建好了一个 zhisheng
的队列,当程序开始运行时,消息接收类会持续监听队列 zhisheng
中即将到来的消息。
5、运行项目
需要在启动类中注入发送消息的类,并且提供 init 方法,在 init 方法中调用发送消息类的 send() 方法
@PostConstruct
public void init() {
rabbitMQClient.send("发送消息----zhisheng-----");
}
需要注意的是:init() 方法带有 @PostConstruct 注解,被 @PostConstruct 修饰的方法会在构造函数之后执行。
启动项目就可以发现控制台已经接收到消息了。
6、单线程测试性能
看到上面图片中注释掉的代码没?那就是用来测试消息发送的性能的,我发送 10000 条消息看看总共耗时多少。
10000 条消息发送耗时:215ms。 这是在单线程下,下次可以和其他的 MQ 测试对比下,并且也可以在多线程的环境下测试性能。
同时从控制台可以看到发送的速率:
7、多线程测试性能
开了10 个线程,每个线程发送 10000 条消息。
init 方法代码如下:
@PostConstruct
public void init() {
StopWatch stopWatch = new StopWatch();
stopWatch.start();
int threads = 10;
ExecutorService executorService = Executors.newFixedThreadPool(threads);
final CountDownLatch start = new CountDownLatch(1);
final CountDownLatch end = new CountDownLatch(threads);
for (int i = 0; i < threads; i++) {
executorService.execute(() -> {
try {
start.await();
for (int i1 = 0; i1 < 10000; i1++) {
rabbitMQClient.send("发送消息----zhisheng-----");
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
end.countDown();
}
});
}
start.countDown();
try {
end.await();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
executorService.shutdown();
}
stopWatch.stop();
System.out.println("发送消息耗时:" + stopWatch.getTotalTimeMillis());
}
耗时:4063ms
控制台显示如下图:
8、注意
这里测试发送的消息直接是 String 类型的,你也可以测试下 Bean 类,这需要注意需要序列化。