服务端
pom文件
org.springframework.boot
spring-boot-starter-amqp
org.springframework.boot
spring-boot-starter-web
配置文件
server:
port: 8022
spring:
#给项目来个名字
application:
name: rabbitmq-consumer
#配置rabbitMq 服务器
rabbitmq:
host: 127.0.0.1
port: 5672
username: snakey
password: 123
#虚拟host 可以不设置,使用server默认host
virtual-host: /
RabbitMQ配置文件
package com.xiaoge.mqdemo.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* 作者：guoyzh
* 时间：2019/10/13 13:22
* 功能：测试mq
*/
@Configuration
public class RabbitMQConfig {
// 设置使用的交换机
public final static String EXCHANGE_TOP = "test";
public final static String EXCHANGE_FANOUT = "amq.fanout";
// 设置队列
public final static String QUEUE_MSG = "topic.msg";
public final static String QUEUE_MSG_ALL = "topic.msg.all";
public final static String QUEUE_ALL = "topic.#";
@Bean
public Queue pushMsgQueue() {
return new Queue(RabbitMQConfig.QUEUE_MSG, false);
}
@Bean
TopicExchange topExchange() {
return new TopicExchange(EXCHANGE_TOP);
}
@Bean
FanoutExchange fanoutExchange() {
return new FanoutExchange(EXCHANGE_FANOUT);
}
// queue绑定exchange的时候 routingKey(with())中的参数很重要 决定着这个queue会接收哪些路由key的消息
@Bean
Binding bindingTopExchangeMessage() {
return BindingBuilder.bind(pushMsgQueue()).to(topExchange()).with(QUEUE_MSG);
}
@Bean
Binding bindingFanoutExchangeMessage() {
return BindingBuilder.bind(pushMsgQueue()).to(fanoutExchange());
}
}
接口
package com.xiaoge.mqdemo.web;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.xiaoge.mqdemo.config.RabbitMQConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
/**
* 作者：guoyzh
* 时间：2019/10/13 13:24
* 功能：
*/
@Controller
public class TestController {
@Autowired
RabbitTemplate mRabbitTemplate;
@GetMapping("/send_msg")
@ResponseBody
public String sendTopicMessage() {
// 组建消息
String messageId = String.valueOf(UUID.randomUUID());
String messageData = "message: 分诊室1号发来请求";
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
MapmanMap = new HashMap<>();
manMap.put("messageId", messageId);
manMap.put("messageData", messageData);
manMap.put("createTime", createTime);
// 将消息转换为json
ObjectMapper mapper = new ObjectMapper();
String writeValueAsString = null;
try {
writeValueAsString = mapper.writeValueAsString(manMap);
} catch (JsonProcessingException e) {
e.printStackTrace();
}
// 发送消息
mRabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_TOP, RabbitMQConfig.QUEUE_MSG_ALL, writeValueAsString);
return "ok";
}
}
Android
gradle
implementation 'com.rabbitmq:amqp-client:4.4.1'
gradle指定jdk版本 放在android{…}中
compileOptions {
sourceCompatibility JavaVersion.VERSION_1_8
targetCompatibility JavaVersion.VERSION_1_8
}
界面功能
package com.hencego.myapplication;
import android.os.Bundle;
import android.os.Handler;
import android.os.Message;
import android.util.Log;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.TimeoutException;
import androidx.appcompat.app.AppCompatActivity;
/**
* 作者：guoyzh
* 时间：2019/10/13 15:41
* 功能：
*/
public class MainActivity extends AppCompatActivity {
ConnectionFactory factory;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
//连接设置
setupConnectionFactory();
//用于从线程中获取数据，更新ui
final Handler incomingMessageHandler = new Handler() {
@Override
public void handleMessage(Message msg) {
String message = msg.getData().getString("msg");
Log.i("test", "msg:" + message);
}
};
//开启消费者线程
//subscribe(incomingMessageHandler);
new Thread(new Runnable() {
@Override
public void run() {
basicConsume(incomingMessageHandler);
}
}).start();
}
/**
* 连接设置
*/
private void setupConnectionFactory() {
factory = new ConnectionFactory();
factory.setHost("192.168.1.222");
factory.setPort(5672);
factory.setUsername("snakey");
factory.setPassword("123");
factory.setVirtualHost("/");
}
/**
* 收消息(从发布者那边订阅消息)
*/
private void basicConsume(final Handler handler) {
try {
// 连接
Connection connection = factory.newConnection();
// 通道
final Channel channel = connection.createChannel();
String uuid = String.valueOf(UUID.randomUUID());
String QUEUE_NAME = "topic." + uuid;
// 创建非永久队列
channel.queueDeclare(QUEUE_NAME, false, false, true, null);
// 绑定队列到交换机
channel.queueBind(QUEUE_NAME, "test", "topic.#");
// 接收消息 实现Consumer的最简单方法是将便捷类DefaultConsumer子类化。可以在basicConsume 调用上传递此子类的对象以设置订阅：
channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
String msg = new String(body);
long deliveryTag = envelope.getDeliveryTag();
channel.basicAck(deliveryTag, false);
//从message池中获取msg对象更高效
Message uimsg = handler.obtainMessage();
Bundle bundle = new Bundle();
bundle.putString("msg", msg);
uimsg.setData(bundle);
handler.sendMessage(uimsg);
}
});
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
}