Android应用集成RabbitMQ消息处理指南

这篇具有很好参考价值的文章主要介绍了Android应用集成RabbitMQ消息处理指南。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

Android应用集成RabbitMQ消息处理指南,Android控件的应用及使用场景,开发技巧,MQ,android,rabbitmq,分布式

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来传递数据信息包。

效果展示

Android应用集成RabbitMQ消息处理指南,Android控件的应用及使用场景,开发技巧,MQ,android,rabbitmq,分布式
Android应用集成RabbitMQ消息处理指南,Android控件的应用及使用场景,开发技巧,MQ,android,rabbitmq,分布式

具体实现

Android应用集成RabbitMQ消息处理指南,Android控件的应用及使用场景,开发技巧,MQ,android,rabbitmq,分布式
Android应用集成RabbitMQ消息处理指南,Android控件的应用及使用场景,开发技巧,MQ,android,rabbitmq,分布式

/**
 * @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优秀的性能和灵活性,使其可以处理从简单的请求-响应交互到复杂的异步处理场景

既可以用于系统间的异步解耦,也可以实现应用内不同组件的解耦。它非常适合用于分布式系统之间的数据交换与集成,已成为企业级分布式架构的重要组件之一。

总体来说,RabbitMQ作为一款易用、稳定、功能强大的消息中间件,可以提供高可用、可扩展、低时延的消息服务,在实际项目中使用广泛。并且在当前的技术栈中占有非常重要的地位,是必须要掌握的技能之一。文章来源地址https://www.toymoban.com/news/detail-738163.html

到了这里,关于Android应用集成RabbitMQ消息处理指南的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 【RabbitMQ与SpringBoot集成测试收发消息】

    安装环境:虚拟机VMWare + Centos7.6 + Maven3.6.3 + JDK1.8 RabbitMQ版本:rabbitmq-server-3.8.8-1.el7.noarch.rpm 编程工具Idea + 运行JDK为17 在RabbitMQ的UI界面或命令行上 创建新的Virtual Host ,取名为 vhtest02 ,如下图所示: 使用Idea的 Spring Initializr 创建生产者工程 springrabbitmqtest ,坐标如下: 配置

    2024年02月13日
    浏览(38)
  • Android应用-flutter使用Positioned将控件定位到底部中间

    要将Positioned定位到屏幕底部中间的位置,你可以使用MediaQuery来获取屏幕的高度,然后设置Positioned的bottom属性和left或right属性,一般我们left和right都会设置一个值让控制置于合适的位置,那么如何使其位于底部中央? 以下是一个示例代码: 在这个例子中,Positioned包含一个具

    2024年02月03日
    浏览(46)
  • 【SpringBoot笔记29】SpringBoot集成RabbitMQ消息队列

    这篇文章,主要介绍SpringBoot如何集成RabbitMQ消息队列。 目录 一、集成RabbitMQ 1.1、引入amqp依赖 1.2、添加连接信息 1.3、添加RabbitMQ配置类

    2023年04月08日
    浏览(55)
  • Android应用开发-Flutter的LongPressDraggable控件回调函数onDraggableCanceled使用

    以下是如何使用 onDraggableCanceled 的示例: velocity 参数表示拖动被取消时的速度信息。 offset 参数表示拖动被取消时的偏移量信息。 这个回调通常用于在拖动被取消时执行一些清理工作或展示一些反馈。例如,你可能想要将拖动对象返回到原始位置,或者显示一个提示,告诉用

    2024年03月08日
    浏览(42)
  • Java在物联网领域的应用非常广泛,涵盖了设备连接、数据处理、应用程序开发、安全性、嵌入式系统开发、消息队列和流处理、机器学习和人工智能以及跨平台和多语言集成等方面

    Java作为一种通用编程语言,在物联网(IoT)领域的应用也非常广泛。以下是一些Java在物联网中的典型应用: 开发物联网应用程序 :Java是一种高级编程语言,具有丰富的库和工具,使得开发物联网应用程序变得容易。Java可以用于开发各种物联网应用程序,如智能家居、智能

    2024年02月03日
    浏览(72)
  • 清空所有队列和消息的RabbitMQ操作指南

    RabbitMQ是一种功能强大的消息代理,用于在应用程序之间传递消息。有时候,我们可能需要清空RabbitMQ中的所有队列和消息,以便重新开始或进行测试。本文将介绍如何使用RabbitMQ的管理插件和AMQP协议来清空所有队列和消息。 步骤1:安装RabbitMQ管理插件 要使用RabbitMQ的管理插

    2024年02月06日
    浏览(39)
  • 在WPF应用中使用GongSolutions.WPF.DragDrop实现列表集合控件的拖动处理

    WPF应用中,控件本身也可以通过实现事件代码实现拖动的处理,不过如果我们使用GongSolutions.WPF.DragDrop来处理,事情会变得更加简单轻松,它支持很多控件的拖动处理,如ListBox, ListView, TreeView, DataGrid等源自ItemsControl的控件,本篇随笔介绍在工作流模块中拖动TreeView和DataGrid列表

    2024年02月05日
    浏览(38)
  • RabbitMQ消息堆积方案处理

    在消息队列中,消息堆积是生产环境中的需要考虑的问题,一旦消息产生积压,来不及消费,可能会导致MQ服务器宕机,而解决消息积压有这样一些方案解决: 1.增加消费者数量 可以根据业务情况适当添加多台服务器部署消费者服务实例,消费者数量增加,可以有效提高消息

    2024年02月11日
    浏览(52)
  • RabbitMQ怎么处理消息事务

    在 RabbitMQ 中,可以通过以下两种方式实现消息事务: 发送方确认(Publisher Confirms) :这是 RabbitMQ 提供的一种轻量级事务机制。在发送消息之前,发送方可以要求 RabbitMQ 确认消息是否成功投递到交换机(Exchange)中。如果确认失败,发送方可以选择重试或者处理发送失败的情

    2024年02月07日
    浏览(40)
  • rabbitmq消息异常处理

    提示:文章写完后,目录可以自动生成,如何生成可参考右边的帮助文档 在使用rabbitmq时,会因为各种原因(网络波动,系统宕机,程序异常等)导致消息发送失败。rabbitmq也提供了相应的处理机制。 提示:以下是本篇文章正文内容,下面案例可供参考 生产法发送失败 配置回调

    2023年04月09日
    浏览(35)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包