RabbitMQ环境搭建就不说了
使用Exchange 进行推送，会推送到该交换机下面所有创建的Queues，如果使用Queues内推送，你会发现设备会轮流收到服务端的推送。
image.png
直接开始代码
RabbitMQUtils
/**
* Author: RealChen
* Date: 2019/12/9
* Description:
*/
public class RabbitMQUtils {
private static String exchangeName = "test_topic_exchange";
private static final String routingKey = "test_queue";
private static ConnectionFactory factory = new ConnectionFactory();
/**
* Rabbit配置
*/
public static void setupConnectionFactory() {
String userName = "test";
String passWord = "12345";
String hostName = "10.x.x.x";
int portNum = 5672;
factory.setUsername(userName);
factory.setPassword(passWord);
factory.setHost(hostName);
factory.setPort(portNum);
}
/**
* 发消息
*/
public static void basicPublish(String sendMsg) {
new Thread() {
@Override
public void run() {
super.run();
try {
//连接
Connection connection = factory.newConnection();
//通道
Channel channel = connection.createChannel();
//消息发布
byte[] msg = sendMsg.getBytes();
//rountingKey 自己任意填写
channel.basicPublish(exchangeName, routingKey, null, msg);
} catch (IOException | TimeoutException e) {
e.printStackTrace();
}
}
}.start();
}
/**
* 收消息
*/
public static void basicConsume() {
new Thread() {
@Override
public void run() {
super.run();
try {
//连接
Connection connection = factory.newConnection();
//通道
final Channel channel = connection.createChannel();
//声明了一个交换和一个服务器命名的队列，然后将它们绑定在一起。
channel.exchangeDeclare(exchangeName, "direct", true);
AMQP.Queue.DeclareOk declareOk = channel.queueDeclare();
String queueName = declareOk.getQueue();
channel.queueBind(queueName, exchangeName, routingKey);
//实现Consumer的最简单方法是将便捷类DefaultConsumer子类化。可以在basicConsume 调用上传递此子类的对象以设置订阅：
channel.basicConsume(queueName, false, "administrator", 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 rountingKey = envelope.getRoutingKey();
//contentType字段，如果尚未设置字段，则为null。
String contentType = properties.getContentType();
//接收到的消息
String msg = new String(body, "UTF-8");
LogUtils.d("推送过来的消息=====" + msg);
//交付标记
long deliveryTag = envelope.getDeliveryTag();
channel.basicAck(deliveryTag, false);
//发送广播
sendBroadcast(msg);
}
});
} catch (IOException | TimeoutException e) {
e.printStackTrace();
}
}
}.start();
}
/**
* 发送广播
*/
private static void sendBroadcast(String msg) {
LogUtils.d("=========sendBroadcast");
Intent intent = new Intent();
intent.setFlags(Intent.FLAG_ACTIVITY_NEW_TASK);
intent.setAction("com.xxx.xxxx");
intent.putExtra("content", msg);
intent.setComponent(new ComponentName(MyApplication.getInstance(), MyApplication.getInstance().getPackageName() + ".receiver.MyReceiver"));
MyApplication.getInstance().sendBroadcast(intent);
}
}
MyReceiver
public class MyReceiver extends BroadcastReceiver {
private NotificationManager mNotificationManager;
private Notification mNotification;
private int notificationId;
private Intent mIntent;
private PendingIntent mPendingIntent;
private PushContentModel pushContentModel;
@Override
public void onReceive(Context context, Intent intent) {
String action = intent.getAction();
String msg = intent.getStringExtra("content");
try {
pushContentModel = FastJsonUtils.getModel(msg, PushContentModel.class);
} catch (Exception e) {
LogUtils.d("====消息格式异常");
}
if (null == pushContentModel) {
return;
}
LogUtils.d("====", action);
notificationId = (int) TimeUtils.getNowMills();
switch (action) {
case "com.ztt.ztam":
mIntent = new Intent(context, MainActivity.class);
mIntent.setFlags(Intent.FLAG_ACTIVITY_NEW_TASK | Intent.FLAG_ACTIVITY_CLEAR_TOP);
mPendingIntent = PendingIntent.getActivity(context, 0, mIntent, PendingIntent.FLAG_UPDATE_CURRENT);
//相应处理
if (mNotificationManager == null) {
mNotificationManager = (NotificationManager) context.getSystemService(context.NOTIFICATION_SERVICE);
}
//8.0 以后需要加上channelId 才能正常显示
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
String channelId = "default";
String channelName = "默认通知";
mNotificationManager.createNotificationChannel(new NotificationChannel(channelId, channelName, NotificationManager.IMPORTANCE_HIGH));
}
mNotification = new Notification();
//构建通知
Notification.Builder builder;
if (android.os.Build.VERSION.SDK_INT >= android.os.Build.VERSION_CODES.O) {
builder = new Notification.Builder(context, "default")
.setSmallIcon(R.mipmap.ic_launcher)
.setContentTitle(pushContentModel.getTitle())
.setContentText(pushContentModel.getContent())
.setLargeIcon(BitmapFactory.decodeResource(context.getResources(), R.mipmap.ic_launcher))
.setContentIntent(mPendingIntent);
} else {
builder = new Notification.Builder(context)
.setSmallIcon(R.mipmap.ic_launcher)
.setContentTitle(pushContentModel.getTitle())
.setContentText(pushContentModel.getContent())
.setLargeIcon(BitmapFactory.decodeResource(context.getResources(), R.mipmap.ic_launcher))
.setContentIntent(mPendingIntent);
}
mNotification = builder.build();
//mNotification.sound = Sound;
mNotification.defaults |= Notification.DEFAULT_VIBRATE;
mNotification.flags |= Notification.FLAG_AUTO_CANCEL;
mNotificationManager.notify(notificationId, mNotification);
break;
}
}
}
网友评论