时间过得真快，转眼间接手的项目已经到了上线冲刺期了，所以最近几天一直很忙。在提前完成了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发送消息的方法流程就完成了。
网友评论