RabbitMQ消息队列Android的实现方式之发送（一）

时间过得真快，转眼间接手的项目已经到了上线冲刺期了，所以最近几天一直很忙。在提前完成了Android端一部分开发任务后，抽出时间梳理一下关于RabbitMQ消息队列在Android中使用的方法，希望能给在项目中需要RabbitMQ的新手一点帮助。

关于RabbitMQ是什么？

MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。其中较为成熟的MQ产品有IBM WEBSPHERE MQ等等。

RabbitMQ中几个重要的关键字段

Exchange: RabbitMQ中的交换机，在这里定制通讯规则；

Queue: 消息队列；

Channel: 收发消费所使用的通道；

Bind: 绑定了Queue和Exchange，定义好队列名和交换机规则使用Bind绑定即可；

RabbitMQ初始化连接代码


public void initConnectionFactory() {//RabbitMQ初始化连接

    ConnectionFactory factory = new ConnectionFactory();

    factory.setHost(IP地址);

    factory.setPort(端口号);

    factory.setUsername(用户名);

    factory.setPassword(密码);

    factory.setAutomaticRecoveryEnabled(false);//设置断开后是否自动连接

}

创建消息生产者代码


public void publishToAMQP(final Handler handler) {//创建生产者，并监听生产信息

    publishThread = new Thread(() ->{//创建子线程

        while (true) {

            try {

                connection = factory.newConnection();//创建新的连接

                Channel ch= connection.createChannel();//创建通道

                ch.confirmSelect();

                while (true) {//轮询监听生产信息

                    String message= queue.takeFirst();//检索队列

                    try {

                        //定义中间件规则，定义生产队列

                        ch.basicPublish("CalonDirectExchange", "test", null, message.getBytes());

                        Log.d("", "[s] " + message);

                        ch.waitForConfirmsOrDie();

                    } catch (Exception e) {

                        Log.d("", "[f] " + message);

                        queue.putFirst(message);

                        throw e;

}

}

            } catch (InterruptedException e) {

                break;

            } catch (Exception e) {

                Message message= new Message();

                message.what = 400;

                Bundle bundle= new Bundle();

                if (reconnection_count > 5) {//自定义重连方法规则

                    String strConn= "尝试重连失败！";

                    bundle.putString("conn_interrupt", strConn);

                    bundle.putBoolean("close_conn", true);

                    message.setData(bundle);

                    handler.sendMessage(message);

                    reconnection_count = 1;

}

                String strConn= "连接中断，正在尝试第" + reconnection_count + "次重连";

                bundle.putBoolean("close_conn", false);

                bundle.putString("conn_interrupt", strConn);

                message.setData(bundle);

                handler.sendMessage(message);

                reconnection_count++;

                Log.d("", "Connection broken: " + e.getClass().getName());

                try {

                    Thread.sleep(5000); //sleep and then try again

                } catch (InterruptedException e1) {

                    break;

}

}

}

    });

    publishThread.start();//start thread

}

发送临时消息代码


private BlockingDeque<String> queue = new LinkedBlockingDeque<>();//创建队列容器

public void publishMessage(String message) {//发送临时消息

    try {

        Log.d("", "[q] " + message);

        queue.putLast(message);

    } catch (InterruptedException e) {

        e.printStackTrace();

}

}

至此，RabbitMQ发送消息的方法流程就完成了。

最后编辑于：2019.07.11 13:56:49