RabbitMQ
RabbitMQ官网直通车 —> ✈✈✈✈✈✈
1、前言
最近工作繁忙,好久没有更新博文了。
对于互联网饱和的今天,如何做到不同系统之间传递信息与通信?
在实际项目中,多个端例如:ios、android、pc、小程序采用从RabbitMQ上获取实时包消息,然后根据此实时包消息来做响应处理。
随着互联网技术的发展,系统之间的耦合度越来越高。为了实现系统间的解耦,消息中间件应运而生。其中作为一款优秀的开源消息中间件
,RabbitMQ凭借其易用、高可靠、多协议支持等特性,被广泛应用于异步处理、任务队列等领域,成为实际项目的首选。
但是对于许多人来说,RabbitMQ还是比较陌生的。概念多、不知如何上手使用,这成为很多人学习和使用RabbitMQ的障碍。接下来,给大家介绍关于RabbitMQ的内容。
2、RabbitMQ简介
2.1、什么是RabbitMQ
MQ全称为Message Queue, 消息队列(MQ)是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息(针对应用程序的数据)来通信,而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信,而不是通过直接调用彼此来通信,直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。其中较为成熟的MQ产品有IBM WEBSPHERE MQ等等。
RabbitMQ是采用Erlang语言实现的开源消息队列系统
,是当前最主流的消息中间件之一。
通俗来讲的话,RabbitMQ就像是一个邮局,负责接收和转发信件。
- 生产者就像邮寄的客户,把信件(消息)投递到RabbitMQ邮局。
- 消费者就像收信人,从RabbitMQ邮局取走并处理信件。
- RabbitMQ保证每封信件(消息)都会安全可靠地到达收信人(消费者),它不会丢失或者重复发送信件。
- RabbitMQ可以给不同的收信人(消费者)分发同一封信件的复印件,这样多个收信人可以并行处理信件。
- 如果RabbitMQ邮局堆积太多信件来不及处理,它会把信件排队,先进先出依次发送。
- RabbitMQ还可以给特殊的信件设置优先级,让重要的信件优先处理。
- RabbitMQ可以根据收信人的处理能力分发信件,不会让一个收信人处理过多信件。
所以简单来说,RabbitMQ就是一个智能的邮局系统,保证每个消息都被安全可靠地发送和处理。它非常适合用于不同系统之间传递消息与通信。
2.2、RabbitMQ的特点
- 支持多种消息协议,包括STOMP、MQTT等。
- 支持消息队列,可以缓存消息。
- 支持消息持久化,防止消息丢失。
- 支持复杂的路由规则,可以实现发布-订阅、负载均衡等功能。
- 高可靠性,支持集群模式。
- 管理界面友好,使用方便。
2.3、RabbitMQ的工作原理
- 生产者(Publisher)生成消息并发布到RabbitMQ服务器。
- 消费者(Consumer)从RabbitMQ服务器拉取消息并进行处理。
- RabbitMQ充当了一个消息代理(Message Broker)的角色,负责接收、存储和转发消息。
2.4、RabbitMQ中几个重要的概念
- Producer(生产者):发送消息的应用程序是Producer。Producer将消息发布到RabbitMQ中去。
- Consumer(消费者):接收消息的应用程序是Consumer。Consumer从RabbitMQ中获取消息进行处理。
- ConnectionFactory(工厂类):它是是RabbitMQ Java客户端用于创建连接的工厂类。它封装了用于创建连接的所有参数配置。
- Exchange(交换机):交换机用于接收生产者发送的消息,并根据路由键将消息路由到指定的队列中。
- Queue(消息队列):消息队列是RabbitMQ内部存储消息的缓冲区。Producer发送的消息会先存储在Queue中,Consumer从Queue中获取消息进行处理。
- Channel(通道):进行消息读写的通道,它是建立在Connection上的一个虚拟连接。为了实现并发,同时方便业务和异常隔离,最佳实践是基于单个Connection建立多个Channel,而不是直接基于Connection操作。
- Routing Key(路由键):生产者将消息发布到Exchange时,会指定一个Routing Key。Exchange根据这个Routing Key决定将消息路由到哪个队列。
- Binding(绑定):绑定是Exchange和Queue之间的关联关系。绑定中可以包含routing key。
- Virtual host(虚拟主机):RabbitMQ可以创建多个虚拟主机,用于进行权限管理和进行逻辑隔离。
- Message Acknowledgment (消息确认):消费者可以启用消息确认机制,在收到消息并处理后发送确认回执给RabbitMQ,RabbitMQ才会将消息从Queue中移除。
3、在Android Studio中集成RabbitMQ
3.1、在Manifest中添加权限:
<uses-permission android:name="android.permission.INTERNET" />
3.2、在build.gradle(:app)下添加依赖:
implementation 'com.rabbitmq:amqp-client:5.19.0'
耐心等待as同步完成后,就可以使用RabbitMQ的相关api了。
4、建立连接
4.1、创建ConnectionFactory对象
这个对象包含了创建连接需要的配置,比如RabbitMQ主机地址,端口,虚拟主机,用户名密码等。
ConnectionFactory factory = new ConnectionFactory();
// 连接配置
// factory.setHost(Config.MQ_ADDRESS); // 服务器ip
// factory.setPort(Integer.parseInt(Config.MQ_PORT)); // 端口
// factory.setUsername(Config.MQ_USERNAME); // 用户名
// factory.setPassword(Config.MQ_PASSWORD); // 密码
// factory.setVirtualHost(Config.MQ_VIRTUALHOST);
factory.setUri(url);
4.2、使用ConnectionFactory创建连接
调用ConnectionFactory的createConnection()
方法可以创建一个连接对象。
Connection connection = factory.newConnection();
4.3、创建Channel
在连接上创建一个通道,用于进行消息发布或者消费。
Channel channel = connection.createChannel();
4.4、声明队列、交换机等
使用channel进行队列、交换机等的声明。
4.5、生产或消费消息
通过channel发送或接收消息。
4.6、关闭连接
使用完成后关闭连接和channel。
channel.close();
connection.close();
5、发送消息
5.1、创建连接和通道
使用ConnectionFactory创建连接,然后在连接上创建通道。
5.2、声明队列
如果队列不存在,需要提前声明队列。
5.3、准备消息内容
定义要发送的消息内容体,可以是字节数组或字符串等。
5.4、发布消息
使用通道对象调用basicPublish方法,它需要exchange名称,routing key和消息内容。
5.5、关闭资源
发送完成后可以关闭通道和连接。
5.6、案例
// 构建消息内容
String message = "Hello World!";
// 发布消息到队列
channel.basicPublish(EXCHANGE_NAME, "routingkey", null, message.getBytes());
// 关闭资源
channel.close();
connection.close();
6、接收消息
6.1、创建连接和通道
使用ConnectionFactory创建连接,然后在连接上创建通道。
6.2、声明队列
如果队列不存在则需要先声明队列。
6.3、定义消费者
实现Consumer接口,定义消息处理逻辑。
6.4、监听队列
使用通道对象调用basicConsume方法,监听指定队列。
6.5、接收消息
RabbitMQ将向Consumer递送消息,在handleDelivery方法中可以获取消息并处理。
6.6、确认消息
处理完成后,调用channel的basicAck方法手动确认消息。
6.7、关闭资源
最后需要关闭通道和连接。
6.8、案例
channel.basicConsume(QUEUE_NAME, true, consumer);
public class MyConsumer implements Consumer {
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
// 获取消息并处理
String message = new String(body, "UTF-8");
System.out.println("Received message: " + message);
// 确认消息
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
7、确认机制
RabbitMQ中的确认机制主要分为两种:
7.1、发布确认(Publisher Confirms)
这是一种单向确认机制,允许生产者知道消息是否被 RabbitMQ 接收。
在 Channel 上启用确认模式后,所有的消息都会被指派一个唯一的 ID(从 1 开始),一旦消息被投递到所有匹配的队列后,RabbitMQ 就会发送一个确认给生产者(包含消息的唯一 ID),这就使得生产者知道消息已经被处理了。
如果消息和队列是可持久化的,那么确认机制本身对消息持久化也是必要的。
在 Java 客户端中,可以通过 Channel 的 confirmSelect
方法将 Channel 设置为确认模式:
channel.confirmSelect();
之后可以添加监听器监听确认和未确认的消息:
channel.addConfirmListener(new ConfirmListener(){
// ...
});
7.2、消费者确认(Consumer Acknowledgments)
这是一种双向确认机制,不仅可以告诉生产者消息已送达,也可以告诉 RabbitMQ 消息已经被消费者接收并处理完毕。
开启确认模式后,消息一旦被消费者接收,就会从 RabbitMQ 的消息缓冲区中移除。如果消费者在处理消息时发生故障或者异常退出,未处理完毕的消息就会被 RabbitMQ 重新派发给其他消费者,以此来确保消息不会丢失。
通过正确使用确认机制,既可以提高 RabbitMQ 的性能和消息处理能力,也可以确保业务流程的完整性。所以在实际使用中,确认机制是非常重要的。
在 Java 客户端中,可以在 basicConsume 时设置autoAck=false
,之后手动调用 basicAck
实现确认:
channel.basicConsume(queue, false, consumer);
consumer.handleDelivery(envelope, properties, body) {
//...
channel.basicAck(envelope.getDeliveryTag(), false);
}
8、确认主题
RabbitMQ有几种常见的主题类型,选择使用哪一种主要根据具体的应用场景:
8.1、Direct exchange(默认)
直接交换机,按照routing key完全匹配分发消息到队列。场景是需要指定的队列接收消息。
8.2、Fanout exchange
扇出交换机,会将消息分发到所有绑定的队列。场景是需要广播消息。
8.3、Topic exchange
主题交换机,按照routing key的规则匹配分发消息到队列。场景是需要根据规则分发消息到不同队列。
8.4、Headers exchange
头交换机,按照发送消息的headers属性进行匹配分发消息。场景是需要根据消息头进行路由分发。
选择使用哪种主题类型,主要根据实际的业务需求来判断:
- 如果需要直接将消息发送到指定的队列,使用direct交换机。
- 如果需要广播消息到所有队列,使用fanout交换机。
- 如果需要基于规则匹配分发消息,使用topic交换机。
- 如果需要根据消息头属性进行分发,使用headers交换机。
在代码中,声明交换机时指定类型即可,如下:
channel.exchangeDeclare("myExchange","topic");
9、使用示例
RabbitMQ在Android、IOS、小程序等多端应用中典型应用场景和作用:
9.1、异步处理
各端可以通过RabbitMQ实现任务的异步处理,避免用户等待,提升用户体验。比如小程序下单后,通过RabbitMQ异步把订单信息发送给服务器。
9.2、推送通知
可以通过RabbitMQ实现移动端的消息推送通知,如订单发货通知等。
9.3、数据传输
移动端和服务器端可以通过RabbitMQ进行数据的传输,避免直接耦合,提高传输灵活性。
9.4、负载均衡
RabbitMQ可以在多端和服务器之间进行负载均衡,防止服务器压力过大。
9.5、流量削峰
使用RabbitMQ的消息队列处理请求峰值,防止服务器被瞬时压垮。
9.6、服务解耦
不同端只依赖RabbitMQ进行通信,不需要关注对方技术实现细节,实现服务解耦。
9.7、弹性扩容
通过RabbitMQ可以方便各端与服务器的弹性扩容。
9.8、离线操作支持
移动端可以通过RabbitMQ实现某些离线操作,待网络恢复后再同步到服务器。
实际应用:
本人在实际物联网项目中,用户通过无线设备测量身体指标后,设备通过网络把数据给到后台;后台通过解析数据后,通过MQ把数据给到每个端,通过收到的信息包各个端做相应的处理。
本人使用ConnectionFactory
在创建连接需要配置的时候,通过配置Url
来建立连接等。通过startConsumer
开始消费并监听指定队列,然后定义回调接口DeliverCallback
来接收信息包。最终通过EventBus
来传递数据信息包。
效果展示
具体实现
/**
* @author 拉莫帅
* @date 2023/10/24
* @address
* @Desc EventBus
*/
public class MessageEvent {
private String message;
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
}
/**
* @author 拉莫帅
* @date 2023/10/24
* @address
* @Desc rabbitMQ
*/
public class RabbitMQUtil {
private Connection connection;
private Channel channel;
private ConnectionFactory factory;
private String queueName;
public RabbitMQUtil(String httpType, String userName, String password, String host, String port, String virtualHost, String exChangeName, String bindingKey) {
new Thread(new Runnable() {
@Override
public void run() {
if (connection == null) {
// 创建一个连接
factory = new ConnectionFactory();
try {
StringBuilder builder = new StringBuilder();
StringBuilder stringBuilder = builder.append(httpType).append("://").append(userName).append(":").append(password)
.append("@").append(host).append(":").append(port).append("/").append(virtualHost);
String url = stringBuilder.toString();
Log.e("RabbitMQ", "Url " + url);
// 连接配置
// factory.setHost(Config.MQ_ADDRESS); // 服务器ip
// factory.setPort(Integer.parseInt(Config.MQ_PORT)); // 端口
// factory.setUsername(Config.MQ_USERNAME); // 用户名
// factory.setPassword(Config.MQ_PASSWORD); // 密码
// factory.setVirtualHost(Config.MQ_VIRTUALHOST);
factory.setUri(url);
// 创建一个新的代理连接
connection = factory.newConnection();
// 使用内部分配的通道号创建一个新通道
channel = connection.createChannel();
channel.exchangeDeclare(exChangeName, "topic", true); // 声明一个转发器
queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, exChangeName, bindingKey); // 绑定一个到转发器
Log.e("Waiting for logs.", "");
startConsumer();
} catch (URISyntaxException e) {
e.printStackTrace();
} catch (NoSuchAlgorithmException e) {
e.printStackTrace();
} catch (KeyManagementException e) {
e.printStackTrace();
} catch (Exception e) {
e.printStackTrace();
}
}
}
}).start();
}
/**
* 开始消费
*/
public void startConsumer() throws Exception {
Log.e("startConsumer", "");
// 定义回调接口DeliverCallback
DeliverCallback callback = (consumerTag, message) -> {
String result = new String(message.getBody(), "UTF-8");
Log.e("DeliverCallback >>>", result);
// 创建一个事件
MessageEvent event = new MessageEvent();
event.setMessage(result);
// 通过EventBus发送事件
EventBus.getDefault().post(event);
};
// 启动基本消费,并传入回调接口
channel.basicConsume(queueName, true, callback, consumerTag -> {
});
}
/**
* 关闭连接
*/
public void close() throws Exception {
channel.close();
connection.close();
}
}
public class MainActivity extends AppCompatActivity {
private static final String bindingKey = "topic.chain=2.region=3.area=4.pharmacy=5.";
private RabbitMQUtil rabbitMQUtil;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
initMQ();
}
private void initMQ() {
if (rabbitMQUtil == null) {
rabbitMQUtil = new RabbitMQUtil(Config.MQ_HTTP_TYPE, Config.MQ_USERNAME, Config.MQ_PASSWORD,
Config.MQ_ADDRESS, Config.MQ_PORT, Config.MQ_VIRTUALHOST, Config.MQ_EXCHANGE, bindingKey);
}
}
@Override
public void onStart() {
super.onStart();
EventBus.getDefault().register(this);
}
@Override
public void onStop() {
super.onStop();
EventBus.getDefault().unregister(this);
}
// 接收事件
@Subscribe(threadMode = ThreadMode.MAIN)
public void onMessage(MessageEvent event) {
String message = event.getMessage();
Log.e("接收MQ +++++++++++++", message);
// 更新UI
// ...
}
@Override
protected void onDestroy() {
super.onDestroy();
new Thread(new Runnable() {
@Override
public void run() {
try {
rabbitMQUtil.close();
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
}
}
10、总结
RabbitMQ优秀的性能和灵活性,使其可以处理从简单的请求-响应交互到复杂的异步处理场景。
它既可以用于系统间的异步解耦,也可以实现应用内不同组件的解耦。它非常适合用于分布式系统之间的数据交换与集成,已成为企业级分布式架构的重要组件之一。文章来源:https://www.toymoban.com/news/detail-738163.html
总体来说,RabbitMQ作为一款易用、稳定、功能强大的消息中间件,可以提供高可用、可扩展、低时延的消息服务,在实际项目中使用广泛。并且在当前的技术栈中占有非常重要的地位,是必须要掌握的技能之一。文章来源地址https://www.toymoban.com/news/detail-738163.html
到了这里,关于Android应用集成RabbitMQ消息处理指南的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!