基于SpringBoot、RabbitMQ的Android消息推送平台搭建

消息推送，类似于微信来新消息时出现在通知栏那种情景。很多APP都有这个功能。现在有很多第三方平台可以实现这个需要，但是有的公司对所要推送的消息保密要求比较高，不希望被第三方看到，可以使用此种方式进行消息推送。

下面使用SpringBoot、RabbitMQ搭建一个消息推送平台，实现java、C#或者python等后台把消息推送到Android客户端的功能。文末有源码。RabbitMQ的安装网上有很多教程，这里不再讲述。

一、java服务端的代码如下：

这个java程序只是为了向外提供接口，然后java、C#或者python等后台调用此接口，把所要推送的消息传递给此java程序，此程序再传递给RabbitMQ服务器。具体业务逻辑不在这个程序中。

pom.xml中引入如下依赖：

 <parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.0.6.RELEASE</version></parent><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency></dependencies>

application.yml文件如下：

server:port: 10086 #设置微服务端口spring:application:name: app_msg_push #设置微服务名

启动类如下：

@SpringBootApplication
public class MsgPushApplication
{public static void main(String[] args){SpringApplication.run(MsgPushApplication.class, args);}
}

获取RabbitMQ链接的类：

/*** 配置RabbitMQ连接*/
public class ConnectionUtils
{public static Connection getConnection() throws IOException, TimeoutException{ConnectionFactory conn = new ConnectionFactory();conn.setHost("127.0.0.1");  //RabbitMQ服务所在的ip地址conn.setPort(15672);   //RabbitMQ服务所在的端口号conn.setVirtualHost("/test");conn.setUsername("test");conn.setPassword("test");return conn.newConnection();}
}

controller如下：

@Controller
public class MsgPushController
{//设置交换机名称private final static String EXCHANGE_NAME = "app_msg_push_exchange";/*** 把消息传递给RabbitMQ服务器** @param msg ：消息* @param to  ：routingKey匹配规格* @return* @throws Exception*/@PostMapping("send_msg")public ResponseEntity<String> sendMessage(@RequestParam(value = "msg", required = true) String msg, @RequestParam(value = "to", required = true) String to) throws Exception{if (StringUtils.isEmpty(msg) || StringUtils.isEmpty(to)){return ResponseEntity.ok("请设置请求参数！");}// 获取到RabbitMQ服务器连接Connection connection = ConnectionUtils.getConnection();// 获取通道Channel channel = connection.createChannel();// 声明exchange，指定类型为topic，true表示进行持久化channel.exchangeDeclare(EXCHANGE_NAME, "topic", true);// 发送消息，并且指定routing key为tochannel.basicPublish(EXCHANGE_NAME, to, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());//如果发送消息到RabbitMQ服务器失败，每隔10秒再次重试channel.confirmSelect();if (!channel.waitForConfirms()){Timer timer = new Timer();TimerTask timerTask = new TimerTask(){@Overridepublic void run(){try{channel.basicPublish(EXCHANGE_NAME, to, null, msg.getBytes());} catch (IOException e){e.printStackTrace();}}};timer.schedule(timerTask, 0, 10000);}channel.close();connection.close();return ResponseEntity.ok("推送成功！");}
}

如上设置好后，调用的链接就是：http://127.0.0.1:10086/send_msg，参数就是msg=“你好”，另一个参数为to=“app.msg.test”，请求方式为POST，可以使用Postman或者RestClient进行测试。

二、Android客户端的代码如下：

在模块中的build.gradle中引入如下依赖：

 implementation 'com.rabbitmq:amqp-client:4.4.1'

链接RabbitMQ的工具类如下：

/*** 配置RabbitMQ连接*/
public class ConnectionUtils
{public static Connection getConnection() throws IOException, TimeoutException{ConnectionFactory conn = new ConnectionFactory();conn.setHost("127.0.0.1");  //RabbitMQ服务所在的ip地址conn.setPassword("5672");   //RabbitMQ服务所在的端口号conn.setVirtualHost("/test");conn.setUsername("test");conn.setPassword("test");return conn.newConnection();}
}

SP工具类如下：

/*** SharedPreferences的一个工具类。 调用setParam就能保存String, Integer, Boolean, Float,* Long类型的参数。同样调用getParam就能获取到保存在手机里面的数据* */
public class SPUtils {/*** 保存在手机里面的文件名*/private static final String FILE_NAME = "config";/*** 保存数据的方法，我们需要拿到保存数据的具体类型，然后根据类型调用不同的保存方法* * @param context*            上下文* * @param key*            字段名* * @param object*            字段值*/public static void setParam(Context context, String key, Object object) {// 删除String type = object.getClass().getSimpleName();SharedPreferences sp = context.getSharedPreferences(FILE_NAME,Context.MODE_PRIVATE);SharedPreferences.Editor editor = sp.edit();if ("String".equals(type)) {editor.putString(key, (String) object);} else if ("Integer".equals(type)) {editor.putInt(key, (Integer) object);} else if ("Boolean".equals(type)) {editor.putBoolean(key, (Boolean) object);} else if ("Float".equals(type)) {editor.putFloat(key, (Float) object);} else if ("Long".equals(type)) {editor.putLong(key, (Long) object);}editor.commit();}/*** 得到保存数据的方法，我们根据默认值得到保存的数据的具体类型，然后调用相对于的方法获取值* * @param context*            上下文* * @param key*            字段名* * @param defaultObject*            字段默认值* * @return*/public static Object getParam(Context context, String key,Object defaultObject) {String type = defaultObject.getClass().getSimpleName();SharedPreferences sp = context.getSharedPreferences(FILE_NAME,Context.MODE_PRIVATE);if ("String".equals(type)) {return sp.getString(key, (String) defaultObject);} else if ("Integer".equals(type)) {return sp.getInt(key, (Integer) defaultObject);} else if ("Boolean".equals(type)) {return sp.getBoolean(key, (Boolean) defaultObject);} else if ("Float".equals(type)) {return sp.getFloat(key, (Float) defaultObject);} else if ("Long".equals(type)) {return sp.getLong(key, (Long) defaultObject);}return null;}
}

MainActivity如下：

public class MainActivity extends AppCompatActivity
{@Overrideprotected void onCreate(Bundle savedInstanceState){super.onCreate(savedInstanceState);setContentView(R.layout.activity_main); //页面自己定义，消息推送无需任何控件Intent intent = new Intent(this, NotificationService.class);startService(intent); //启动监听消息推送的服务}
}

监听消息推送的Service如下：

/*** 后台监听消息推送的服务*/
public class NotificationService extends Service
{private Context context;private static String EXCHANGE_NAME = "app_msg_push_exchange";private static String QUEUE_NAME = "";@Overridepublic IBinder onBind(Intent intent){return null;}@Overridepublic void onCreate(){  // 当服务第一次被开启的时候调用super.onCreate();context = this;//设置每个用户对应一个队列if (TextUtils.isEmpty((String) SPUtils.getParam(context, "QUEUE_NAME", ""))){SPUtils.setParam(context, "QUEUE_NAME", "app_msg_push_queue_" + UUID.randomUUID());}QUEUE_NAME = (String) SPUtils.getParam(context, "QUEUE_NAME", "");getDataFromMQ();}/*** 从消息队列获取数据*/private void getDataFromMQ(){new Thread(new Runnable(){@Overridepublic void run(){try{// 获取到连接Connection connection = ConnectionUtils.getConnection();// 获取通道final Channel channel = connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, true, false, false, null);// 绑定队列到交换机，同时指定需要订阅的routing key。channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "app.msg.#");// 定义队列的消费者DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException{// body 即消息体final String msg = new String(body);showNotifictionIcon(context, getAppName(context), msg);//手动设置ACKchannel.basicAck(envelope.getDeliveryTag(), false);}};// 监听队列，手动ACKchannel.basicConsume(QUEUE_NAME, false, consumer);} catch (IOException e){e.printStackTrace();} catch (TimeoutException e){e.printStackTrace();}}}).start();}public void showNotifictionIcon(Context context, String title, String content){NotificationManager manager = (NotificationManager) context.getSystemService(context.NOTIFICATION_SERVICE);NotificationCompat.Builder builder = new NotificationCompat.Builder(context);builder.setAutoCancel(true);//点击后消失builder.setLargeIcon((getBitmap(context)));//设置通知栏消息标题的头像builder.setSmallIcon(R.mipmap.ic_launcher);builder.setDefaults(NotificationCompat.DEFAULT_SOUND);//设置通知铃声builder.setContentTitle(title);//设置标题builder.setContentText(content);//设置内容builder.setPriority(Notification.PRIORITY_DEFAULT); //设置该通知优先级//利用PendingIntent来包装我们的intent对象,使其延迟跳转 设置通知栏点击意图builder.setContentIntent(createIntent(context, title + content));manager.notify(new Random().nextInt(20), builder.build());}/*** 创建通知栏消息点击后跳转的intent。*/public PendingIntent createIntent(Context context, String data){Intent intent = new Intent(context, NotifyClickReceiver.class);Bundle mBundle = new Bundle();mBundle.putString("data", data);intent.putExtras(mBundle);intent.setAction("com.example.myapp");PendingIntent contentIntent = PendingIntent.getBroadcast(context, 0, intent, PendingIntent.FLAG_CANCEL_CURRENT);return contentIntent;}/*** 获取应用图标bitmap*/public static Bitmap getBitmap(Context context){PackageManager packageManager = null;ApplicationInfo applicationInfo = null;try{packageManager = context.getApplicationContext().getPackageManager();applicationInfo = packageManager.getApplicationInfo(context.getPackageName(), 0);} catch (PackageManager.NameNotFoundException e){applicationInfo = null;}Drawable d = packageManager.getApplicationIcon(applicationInfo); //xxx根据自己的情况获取drawableBitmapDrawable bd = (BitmapDrawable) d;Bitmap bm = bd.getBitmap();return bm;}/*** 获取应用程序名称*/public static String getAppName(Context context){try{PackageManager packageManager = context.getPackageManager();PackageInfo packageInfo = packageManager.getPackageInfo(context.getPackageName(), 0);int labelRes = packageInfo.applicationInfo.labelRes;return context.getResources().getString(labelRes);} catch (Exception e){e.printStackTrace();}return null;}
}

点击通知栏后需要跳转，使用一个广播监听并获取数据：

/*** 点击通知栏跳转的广播*/
public class NotifyClickReceiver extends BroadcastReceiver
{@Overridepublic void onReceive(Context context, Intent intent){String action = intent.getAction();String data = intent.getStringExtra("data");if (action.equals("com.example.myapp")){Intent i = new Intent(context, getMsgActivity.class);i.putExtra("data", data);i.setFlags(Intent.FLAG_ACTIVITY_NEW_TASK); //从四大组件的其他三个组件跳转到Activity,需要设置FLAG_ACTIVITY_NEW_TASKcontext.startActivity(i);}}
}

获取消息并处理的Activity如下：

public class getMsgActivity extends Activity
{private TextView tv;@Overrideprotected void onCreate(Bundle savedInstanceState){super.onCreate(savedInstanceState);setContentView(R.layout.activity_getmsg);tv = findViewById(R.id.tv);Intent intent = getIntent();String data = intent.getStringExtra("data");tv.setText("获取到新消息：" + data);}
}

别忘了在AndroidManifest.xml中添加权限，并且声名Service和BroadcastReceiver:

......<uses-permission android:name="android.permission.INTERNET" />......<receiver android:name=".NotifyClickReceiver" />
<service android:name=".NotificationService" />
.....

先在几个Android设备上运行上面APP；再通过POST请求访问http://127.0.0.1:10086/send_msg，并设置参数为msg=“你好”，另一个参数为to=“app.msg.test”；运行可见APP都能够接收到消息，亲测可用。后面就可以在java、C#或者python等后台调用此接口，实现消息推送功能。

上面的RoutingKey是写死的app.msg.#，推送的时候是app.msg.test；在实际使用中，读者可以自定义设置RabbitMQ的RoutingKey，以匹配合适的用户进行推送消息。

如果所要推送的人数(即消息队列个数)比较多的话，并发量比较大，可以搭建RabbitMQ集群解决此问题。