马桶🚽Java 上厕所就能看完的小知识! 欢迎关注、点赞 持续更新!
什么是MQ?
MQ(Message Quene) : 翻译为 消息队列,运用生产者和消费者模型,生产者不断向消息队列中生产消息,消费者不断的从队列中获取(消费)消息。因为消息的生产和消费都是异步的,而且只关心消息的发送和接收,没有业务逻辑的侵入,轻松的实现系统间解耦。
MQ种类介绍
# 1.ActiveMQ
ActiveMQ 是Apache出品,最流行的,能力强劲的开源消息总线。它是一个完全支持JMS规范的的消息中间件。丰富的API,多种集群架构模式让ActiveMQ在业界成为老牌的消息中间件,在中小型企业颇受欢迎!
# 2.Kafka
Kafka是LinkedIn开源的分布式发布-订阅消息系统,目前归属于Apache顶级项目。Kafka主要特点是基于Pull的模式来处理消息消费,追求高吞吐量,一开始的目的就是用于日志收集和传输。0.8版本开始支持复制,不支持事务,对消息的重复、丢失、错误没有严格要求,适合产生大量数据的互联网服务的数据收集业务。
# 3.RocketMQ
RocketMQ是阿里开源的消息中间件,它是纯Java开发,具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。RocketMQ思路起源于Kafka,但并不是复制kafka,它对消息的可靠传输及事务性做了优化,目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景。
# 4.RabbitMQ
RabbitMQ是使用Erlang语言开发的开源消息队列系统,基于AMQP协议来实现。AMQP的主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。AMQP协议更多用在企业系统内对数据一致性、**稳定性和可靠性**要求很高的场景,对性能和吞吐量的要求还在其次。
RabbitMQ比Kafka可靠,Kafka更适合IO高吞吐的处理,一般应用在大数据日志处理或对实时性(少量延迟),可靠性(少量丢数据)要求稍低的场景使用,比如ELK日志收集。
RabbitMQ引言
什么是rabbitMQ?
RabbitMQ是实现了高级消息队列协议(AMQP)的开源消息代理软件(亦称面向消息的中间件)。RabbitMQ服务器是用Erlang语言编写的,而群集和故障转移是构建在开放电信平台框架上的。
官方教程
: https://www.rabbitmq.com/#getstarted
RabbitMQ安装
RabbitMQ安装首先要安装Erlang
安装地址:https://www.rabbitmq.com/download.html
具体安装调试请参考。
RabbitMQ界面初识
主界面
登录管理界面 username: guest password: guest
Connections:连接。无论是生产者和消费者都需要在与RabbitMQ建立连接的情况下完成消息的生产与消费,这里可以查看相应的消费情况。
Channels:通道。建立连接后会形成相应的通道,消息的投递与获取都依赖于通道。
Exchanges: 交换机。用于实现消息的路由功能。
Queues:消息队列。消息存在在消息队列中等待消费。
用户
tags为指定用户
-
超级管理员(administrator)
可登陆管理控制台,可查看所有的信息,并且可以对用户,策略(policy)进行操作。
-
监控者(monitoring)
可登陆管理控制台,同时可以查看rabbitmq节点的相关信息(进程数,内存使用情况,磁盘使用情况等)
-
策略制定者(policymaker)
可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。
-
普通管理者(management)
仅可登陆管理控制台,无法看到节点信息,也无法对策略进行管理。
-
其他
无法登陆管理控制台,通常就是普通的生产者和消费者。
虚拟机
为了让各个用户互相不干预工作,RabbitMQ 添加了虚拟机概念,就是一个独立的访问路径,不同用户使用不同路径,各自有自己的队列、交换机,互相不会影响。
进入相关新创建的虚拟机。
为创建好的虚拟机绑定对应用户。
RabbitMQ 消息模型
引入依赖
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.7.2</version>
</dependency>
创建工具
public class RabbitMqConnectionUtil {
private static final ConnectionFactory CONNECTION_FACTORY;
private static Connection connection;
static {
//通过连接工厂并设置相应的ip 端口 用户名密码 及绑定的虚拟机创建连接
CONNECTION_FACTORY = new ConnectionFactory();
CONNECTION_FACTORY.setHost("127.0.0.1");
CONNECTION_FACTORY.setPort(5672);
CONNECTION_FACTORY.setUsername("test");
CONNECTION_FACTORY.setPassword("test");
CONNECTION_FACTORY.setVirtualHost("/test");
}
public static Connection createConnection() {
try {
connection = CONNECTION_FACTORY.newConnection();
} catch (IOException | TimeoutException e) {
e.printStackTrace();
}
return connection;
}
public static void close(Channel channel, Connection connection) throws IOException, TimeoutException {
channel.close();
connection.close();
}
}
工作队列
让多个消费者绑定到一个队列,共同消费队列中的消息。队列中的消息一旦消费,就会消失,因此任务是不会被重复执行的。
在模型中,有以下概念:
- P:生产者,也就是要发送消息的程序
- C:消费者:消息的接受者,会一直等待消息到来。
- queue:消息队列,图中红色部分。类似一个邮箱,可以缓存消息;生产者向其中投递消息,消费者从其中取出消息。
生产者如下:
public class Producer {
public static void main(String[] args) throws IOException, TimeoutException {
// 获取连接
Connection connection = RabbitMqConnectionUtil.createConnection();
// 获取通道
Channel channel = connection.createChannel();
/**
创建队列
参数如下:
queue - 队列的名称
durable - 宣布一个持久的队列(队列不受服务器重启,数据会消失)
exclusive - 宣布独占队列(限于此连接)
autoDelete - 一个自动删除队列(服务器将删除它在使用时不再)
参数 - 其它性质(结构参数)队列
*/
channel.queueDeclare("work", true, false, false, null);
for (int i = 0; i < 10; i++) {
channel.basicPublish("", "work", null, ("work queue message has been sent --"+i).getBytes());
}
RabbitMqConnectionUtil.close(channel,connection);
}
}
消费者获取消息:
再创建一个Consumer02 代码相同。
public class Consumer01 {
public static void main(String[] args) throws IOException {
Connection connection = RabbitMqConnectionUtil.createConnection();
Channel channel = connection.createChannel();
channel.queueDeclare("work", true, false, false, null);
// 获取对应消息 autoAck 自动确认消息 在这里使用的默认消息消费者
channel.basicConsume("work",true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(new String(body));
}
});
}
}
先运行消费者等待消费队列中数据,在打开生产者生产消息结果如下:
这种形式是公平分配形式,在程序运行时将对应的未确认消息分配给对应的消费者。然后再进行消费。(自动确认消息)
接下来我们切换为手动确认消息,实现类似于能者多劳的分配模式。
未完待续。。。