通过Activemq搭建消息推送服务器实现局域网内android APP的消息推送

「爱情、让人受尽委屈。」 2022-06-01 03:25 983阅读 0赞

（1）消息推送的作用  
目前的手机APP多数都具有消息推送的功能。消息推送我认为其作用和价值有二：一是向所有APP用户发布消息公告；二是在业务流中及时向特定用户发出提醒，告知某请求的处理进度。  
  
  
（2）消息推送的方式和解决方案  
消息推送的方式有多种，大的方面分为pull和push两种方式。pull方式的主要解决方案就是定时轮询，定时隔一段时间去服务器获取一下信息，但这种方式耗费网络流量，同时也耗费电量，实时性也不是很好，因此不是一个好的解决方案。push方式的解决方案有多种。  
第一，SMS的方式。通过拦截SMS消息并且解析消息内容来实现。这个方案可以实现实时操作，但成本较高，需要向移动公司缴纳费用。  
第二，持久连接方式。通过与服务器保持持久连接来实现消息的实时推送。IOS采用的便是这种方式。但IOS可以不代表android也可以。因为IOS对APP管控非常严格，每台手机仅仅保持一个与服务器的连接。但Android很难实现一个可靠的服务。  
第三，GCM推送消息。GCM，全称为Google Cloud Messaging，译为Google云端通讯。它能够让第三方应用的开发者把通知消息或信息从服务器发送到所有使用这个应用的安卓系统或Chrome浏览器的应用或拓展上。大部分国内应用没有使用GCM服务，而大量国外应用使用了GCM服务。原因很简单，思考下google搜索为什么不能用就能明白了。那么，我们Android系统上的应用又是如何在没有使用GCM服务的情况下把消息推送给我们的呢？继续看下面的几种方案。  
第四，采用第三方推送。如极光，信鸽，小米等。有收费的也有免费的，并且都有相关的参考文档，不再赘述。  
第五，采用使用了GCM服务的应用，在系统层级方面由第三方应用的服务器把消息发送给GCM服务器再转接到各个用户。比如比如Facebook、Twitter等。这个方案没有具体研究。  
第六，自己搭建推送平台。稍后详述。  
  
  
（3）消息推送平台的搭建  
由于开发的APP需要在专网中运行，不具备采用第三方推送服务的网络环境，同时客户不想出SMS的费用，在国内GCM又可望而不可及，客户又想把业务流程实时推送给指定用户没办法只好硬着头皮自己搭建推送平台了。  
实现方式概括起来说就是采用apache的activemq发送和接收消息，采用发布/订阅的模式来实现消息的实时推送。activemq是什么就不细说了，有兴趣的同行很容易在网上找到详细资料。我之所以选择activemq，其原因在于其支持 OpenWire,Stomp REST,WS Notification,XMPP,AMQP等协议，完全支持完全支持JMS1.1和J2EE 1.4规范，同时可以很容易内嵌到使用Spring的系统里面去。而我们项目的服务端就是基于spring框架的。这么说来activemq有点像为我量身定做的。尽管如此，使用过程中仍遇到了不少坑，费了不少力气。下面进行详细说明。  
第一，从官网下载activemq并解压。我采用的是5.15.2版本。64位系统执行\\apache-activemq-5.15.2\\bin\\win64\\activemq.bat；32位系统执行\\apache-activemq-5.15.2\\bin\\win32\\activemq.bat即可完成activemq的启动。需要注意的是该版本activemq需要jdk1.8及以上的支持。其开放的端口可以到配置文件apache-activemq-5.15.2\\conf\\ activemq.xml中去看，修改端口时对transportConnector节点进行修改重启服务即可。启动完成后，浏览器输入http://localhost:8161/admin/即可验证是否正常启动。用户名和密码默认是admin/admin。  
第二，消息生产者环境的配置和编码（即Spring+Activemq的集成方法）。  
首先，maven环境pom.xml文件中增加依赖：

<dependency>
    		  <groupId>org.apache.activemq</groupId>
    		  <artifactId>activemq-all</artifactId>
    		  <version>5.9.0</version>
    		</dependency>
            <dependency>  
                <groupId>org.springframework</groupId>  
                <artifactId>spring-jms</artifactId>  
                <version>${spring.version}</version>  
            </dependency>

其次，spring配置文件中增加如下内容：  
至于为什么增加这些内容，不再详述，大家可以参考http://blog.csdn.net/lifetragedy/article/details/51836557。

<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
            <property name="brokerURL" value="${broker_url}" />
        </bean>
        
        <bean id="connectionFactory"
            class="org.springframework.jms.connection.CachingConnectionFactory">
            
            <property name="targetConnectionFactory" ref="targetConnectionFactory" />
        </bean>
        
        <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
            
            <property name="connectionFactory" ref="connectionFactory" />
        </bean>
        
        <bean id="queueDestination" name="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
            <constructor-arg>
                <value>queue</value>
            </constructor-arg>
        </bean>
        
        
        <bean id="topicDestination" name="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
            <constructor-arg value="topic"/>
        </bean>

再次，书写service层代码。需要注意的是在下面的示例中，没有采用上面配置文件中注入的topicDestination。原因是上面的topicDestination配置的参数值是topic，是固定的。当系统群发消息的话可以直接使用topicDestination。但项目中特定的通知只想发给特定的人员，因此为不同的人员创建了不同的topic，从而保证消息不会被不该看到的人看到。

@Service
    public class NotificationProducerServiceImpl implements NotificationProducerService {
    
    	@Resource
    	private JmsTemplate jmsTemplate;
    	private static final Logger log = LoggerFactory.getLogger(NotificationProducerServiceImpl.class);
    
    
    	public void sendMessage(final NotificationVo notification) {
    		Destination desc;
    		Connection connection = null;
    		Session session = null;
    		try {
    			ConnectionFactory connectionFactory = jmsTemplate.getConnectionFactory();
    			connection = connectionFactory.createConnection();
    			connection.start();
    			session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
    			desc = session.createTopic(notification.getNoticeTo());
    
    
    			jmsTemplate.send(desc, new MessageCreator() {
    				public Message createMessage(Session session) throws JMSException {
    					return session.createTextMessage(notification.getNoticeTime() + "|" + notification.getTitle() + "|"
    							+ notification.getMessage() + "|" + notification.getConId() + "|" + notification.getId() + "|" + notification.getNoticeTo());
    				}
    			});
    			session.commit();
    		} catch (JMSException e) {
    			log.error("推送消息发送失败！", e);
    		} finally {
    			try {
    				if (session != null) {
    					session.close();
    				}
    				if (connection != null) {
    					connection.close();
    				}
    			} catch (JMSException e) {
    				log.error("推送消息发送后，连接关闭失败！");
    			}
    		}
    	}
    }

第三，消息消费者环境的配置和编码。（即android端消息的获取和展示）  
Android端直接用原生的MQTT来做推送的比较少，而eclipse paho这个封装好的API似乎比较好用在Android端的推送上，于是就采用这个包来做。  
将逻辑写在Service可以使程序在后台执行时也收到推送。但是有个问题要处理，就是离线的消息不能丢失，所以要使消息持久化。而ActiveMQ实现了持久化订阅者的操作。在上代码前，先对两个概念：持久化传输和持久化订阅来进行介绍。  
ActiveMQ持久化传输：  
ActiveMQ支持两种传输模式：持久传输和非持久传输(persistent and non-persistent delivery)。可以通过MessageProducer类的setDeliveryMode方法设置传输模式。  
持久传输和非持久传输最大的区别是：采用持久传输时，传输的消息会保存到磁盘中(messages are persisted to disk/database)，即“存储转发”方式。先把消息存储到磁盘中，然后再将消息“转发”给订阅者。  
•采用非持久传输时，发送的消息不会存储到磁盘中。  
•采用持久传输时，当Borker宕机恢复后，消息还在。采用非持久传输，Borker宕机重启后，消息丢失。  
ActiveMQ默认的传输模式是持久传输。对于我们的项目而言，为了保证消息不丢失，我们项目中也要采用持久传输。  
在activemq.xml中有如下配置，这说明ActiveMQ使用KAHADB作为其持久化存储。KAHADB速度没有AMQ快，可是KAHADB具有极强的垂直和横向扩展能力，恢复时间比AMQ还要短，而且在作MQ的集群时使用KAHADB可以做到Cluster+Master Slave的这样的完美高可用集群方案。  
<persistenceAdapter>    
           <kahaDB directory="$\{activemq.data\}/kahadb"/>    
</persistenceAdapter>   
接下来介绍持久订阅者和非持久订阅者。持久订阅者和非持久订阅者针对的是订阅发布模式而不是点对点模式。当Broker发送消息给订阅者时，如果订阅者处于inactive状态：持久订阅者可以收到消息，原理为：持久订阅时，客户端向JMS 服务器注册一个自己身份的ID，当这个客户端处于离线时，JMS Provider 会为这个ID保存所有发送到主题的消息，当客户再次连接到JMS Provider时，会根据自己的ID得到所有当自己处于离线时发送到主题的消息。而非持久订阅者则收不到消息。  
持久订阅者/非持久订阅者，只影响离线的时候消息(包括持久消息和非持久消息)是否能接收到，和消息是否持久无关；持久消息/非持久消息，只是影响jmsprovider宕机后。消息是否会丢失，如果永远不会宕机，那么持久消息和非持久消息没有区别。  
基于paho包的实现持是否是持久化订阅者和两个地方有关系，首先是连接时连接选项也就是MqttConnectOptions有关，该对象可以设置心跳时间、超时时间等，但和是否是持久化订阅者有关的则是setCleanSession这个成员方法，它接收的参数是boolean，参数为true表示清除缓存，也就是非持久化订阅者，这个时候只要参数设为true，一定是非持久化订阅者。而参数设为false时，表示服务器保留客户端的连接记录，但此时也不一定是持久化订阅者，这个时候第二个相关的就出场了，就是订阅时的服务质量，也就是qos。服务质量的概念大致如下：  
最多一次（0） 最少一次（1）只一次（2）  
因此，调用连接选项的成员方法setCleanSession，设置为false，然后订阅时参数中的服务质量设置为1或2，即可完成持久化订阅。  
但是我在使用的时候，发现服务质量设置(Qos)，最主要的是下面两个设置。

MqttConnectOptions options = new MqttConnectOptions();
    options.setCleanSession(false);
    //因为是消息消费者，不是消息生产者，不往ActiveMQ发送消息，所以new byte[0]就可以了。
    //第二个参数设成false非常重要，否则会源源不断地收到重复的消息。
    options.setWill(topic, new byte[0], 0, false);
    mqttClient.connect(options);
    //topic要与消息生产者的订阅主题要保持一致，才能收到消息
    mqttClient.subscribe(topic);

持久订阅者和非持久订阅者的确认方法可以参考如下网址。  
http://blog.csdn.net/wtrnash/article/details/72588744  
关键的代码如下：  
首先，书写Service，用于监听推送消息。

String ANDROID_ID = Settings.System.getString(getContentResolver(), Settings.System.ANDROID_ID);
    	   String topic = "topic";
               mqttClient = new MqttClient(ElephantmonitorHttpClient.BROKER_URL, ANDROID_ID, new MemoryPersistence());
    
    
                mqttClient.setCallback(new PushCallback(this));//PushCallback是收到推送消息的回调类
                MqttConnectOptions options = new MqttConnectOptions();
                options.setAutomaticReconnect(true);
                options.setCleanSession(false);
                options.setConnectionTimeout(10);
                options.setKeepAliveInterval(20);
                //设定（持久订阅，不重复获取推送消息）
                options.setWill(topic, new byte[0], 0, false);
                mqttClient.connect(options);
    
    
                //Subscribe to all subtopics of homeautomation
                mqttClient.subscribe(topic);

其次，收到消息后回调PushCallback实现消息的展示。PushCallback要实现MqttCallback接口，关键代码如下：

@Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {
            if (StringUtils.isEmpty(message.toString())) {
                return;
            }
            //解析推送信息(0:时间，1：标题，2：内容，3：申请ID，4：消息ID，5:推送对象)
            String[] strMessageArr = message.toString().split("\\|");//消息生产者对内容用|进行了分割。
            NotificationVo vo = new NotificationVo();
            vo.setNoticeTime(strMessageArr[0]);
            vo.setTitle(strMessageArr[1]);
            vo.setMessage(strMessageArr[2]);
            vo.setConId(strMessageArr[3]);
            vo.setId(strMessageArr[4]);
            vo.setNoticeTo(strMessageArr[5]);
    
    
            //保存消息到本地，并且设为未读[用SQLiteDatabase进行的存储，非常简单就不上代码了]
            saveNoticeToLocal(vo);
            //发送广播，更改主页上的未读消息计数
            sendBroadCast();
    
    
            final NotificationManager notificationManager = (NotificationManager)
                    context.getSystemService(Context.NOTIFICATION_SERVICE);
            //创建pendingIntent对象
            PendingIntent pendingIntent = setPendingIntent(vo);
            Notification.Builder builder = new Notification.Builder(context);
            builder.setContentTitle(vo.getTitle())
                    .setContentText(vo.getMessage())
    //                .setTicker("摘要")//通知显示的内容
                    .setSmallIcon(R.drawable.logo)//设置小图标，4.x在右边，5.x在左边
    //                .setLargeIcon(bitmap)//设置大图标
                    .setDefaults(Notification.DEFAULT_LIGHTS | Notification.DEFAULT_VIBRATE | Notification.DEFAULT_SOUND)//设置默认的呼吸灯+震动+声音
                    .setContentIntent(pendingIntent)//打开通知后做什么
                    .setAutoCancel(true);//点击后自动消失
    
    
            Notification notification = builder.build();
            notification.flags |= Notification.FLAG_ONLY_ALERT_ONCE;//Notification.FLAG_INSISTENT;//不查看通知就一直发出声音和震动
            //发布通知
            if (notificationManager != null) {
                notificationManager.notify(Integer.parseInt(vo.getNoticeTime().substring(4,14)), notification);//用通知生成的时间做ID
            }
        }
       /**
         * 消息接收后的处理
         * @param vo 接收到的推送信息
         * @return 处理结果
         */
        private PendingIntent setPendingIntent(NotificationVo vo) {
    
    
            Intent intent = new Intent(context, MainActivity.class);
            intent.setFlags(Intent.FLAG_ACTIVITY_NEW_TASK);
            return PendingIntent.getActivity(context, 1, intent, 0);
        }
       /**
         * 发送广播，更新主页上的未读消息计数
         */
        private void sendBroadCast() {
            SQLiteDatabase dbInfo =context.openOrCreateDatabase("hmp.db", Context.MODE_PRIVATE, null);
            String sql = "select count(*) as c from d_notification where read_status ='0' ";
            Cursor cursor = dbInfo.rawQuery(sql, null);
            int count = 0;
            if(cursor.moveToNext()) {
                count = cursor.getInt(0);
            }
            Intent intent = new Intent();
            intent.setAction("REFRESH_NOTICE_COUNT");
            intent.putExtra("unReadCnt", count + StringUtil.EMPTY);
            context.sendBroadcast(intent);
        }

再次，在APP的主Activity(MainActivity)中,启动服务。

//启动消息通知服务
        Intent intent = new Intent(MainActivity.this, MQTTService.class);
        startService(intent);

最后，MainActivity中增加广播的监听，用于显示未读消息的件数。

private NoticeBroadcastReceiver mBroadcastReceiver = null;
        @Override
        public void onResume() {
            super.onResume();
    
    
            //实例化BroadcastReceiver子类 &  IntentFilter
            mBroadcastReceiver = new NoticeBroadcastReceiver();
            IntentFilter intentFilter = new IntentFilter();
    
    
            //设置接收广播的类型
            intentFilter.addAction("REFRESH_NOTICE_COUNT");
    
    
            //调用Context的registerReceiver（）方法进行动态注册
            getActivity().registerReceiver(mBroadcastReceiver, intentFilter);
        }
    
    
        @Override
        public void onPause() {
            super.onPause();
    
    
            //注销广播
            getActivity().unregisterReceiver(mBroadcastReceiver);
        }
    
    
       /**
         * 内部类
         */
        public class NoticeBroadcastReceiver extends BroadcastReceiver {
    
    
            //接收到广播后自动调用该方法
            @Override
            public void onReceive(Context context, Intent intent) {
                //写入接收广播后的操作
                String cnt = intent.getStringExtra("unReadCnt");
                setNoticeUnreadCnt(cnt);
            }
        }
    
    
        
        /**
         * 设置未读消息数的显示，BadgeView使得在某TextView的右上角以红点+数字的方式展示成为可能
         * @param count 未读消息数
         */
        public void setNoticeUnreadCnt(String count) {
            /*消息*/
            final TextView tv = (TextView)view.findViewById(R.id.message);
    	BadgeView badge = new BadgeView(getActivity(), tv);
            if (Integer.parseInt(count) > 0) {
                badge.setText(count);
                badge.show();
                badge.setTextSize(getResources().getDimensionPixelSize(R.dimen.font_size_navbar));
                badge.setBadgePosition(BadgeView.POSITION_TOP_RIGHT);
            } else {
                badge.setText(StringUtil.EMPTY);
                badge.hide();
            }
        }