一,引言
重要的事情说三遍:在学一门新技术之前,一定要先看官方文档,看官方文档,看官方文档。https://www.rabbitmq.com
二,rabbitmq的安装
在服务器或者虚拟机中安装东西不可能还使用原生方式安装吧,因此这里推荐使用docker安装,这里我在自己的服务器中安装一下
下载镜像
docker pull rabbitmq-manager
后台方式运行:
docker run -d -p 5672:5672 -p 15672:15672 --name rabbitmq rabbitmq:3.7.3-management
OK了,这样rabbitmq3.7.3-manager就安装好了,一定要把manager加上,不然图形化界面出现不了
在服务器上开放5672和15672端口
可以查看一下正在运行的容器
docker ps
测试连接,可以发现是没问题的。(如果有,那就关闭一下防火墙或者看看端口是否开放)
curl 0.0.0.0:5672
浏览器测试:服务器ip:15672
默认的账号密码都是guest,输入到达以下界面,那么rabbitmq就成功安装了
三,什么是MQ
MQ
(Message Quene) : 翻译为 消息队列
,通过典型的 生产者
和消费者
模型,生产者不断向消息队列中生产消息,消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的,而且只关心消息的发送和接收,没有业务逻辑的侵入,轻松的实现系统间解耦。别名为 消息中间件
通过利用高效可靠的消息传递机制进行平台无关的数据交流,并基于数据通信来进行分布式系统的集成。
当然MQ分别有ActiveMQ
、RabbitMQ
,炙手可热的Kafka
,阿里巴巴自主开发RocketMQ
等。这里主要说说流行的rabbitMq和kafka之间的区别
kafka
Kafka是LinkedIn开源的分布式发布-订阅消息系统,目前归属于Apache顶级项目。
Kafka主要特点是基于Pull的模式来处理消息消费,
追求高吞吐量,一开始的目的就是用于日志收集和传输。
0.8版本开始支持复制,不支持事务,对消息的重复、丢失、错误没有严格要求,
适合产生大量数据的互联网服务的数据收集业务
RabbitMq
RabbitMQ是使用Erlang语言开发的开源消息队列系统,基于AMQP协议来实现。
AMQP的主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。
AMQP协议更多用在企业系统内对数据一致性、稳定性和可靠性要求很高的场景,
对性能和吞吐量的要求还在其次。
Rabbit优劣势
RabbitMQ比Kafka可靠,Kafka更适合IO高吞吐的处理,
一般应用在大数据日志处理或对实时性(少量延迟),
可靠性(少量丢数据)要求稍低的场景使用,比如ELK日志收集。
四,rabbitmq六种基本类型
可以参照官网:rabbitmq.com/#getstarted
1,hello类型
2,工作队列
Work queues
,也被称为(Task queues
),任务模型。当消息处理比较耗时的时候,可能生产消息的速度会远远大于消息的消费速度。长此以往,消息就会堆积越来越多,无法及时处理。此时就可以使用work 模型:让多个消费者绑定到一个队列,共同消费队列中的消息。队列中的消息一旦消费,就会消失,因此任务是不会被重复执行的。
3,发布/订阅
- 可以有多个消费者
- 每个消费者有自己的queue(队列)
- 每个队列都要绑定到Exchange(交换机)
- 生产者发送的消息,只能发送到交换机,交换机来决定要发给哪个队列,生产者无法决定
4,路由
- 队列与交换机的绑定,不能是任意绑定了,而是要指定一个
RoutingKey
(路由key) - 消息的发送方在 向 Exchange发送消息时,也必须指定消息的
RoutingKey
。 - Exchange不再把消息交给每一个绑定的队列,而是根据消息的
Routing Key
进行判断,只有队列的Routingkey
与消息的Routing key
完全一致,才会接收到消息
5,话题
与上之相比几乎相同,只不过该消息队列可以使用通配符
五,rabbitmq中虚拟机的创建以及消息用户的创建
在exchanges中添加虚拟机
然后添加用户,并将该用户加入到那个虚拟机下面
六,springboot整合rabbitMq
新建springboot项目,这里不多说
1,需要的依赖
<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>
<!--rabbitmq-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!--rabbitmq-test-->
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.13</version>
<scope>test</scope>
</dependency>
</dependencies>
2,application.yml
spring:
application:
name: rabbitmq #在这里可能用不上,主要是微服务中用的,给这个服务取个名字
rabbitmq:
host: 119.23.248.211 #服务器ip,或者虚拟机ip
port: 5672 #开启的端口号
username: root #用户的用户名,需要在虚拟机中进行添加设置
password: 123456 #同上,密码
virtual-host: /zhs #虚拟机名称
3,将测试类这个主类作为我们的生产者,用来模拟测试
package com.zheng;
import org.junit.jupiter.api.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
@SpringBootTest(classes = DemoApplication.class)
@RunWith(SpringRunner.class)
class DemoApplicationTests {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* @param: hello:消息队列的名称
*/
@Test
void contextLoads() {
rabbitTemplate.convertAndSend("hello","hello wrold");
}
//work 工作模型
@Test
public void work(){
for (int i = 0; i < 10; i++) {
rabbitTemplate.convertAndSend("work","hello work!");
}
}
//Fanout 广播模型
@Test
public void testFanout() throws InterruptedException {
rabbitTemplate.convertAndSend("logs","","这是日志广播");
}
//Router 路由模型
@Test
public void testDirect(){
rabbitTemplate.convertAndSend("directs","error","error 的日志信息");
}
//Topic 订阅模型
@Test
public void testTopic(){
rabbitTemplate.convertAndSend("topics","user.save.findAll","user.save.findAll 的消息");
}
}
4,消费者
HelloCustomer(hello类型)
package com.zheng.config;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* @param: queuesToDeclare:对队列做出的声明
* durable:是否做持久化
* @author zhenghuisheng
*/
@Component
@RabbitListener(queuesToDeclare = @Queue(value = "hello",durable = "true",autoDelete = "false"))
public class HelloCustomer {
@RabbitHandler //当从队列中取出消息的时候的回调方法
public void toReceive(String message){
System.out.println("message" + message);
}
}
WorkCustomer(工作队列)
package com.zheng.config;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* 使用轮询的方式进行,即一个服务提供者两个消费者
*/
@Component
public class WorkCustomer {
@RabbitListener(queuesToDeclare = @Queue("work"))
public void receive1(String message){
System.out.println("work message1 = " + message);
}
@RabbitListener(queuesToDeclare = @Queue("work"))
public void receive2(String message){
System.out.println("work message2 = " + message);
}
}
FanoutCustomer(发布订阅)
package com.zheng.config;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* @author zhenghuisheng
* @date 2021/2/23 19:00
*/
@Component
public class FanoutCustomer {
@RabbitListener(bindings = @QueueBinding(
value = @Queue,
exchange = @Exchange(name="logs",type = "fanout")
))
public void receive1(String message){
System.out.println("message1 = " + message);
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue, //创建临时队列
exchange = @Exchange(name="logs",type = "fanout") //绑定交换机类型
))
public void receive2(String message){
System.out.println("message2 = " + message);
}
}
RouterCustomer(路由)
package com.zheng.config;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* @author zhenghuisheng
* @date 2021/2/23 19:00
*/
@Component
public class RouterCustomer {
/**
* @param message
* exchange:交换机
*/
@RabbitListener(bindings = @QueueBinding(
value = @Queue,
exchange = @Exchange(name="logs",type = "fanout")
))
public void receive1(String message){
System.out.println("message1 = " + message);
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue, //创建临时队列
exchange = @Exchange(name="logs",type = "fanout") //绑定交换机类型
))
public void receive2(String message){
System.out.println("message2 = " + message);
}
}
TopicCustomer(话题)
package com.zheng.config;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* @author zhenghuisheng
* @date 2021/2/23 19:00
*/
@Component
public class TopicCustomer {
@RabbitListener(bindings = {
@QueueBinding(
value = @Queue,
key = {
"user.*"},
exchange = @Exchange(type = "topic",name = "topics")
)
})
public void receive1(String message){
System.out.println("message1 = " + message);
}
@RabbitListener(bindings = {
@QueueBinding(
value = @Queue,
key = {
"user.#"},
exchange = @Exchange(type = "topic",name = "topics")
)
})
public void receive2(String message){
System.out.println("message2 = " + message);
}
}
七,测试
这里值测试两个,hello类型和工作队列。在进行 @Test 之后之后,刷新可以发现如下,当然控制台也有消费者如何接收信息的打印。可以发现两个队列已经加载进来了
如需代码可以访问:https://gitee.com/zhenghuisheng/rabbitmq
八,应用场景
主要是三种:异步处理,应用解耦,流量削峰。