终篇:理解并使用RabbitMQ

作者: 前山饭店 | 来源:发表于2017-08-15 20:56 被阅读146次

    说明

    RabbitMQ中MQ是MessageQueue的简写,整体的意思应该是像兔子一样快的高效消息中间件,组件本质上就是生产者消费者模式,由一边接受消息,另一边转发消息,其中可以对消息进行缓存,转发或清除。

    一般实际中使用情况是推送和IM(及时通讯)

    如何使用

    首先要含有封装底层的jar包,下载RabbitMQ-java-Client,下载之后放在AndroidStudio项目中libs包中编译,或者在module中build.gradle中添加依赖:

    dependencies {
        compile 'com.rabbitmq:amqp-client:4.2.0'
    }
    

    RabbitMQ是由pivotal公司开源的一个项目,在Github上可以查看的源码。

    需要访问网络,不要忘记在清单文件中添加网络权限:

    <uses-permission android:name="android.permission.INTERNET"/>
    

    一张简易的图表示工作原理:

    RabbitMQ.png

    连接

    首先配置连接服务端基本信息:

        ConnectionFactory factory = new ConnectionFactory();  //创建连接工厂
        
        //设置服务端连接认证:ip号,端口号,登录名,密码
        factory.setHost(“HOST_ID");
        factory.setPort(PORT);
        factory.setUsername("USERNAME");
        factory.setPassword("PASSWORD")
        
        factory.setAutomaticRecoveryEnabled(true); //设置自动恢复连接
        factory.setNetworkRecoveryInterval(5000);  //设置自动连接间隔(毫秒)
    

    以上都是一些基本配置,设置服务端上认证信息,设置恢复连接机制,在已经连接过之后断开连接,会主动尝试连接服务,直至连接上服务或者关闭连接工厂。

    如果第一没有连接上,则会报错。需要循环创建连接,直至第一次连接很成功之后才会自动连接,包括所有的通道(Channel)。

    现在有了连接工厂,创建一个连接就够多个通道使用,也是高效率的方式:

    Connection connection = factory.newConnection();
    

    注意这个需要在子线程中执行,否则会报错误: Android.os.NetworkOnMainThreadException。

    发送消息

    • 建立通道及配置
    • 发送消息
    Channel channelSend = connection.createChannel();
    //第一种方式,直接指定消费队列
    1.channelSend.queueDeclare(queueName, false, false, false, null);
      channelSend.basicPublish("", queueName, null, message.getBytes());
        
    //第二种方式,指定转换器和绑定密钥(routingkey)
    2.channelSend.exchangeDeclare(EXCHANGENAME, "topic"); //类型四种:fanout,direct,topic,handers
      channelSend.basicPublish(EXCHANGENAME, routingkey, null, message.getBytes());
    

    就是这么简单,但还是需要简单说明。

    完整的流程:发送者->转换器->队列->接受者

    第一种情况,虽然没有声明转换器,但是会使用匿名转换器,发送到routingKey为队列名的队列中。

    第二种情况,只说明转换器的四种类型特点:

    • fanout类型:忽略routingkey的值发送给所有绑定该类型的队列中,
    • direct类型:根据routingkey的值发送给匹配该值的队列中,
    • topic类型:它的routingKey可以使用*和#来表示,前者表示一个连着的单词,如work;后者表示零个或多个连着的单词,如work.first。只要发送消息的routingKey匹配接受者设置绑定exchange的routingKey的值,接受者队列就可以收到到由exchange发送的消息。
    • headers类型:以匹配键值对的形式发送和接受消息。匹配有两种方式all和any。用的比较少,所以就没有深入了解。

    接受消息

    • 建立通道及配置
    • 设置监听通道消息
    Channel channel = connection.createChannel();
    
    //不定义装换器(默认匿名装换器),命名队列,获取实时和缓存在队列中的消息
    1. final Channel channel = connection.createChannel();
        channel.queueDeclare(queueName, false, false, false, null);
    
        channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                String message = new String(body, "UTF-8");
                Log.d(TAG, queueName + "接受消息->" + message);
                channel.basicAck(envelope.getDeliveryTag(), false);  //消息应答
            }
        });
    
    //自定义装换器名称和类型,匿名队列,监听符合routingKey的消息,只能获取实时消息            
    2. final Channel channel = connection.createChannel();
    
        channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE);
        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE_NAME, routingKey);
    
        channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                String message = new String(body, "UTF-8");
                Log.d(TAG, routingKey + "接受消息->" + message);
                channel.basicAck(envelope.getDeliveryTag(), false);  //消息应答
            }
        });
            
    //自定义装换器名称和类型,命名队列,监听符合routingKey的消息,获取实时和缓存在队列中的消息
    3. final Channel channel = connection.createChannel();
    
        channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE);
        channel.queueDeclare(queueName, false, false, false, null);
        channel.queueBind(queueName, EXCHANGE_NAME, routingKey);  //设置绑定
    
        channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                String message = new String(body, "UTF-8");
                Log.d(TAG, routingKey + queueName + "接受消息->" + message);
                channel.basicAck(envelope.getDeliveryTag(), false);  //消息应答
            }
        });
    
    

    以上是主要的三种监听消息方式:

    第一种,匿名转换器,命名队列,监听队列中的消息。只能一对一发送并且接受,局限比较大。

    第二种,指定转换器类型,匿名队列,监听实时routingKey的消息,筛选出匹配routingKey消息进行发送,所以有可能发送到多个队列中。每次连接都是获取新的空的队列,并且失去连接后队列将被删除,所以只能获取实时的消息。

    第三种,指定装换器类型,命名队列,监听队列中routingKey的消息,筛选出匹配routingKey消息进行发送,所以有可能发送到多个队列中。获取实时和缓存在队列中的消息

    匿名队列的特点:

    1. 一旦被创建就是一个新的空的队列
    2. 一旦失去消费者连接该队列就会被服务端删除

    命名队列特点:

    1. 默认循环均匀发送消息给多个消费者,但是可以设置channel.basicQos(1)高效的发送消息给多个消费者。
    2. 消息应答机制,服务端确认消息被销毁才会移除该消息。
    3. 队列和消息持久化,在服务端奔溃或者重启是可以保存队列和消息。
    4. 一个消息只能被一个消费者消费一次。

    关闭

    if (channelSend != null && channelSend.isOpen()) {
        try {
            channelSend.close();
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
    }
    if (channelReceive != null && channelReceive.isOpen()) {
        try {
            channelReceive.close();
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
    }
    if (connection != null && connection.isOpen()) {
        try {
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    

    封装RabbitMQ-Android的简单使用,欢迎下载使用,并且反馈

    注:这是RabbitMQ-java版Client的指导教程翻译系列文章,欢迎大家批评指正
    第一篇Hello Word了解RabbitMQ的基本用法
    第二篇Work Queues介绍队列的使用
    第三篇Publish/Subscribe介绍转换器以及其中fanout类型
    第四篇Routing介绍direct类型转换器
    第五篇Topics介绍topic类型转换器
    第六篇RPC介绍远程调用

    相关文章

      网友评论

        本文标题:终篇:理解并使用RabbitMQ

        本文链接:https://www.haomeiwen.com/subject/ilovrxtx.html