一文带你了解RabbitMQ

1. MQ的介绍

MQ:（message queue）消息队列， 应用程序与应用程序之间的
连接
当程序调用另一个程序时，而不需要另一个程序及时返回结果，而
且另一个程序耗时比较长。（什么样的程序适用MQ）

2. MQ的作用

3. MQ需要遵循哪些协议。

1. JMS: java自带的一种协议
2. AMQP: 新建的一种协议

4.常见的MQ产品

RabbitMQ:
RocketMQ:
ActiveMQ:
KAFKA:

5. RabbitMQ消息队列

2007年，Rabbit 技术公司基于 AMQP 标准开发的 RabbitMQ 1.0
发布。RabbitMQ 采用 Erlang 语言开发。Erlang 语言由 Ericson 设
计，专门为开发高并发和分布式系统的一种语言，在电信（电商）
领域使用广泛。

6. 使用RabbitMQ

rabbitmq支持六种队列模式。 使用最多就是前5中，RPC
6.1 简单模式
1. 创建一个父工程，并引入公共依赖
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.6.0</version>
</dependency>
2. 创建生产者工程
package com.ykq.simple;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class SimpleProduct {
public static void main(String[] args)
throws Exception {
//0. 创建连接工程对象
ConnectionFactory factory=new
ConnectionFactory();
//1.对连接工程对象进行配置
factory.setVirtualHost("/aaa");//默认为
/
factory.setUsername("ykq");//默认guest
factory.setPassword("ykq");//默认guest
factory.setHost("192.168.213.138"); //默
认localhost
factory.setPort(5672); //默认就是5672
//2.创建连接对象---->
Connection
connection=factory.newConnection();
//3.获取信道对象
Channel
channel=connection.createChannel();
//4.因为我们使用的为简单模式 没有交换机（默认你
的交换机）
/**
* (String queue, boolean durable,
boolean exclusive, boolean autoDelete,
Map<String, Object> arguments)
* queue: 队列的名称:
* durable: 该队列是否持久化。
* exclusive: 队列的扩展。 该队列是否独占
连接资源
* autoDelete: 是否自动删除
* arg: 该队列的参数 比如: 设置队列中消息存放
的最大个数，该对象的过期时间
*/
channel.queueDeclare("simple_queue",false,false
,false,null);
//5.发消息
/**
* (String exchange, String routingKey,
BasicProperties props, byte[] body)
* exchange: 交换机的名称 因为我们的是简单模
式 ”“ 采用默认交换机
* routekey:路由key 如果是简单模式你的
routkey的值必须与队列的名称相同
* props:属性参数。 比如：可以对消息设置过期
时间
* body: 消息的内容
*/
String msg="hello world
~~~~~~~~~~~~~~~~~~";
channel.basicPublish("","simple_queue",null,msg
.getBytes());
//关闭资源
channel.close();
connection.close();
}
}
3. 创建消费者工程
package com.ykq.simple;
import com.rabbitmq.client.*;
import java.io.IOException;
public class SimpleConsumer {
public static void main(String[] args) throws
Exception {
//0. 创建连接工程对象
ConnectionFactory factory=new
ConnectionFactory();
//1.对连接工程对象进行配置
factory.setVirtualHost("/aaa");//默认为 /
factory.setUsername("ykq");//默认guest
factory.setPassword("ykq");//默认guest
factory.setHost("192.168.213.138"); //默认
localhost
factory.setPort(5672); //默认就是5672
//2.创建连接对象---->
Connection
connection=factory.newConnection();
//3.获取信道对象
Channel channel=connection.createChannel();
//4.因为我们使用的为简单模式 没有交换机（默认你的交
换机）
/**
* (String queue, boolean durable, boolean
exclusive, boolean autoDelete, Map<String, Object>
arguments)
* queue: 队列的名称:
* durable: 该队列是否持久化。
* exclusive: 队列的扩展。 该队列是否独占 连接
资源
* autoDelete: 是否自动删除
* arg: 该队列的参数 比如: 设置队列中消息存放的
最大个数，该对象的过期时间
*/
channel.queueDeclare("simple_queue",false,false,fa
lse,null);
//创建队列，如果存放可以不用创建。
//获取消息
/**
* (String queue, boolean autoAck, Consumer
callback)
* autoAck: 是否自动确认
* callback:当你收到消息后，回调的函数
*/
Consumer consumer=new
DefaultConsumer(channel){
//一旦队列中有效性，就会触发该方法
/**
*
* @param consumerTag 标识 消费者
* @param envelope 获取一些配置内容
* @param properties 属性
* @param body 消息的内容
* @throws IOException
*/
public void handleDelivery(String
consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body)
throws IOException {
System.out.println("00000"+consumerTag);
System.out.println("00000"+envelope);
System.out.println("00000"+properties);
System.out.println("00000"+new
String(body));
}
};
channel.basicConsume("simple_queue",true,consumer)
;
}
}
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
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142