怎么做自己的网站赚钱,泉州做网站排名,人个做外贸用什么网站好,wordpress getshellAndroid应用集成RabbitMQ消息处理指南 RabbitMQ1、前言2、RabbitMQ简介2.1、什么是RabbitMQ2.2、RabbitMQ的特点2.3、RabbitMQ的工作原理2.4、RabbitMQ中几个重要的概念 3、在Android Studio中集成RabbitMQ3.1、在Manifest中添加权限#xff1a;3.2、在build.gradle(:app)下添… Android应用集成RabbitMQ消息处理指南 RabbitMQ1、前言2、RabbitMQ简介2.1、什么是RabbitMQ2.2、RabbitMQ的特点2.3、RabbitMQ的工作原理2.4、RabbitMQ中几个重要的概念 3、在Android Studio中集成RabbitMQ3.1、在Manifest中添加权限3.2、在build.gradle(:app)下添加依赖 4、建立连接4.1、创建ConnectionFactory对象4.2、使用ConnectionFactory创建连接4.3、创建Channel4.4、声明队列、交换机等4.5、生产或消费消息4.6、关闭连接 5、发送消息5.1、创建连接和通道5.2、声明队列5.3、准备消息内容5.4、发布消息5.5、关闭资源5.6、案例 6、接收消息6.1、创建连接和通道6.2、声明队列6.3、定义消费者6.4、监听队列6.5、接收消息6.6、确认消息6.7、关闭资源6.8、案例 7、确认机制RabbitMQ中的确认机制主要分为两种7.1、发布确认(Publisher Confirms)7.2、消费者确认(Consumer Acknowledgments) 8、确认主题RabbitMQ有几种常见的主题类型选择使用哪一种主要根据具体的应用场景8.1、Direct exchange(默认)8.2、Fanout exchange8.3、Topic exchange8.4、Headers exchange 选择使用哪种主题类型,主要根据实际的业务需求来判断: 9、使用示例RabbitMQ在Android、IOS、小程序等多端应用中典型应用场景和作用9.1、异步处理9.2、推送通知9.3、数据传输9.4、负载均衡9.5、流量削峰9.6、服务解耦9.7、弹性扩容9.8、离线操作支持 实际应用效果展示具体实现 10、总结 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 (消息确认)消费者可以启用消息确认机制在收到消息并处理后发送确认回执给RabbitMQRabbitMQ才会将消息从Queue中移除。
3、在Android Studio中集成RabbitMQ
3.1、在Manifest中添加权限
uses-permission android:nameandroid.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 时设置autoAckfalse之后手动调用 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来传递数据信息包。
效果展示 具体实现  {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() {Overridepublic 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, );// 定义回调接口DeliverCallbackDeliverCallback 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.chain2.region3.area4.pharmacy5.;private RabbitMQUtil rabbitMQUtil;Overrideprotected 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);}}Overridepublic void onStart() {super.onStart();EventBus.getDefault().register(this);}Overridepublic 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// ...}Overrideprotected void onDestroy() {super.onDestroy();new Thread(new Runnable() {Overridepublic void run() {try {rabbitMQUtil.close();} catch (Exception e) {e.printStackTrace();}}}).start();}
}10、总结
RabbitMQ优秀的性能和灵活性使其可以处理从简单的请求-响应交互到复杂的异步处理场景。
它既可以用于系统间的异步解耦,也可以实现应用内不同组件的解耦。它非常适合用于分布式系统之间的数据交换与集成已成为企业级分布式架构的重要组件之一。
总体来说RabbitMQ作为一款易用、稳定、功能强大的消息中间件可以提供高可用、可扩展、低时延的消息服务在实际项目中使用广泛。并且在当前的技术栈中占有非常重要的地位是必须要掌握的技能之一。
