程序员之消息队列

1、什么是消息队列
MQ (Message Quene) : 翻译为 消息队列 ,经过典型的 ⽣产者 和 消费者 模型,⽣产者不断向消息队列中⽣产消息，消费者不断的从队列中获取消息。由于消息的⽣产和消费都是异步的，⽽且只关⼼消息的发送和接收，没有业务逻辑的侵⼊,轻松的实现系统间解耦。别名为 消息中间件 经过利⽤⾼效可靠的消息传递机制进⾏平台⽆关的数据交流，并基于数据通讯来进⾏分布式系统的集成。

2、为何要使用MQ
1.解耦
如今我有一个系统A，系统A能够产生一个userId,而后，如今有系统B和系统C都须要这个userId去作相关的操做,能够写成以下操做

若是有一天，系统B的负责人告诉系统A的负责人，如今系统B的SystemBNeed2do(String userId)这个接口再也不使用了，让系统A别去调它了。那么就须要从代码的基础上去修改了。这样紧密的耦合关系会致使不少麻烦，若是使用消息中间件就不会出现以上问题。

2.异步
咱们再来看看下面这种状况：系统A仍是直接调用系统B、C、D

假设系统A运算出userId具体的值须要50ms，调用系统B的接口须要300ms，调用系统C的接口须要300ms，调用系统D的接口须要300ms。那么此次请求就须要50+300+300+300=950ms
而且咱们得知，系统A作的是主要的业务，而系统B、C、D是非主要的业务。好比系统A处理的是订单下单，而系统B是订单下单成功了，那发送一条短信告诉具体的用户此订单已成功，而系统C和系统D也是处理一些小事而已。
那么此时，为了提升用户体验和吞吐量，其实能够异步地调用系统B、C、D的接口。因此，咱们能够弄成是这样的：

系统A执行完了之后，将userId写到消息队列中，而后就直接返回了(至于其余的操做，则异步处理)。
原本整个请求须要用950ms(同步)
如今将调用其余系统接口异步化，只须要100ms(异步)

三、削峰/限流
假设如今咱们每月要搞一次大促，大促期间的并发可能会很高的，好比每秒3000个请求。假设咱们如今有两台机器处理请求，而且每台机器只能每次处理1000个请求。

那多出来的1000个请求，可能就把咱们整个系统给搞崩了...因此，有一种办法，咱们能够写到消息队列中：

系统B和系统C根据本身的可以处理的请求数去消息队列中拿数据，这样即使有每秒有8000个请求，那只是把请求放在消息队列中，去拿消息队列的消息由系统本身去控制，这样就不会把整个系统给搞崩。

3、经常使用的消息中间件
当今市⾯上有不少主流的消息中间件，如⽼牌的 ActiveMQ 、 RabbitMQ ，炙⼿可热的Kafka ，阿⾥巴巴⾃主开发 RocketMQ 等。
#1.ActiveMQ
ActiveMQ 是Apache出品，最流⾏的，能⼒强劲的开源消息总线。它是⼀个彻底⽀持JMS规范的的消息中间件。丰富的API,多种集群架构模式让ActiveMQ在业界成为⽼牌的消息中间件,在中⼩型企业颇受欢迎!

2.Kafka

Kafka是LinkedIn开源的分布式发布-订阅消息系统，⽬前归属于Apache顶级项⽬。Kafka主要特色是基于Pull的模式来处理消息消费，追求⾼吞吐量，⼀开始的⽬的就是⽤于⽇志收集和传输。0.8版本开始⽀持复制，不⽀持事务，对消息的重复、丢失、错误没有严格要求，适合产⽣⼤量数据的互联⽹服务的数据收集业务。

3.RocketMQ

RocketMQ是阿⾥开源的消息中间件，它是纯Java开发，具备⾼吞吐量、⾼可⽤性、适合⼤规模分
布式系统应⽤的特色。RocketMQ思路起源于Kafka，但并非Kafka的⼀个Copy，它
对消息的可靠传输及事务性作了优化，⽬前在阿⾥集团被⼴泛应⽤于交易、充值、流计算、消
息推送、⽇志流式处理、binglog分发等场景。

4.RabbitMQ

RabbitMQ⽐Kafka可靠，Kafka更适合IO⾼吞吐的处理，⼀般应⽤在⼤数据⽇志处理或对实时性（少许延迟），可靠性（少许丢数据）要求稍低的场景使⽤，⽐如ELK⽇志收集。

4、RabbitMQ
基于 AMQP 协议，erlang语⾔开发，是部署最⼴泛的开源消息中间件,是最受欢迎的开源消息中间件之⼀。
官⽹ : https://www.rabbitmq.com/
官⽅教程 : https://www.rabbitmq.com/#getstarted

1.AMQP 协议
AMQP（advanced message queuingprotocol）`在2003年时被提出，最先⽤于解决⾦融领不一样平台之间的消息传递交互问题。顾名思义，AMQP是⼀种协议，更准确的说是⼀种binary wire-level protocol（连接协议）。这是其和JMS的本质差异，AMQP不从API层进⾏限定，⽽是直接定义⽹
络交换的数据格式。这使得实现了AMQP的provider自然性就是跨平台的。如下是AMQP协议模型:

2.RabbitMQ 的安装
下载地址：https://www.rabbitmq.com/download.html

注意：处理要下载rabbitmq的按转包外还要下载ellong环境

3.安装步骤
先安装安装Erlang，再安装rabbitmq.exe,直接下一步，傻瓜式安装，安装完成点击开始会出现这里会出现启动、中止、从新安装等

1.点击

2.输入命令：
rabbitmq-plugins enable rabbitmq_management

3.打开浏览器控制台
http://localhost:15672/
默认帐号 guest guest
若是不能访问,
文件夹为隐藏 须要再文件夹选项中把隐藏文件夹打开显示
C:\Users\XYH\AppData\Roaming\RabbitMQ\db 里面的数据删除 再次安装一下Rabbitmq.exe
而后执行
rabbitmq-plugins enable rabbitmq_management
就能够访问到了

5、web管理界⾯介绍
1.简介

connections：⽆论⽣产者仍是消费者，都须要与RabbitMQ建⽴链接后才能够完成消息的⽣产和消费，在这⾥能够查看链接状况
channels：通道，建⽴链接后，会造成通道，消息的投递获取依赖通道。
Exchanges：交换机，⽤来实现消息的路由
Queues：队列，即消息队列，消息存放在队列中，等待消费，消费后被移除队列。
admin:用户

2.Admin⽤户和虚拟主机管理
a.添加用户

上⾯的Tags选项，实际上是指定⽤户的⻆⾊，可选的有如下⼏个：
==》超级管理员(administrator)：可登录管理控制台，可查看全部的信息，而且能够对⽤户，策略(policy)进⾏操做。
==》监控者(monitoring)：可登录管理控制台，同时能够查看rabbitmq节点的相关信息(进程数，内存
使⽤状况，磁盘使⽤状况等)
==》策略制定者(policymaker)：可登录管理控制台, 同时能够对policy进⾏管理。但⽆法查看节点的相关信息(上图红框标识的部分)。
==》普通管理者(management)：仅可登录管理控制台，⽆法看到节点信息，也⽆法对策略进⾏理。
==》其余：⽆法登录管理控制台，一般就是普通的⽣产者和消费者。

b.建立虚拟主机
虚拟主机:为了让各个⽤户能够互不⼲扰的⼯做，RabbitMQ添加了虚拟主机（Virtual Hosts）的概念。其实就是⼀个独⽴的访问路径，不一样⽤户使⽤不一样路径，各⾃有⾃⼰的队列、交换机，互相不会影响。

c、绑定虚拟主机

6、RabbitMQ 的第⼀个程序
a.AMQP协议

b、RabbitMQ⽀持的消息模型

c.引入依赖
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.7.2</version>
</dependency>

第⼀种模型(直连)

在上图的模型中，有如下概念：
P：⽣产者，也就是要发送消息的程序
C：消费者：消息的接受者，会⼀直等待消息到来。
queue：消息队列，图中红⾊部分。相似⼀个邮箱，能够缓存消息；⽣产者向其中投递消息，消费者从其中取出消息。

开发⽣产者
//建立链接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//设置端ip号
connectionFactory.setHost("127.0.0.1");
//设置端口号
connectionFactory.setPort(5672);
//设置用户名和密码
connectionFactory.setUsername("xyh");
connectionFactory.setPassword("123");
//访问的虚拟通道
connectionFactory.setVirtualHost("/vhost");
//链接
Connection connection = connectionFactory.newConnection();
//建立通道
Channel channel = connection.createChannel();
//通道绑定对应的消息队列
//参数1: 是否持久化 参数2:是否独占队列 参
//数3:是否⾃动删除 参数4:其余属性
channel.queueDeclare("hello",false,false,false,null);
//发布消息
channel.basicPublish("","hello",null,"hello".getBytes());
channel.close();
connection.close();

消费者
//建立链接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//设置端ip号
connectionFactory.setHost("127.0.0.1");
//设置端口号
connectionFactory.setPort(5672);
//设置用户名和密码
connectionFactory.setUsername("xyh");
connectionFactory.setPassword("123");
//访问的虚拟通道
connectionFactory.setVirtualHost("/vhost");
//链接
Connection connection = connectionFactory.newConnection();
//建立通道
Channel channel = connection.createChannel();
channel.queueDeclare("hello",false,false,false,null);
//消费消息
channel.basicConsume("hello",true,new DefaultConsumer(channel){br/>@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(new String(body));
}
});

7、springboot实现简单的rabbitmqbr/>1.导入依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2.配置rabbitmq的配置信息
spring.application.name=rabbitmq-springboot
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=dt87
spring.rabbitmq.password=123
spring.rabbitmq.virtual-host=/dt87
3.消息生产者
@Autowired
private RabbitTemplate rabbitTemplate;br/>@Test
public void test(){
//直接获取rabbitmq的模板对象
rabbitTemplate.convertAndSend("hello","hello world");

}br/>四、消息的消费者
@Component
@RabbitListener(queuesToDeclare = @Queue("hello"))
public class Consumer {br/>//消费消息的方法@RabbitHandlerpublic void test(String message){System.out.println(message);}}}