RabbitMQ是什么
MQ全称为Message Queue, 即消息队列。MQ是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。RabbitMQ则是一个在AMQP基础上完整的，可复用的企业消息系统。
RabbitMQ安装与启动
安装命令
终端中输入以下命令
|
1
2
|
brew update
brew install rabbitmq
|
配置RabbitMQ
安装完成后需要将/usr/local/sbin添加到环境变量中
打开.bash_profile文件
|
1
|
vim ~/.bash_profile
|
将 PATH=/usr/local/sbin:$PATH 加入文件
通过:wq命令保持文件并退出
|
1
|
source ~/.bash_profile
|
启动RabbitMQ
终端中输入以下命令
|
1
|
rabbitmq-server
|
登录Web管理界面
浏览器地址栏输入 http://localhost:15672/，账户密码默认 guest
Hello RabbitMQ
通过一个实例来体验RabbitMQ的使用方法。
添加依赖包
在pom.xml文件中新增dependency
|
1
2
3
4
5
|
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>3.6.5</version>
</dependency>
|
Send实现
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
public class Send {
private final static String QUEUE_NAME = "hello";
public static void main(String[] argv)
throws java.io.IOException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = null;
Channel channel = null;
try {
connection = factory.newConnection();
channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
String message = "Hello Rabbit!";
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
} catch (TimeoutException e) {
e.printStackTrace();
} finally {
try {
channel.close();
} catch (TimeoutException e) {
e.printStackTrace();
}
connection.close();
}
}
}
|
Recv实现
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
public class Recv {
private final static String QUEUE_NAME = "hello";
public static void main(String[] argv)
throws java.io.IOException,
java.lang.InterruptedException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = null;
try {
connection = factory.newConnection();
} catch (TimeoutException e) {
e.printStackTrace();
}
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
Consumer consumer = new DefaultConsumer(channel) {
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" [x] Received '" + message + "'");
}
};
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
|
测试结果
多次运行Send类 [x] Sent 'Hello Rabbit!'可以通过界面查看在消息队列中暂存的消息。
此时运行Recv类可以读取消息队列中的消息
|
1
2
3
4
|
[*] Waiting for messages. To exit press CTRL+C
[x] Received 'Hello Rabbit!'
[x] Received 'Hello Rabbit!'
[x] Received 'Hello Rabbit!'
|
RabbitMQ基本概念
交换器Exchange
- direct交换器：如果路由键匹配的话，消息就被投递到对应的队列。
- fanout交换器：将收到的消息广播到绑定的队列上。
- topic交换器：根据规则，使得来自不同源头的消息能够到达同一队列。