【Java编程系列】comet4j服务器推送实现

r囧r小猫 2022-06-08 01:51 146阅读 0赞

# 热门系列： #

*  ### [【Java编程系列】WebService的使用][Java_WebService] ###
 *  ### [【Java编程系列】在Spring MVC中使用工具类调用Service层时，Service类为null如何解决][Java_Spring MVC_Service_Service_null] ###
 *  ### [【Java编程系列】Spring中使用代码实现动态切换主从库（多数据源）][Java_Spring] ###
 *  ### [【Java编程系列】log4j配置日志按级别分别生成日志文件][Java_log4j] ###
 *  ### [【Java编程系列】使用Java进行串口SerialPort通讯][Java_Java_SerialPort] ###
 *  ### [【Java编程系列】使用JavaMail通过SMTP协议发送局域网(内网)邮件][Java_JavaMail_SMTP] ###
 *  ### [【Java编程系列】解决Java获取前端URL中加号（+）被转换成空格的问题][Java_Java_URL] ###
 *  ### [【Java编程系列】使用List集合对百万数据量高效快速过滤去重筛选][Java_List] ###
 *  ### [【Java编程系列】Java与Mysql数据类型对应表][Java_Java_Mysql] ###
 *  ### [【Java编程系列】Java自定义标签-Tag][Java_Java_-Tag] ###
 *  ### [【Java编程系列】二进制如何表示小数？0.3+0.6为什么不等于0.9？纳尼！！！][Java_0.3_0.6_0.9] ###
 *  ###   [程序人生，精彩抢先看][Link 1] ###

--------------------

# 1.前言 #

最近在项目中有一个需求，简单来说就在用户的页面中，在不刷新不操作的情况下，当数据库有新数据是发给客户端页面的时候，会自动刷新页面的消息列表并显示出来（并非刷新页面，局部刷新）。所以初步分析，按我们以前的方式可能是通过定时器，然后不断调用ajax去请求后台，然后将新数据返回到页面更新。但一个项目，肯定会有很多客户使用，考虑到多个客户端不断请求刷新，这对服务器以及系统自身都是很大的考验，所以经过一番搜索，在网上了解到了comet4j服务器推送技术和websocket，GoEasy等等。

但今天要跟大家分享的是Comet4j这种服务器推送技术，当然我所要说的只是一些实际使用的技术心得，不是原理类阐述（有需要可以度娘一波）。

用comet4j之前，需要改tomcat的server.xml文件，下载导入comet4j-tomcat7.jar和comet4j.js等相关包资源。具体操作以及需要注意的地方，可参考我的上一篇博客:

**[【一天一坑系列】使用Comet4j实现服务器推送实现时的一系列问题][Comet4j]**

; 具体操作可参考以下链接博客:http://blog.csdn.net/ntotl/article/details/51803641;

--------------------

# 2.正题 #

那么现在进入正题，谈一下我所做项目中的使用情况。

根据上面的需求，我一开始的思路是通过每个客户端当前登录用户的id来做为推送数据的区分，因本人项目有多至十几个页面需要推送数据，所以我将每个用户需要推送的数据一次全部通过service层调用后推送出去。且因为是暂定每隔15秒就需要推送刷新一次，所以我一开始是想到用一个无限循环，每次让循环线程阻塞15秒再继续来实现：

while (true) {  
                    try {
                        //每15秒执行一次
                        Thread.sleep(15000);
                        //执行数据查询以及推送
                       ............
                    } catch (Exception ex) {  
                        ex.printStackTrace();  
                    }  
               		
                    
                }

然后在事件触发方法handleEvent（）中每次调用次线程。一开始该方式是可以实现我的页面数据定时刷新的（以为成功了，心里沾沾自喜了一把），但是有一个隐藏的问题，导致后面的大问题发生了，就是项目在运行几十分钟之后，页面就崩溃了。再刷新页面，请求就一直处于挂起pengding状态，在myeclipse控制台也没有报错，也没有出现任何打印内容。并且页面一直出去请求状态，无法显示（可以说只要程序没关闭，会一直转圈圈）。后来没办法，关闭了程序页面才出现请求失败提示。

一个方法，直接把程序搞跪了，心里烦到不行。但是烦解决不了问题，所以后来又打算去百度各种搜索，看有好的解决办法没。但基本都是推送的简单例子，并没有实质性的解决方法。心平气和，思考一阵之后，准备通过实验的方式，去测试问题所在。因为方法是在handleEvent（）中执行的，并且每次是通过以下代码：

public boolean handleEvent(ConnectEvent event) {
    		// TODO Auto-generated method stub
        	final CometConnection conn = event.getConn();
        	HttpServletRequest request = conn.getRequest();
    		HttpServletResponse response = conn.getResponse();
    		
    		//获取当前用户
    		User activeUser = ActiveUser.getInstance().getLoginUserOfSession(request, response);
    		int userid = activeUser.getUserid();
    
    		return true;
    	}

来运行操作。开始通过system的打印大法，来看一下每次请求发起时（在推送之前，即启动推送前），这个CometConnection conn = event.getConn();所建立的连接与推送时所使用的

CometEngine engine = CometContext.getInstance().getEngine();
    engine.getConnections();

获取连接之前关系；通过打印他们的连接id，发现每次请求都会产生新的连接id，并且观察不出什么规律性的联系。但是可以知道的是，每一次只有在jsp页面有建立连接的页面所发出的请求才回进入此handleEvent（）方法。

于是感觉有些麻烦了，因为每一个请求就会生成一个新的连接通道，而又有那么多的用户需要区别，一时之间感觉GG了。办法总比困难多，后来借鉴上面我所推荐的第二个链接地址中的做法。我想到的是将每一个用户的每一次请求所产生的连接id和当前用户id通过缓存类Cache.java

package com.cams.Comet4jUtil;
    
    import org.comet4j.core.event.ConnectEvent;
    
    public class Cache {
    	
    	private ConnectEvent key;
    	
    	private Object value;
    	//超时时限
    	private long timeOut;
    	//是否过期
    	private boolean isExpired;
    	
    	
    	public ConnectEvent getKey() {
    		return key;
    	}
    	public void setKey(ConnectEvent key) {
    		this.key = key;
    	}
    	public Object getValue() {
    		return value;
    	}
    	public void setValue(Object value) {
    		this.value = value;
    	}
    	public long getTimeOut() {
    		return timeOut;
    	}
    	public void setTimeOut(long timeOut) {
    		this.timeOut = timeOut;
    	}
    	public boolean isExpired() {
    		return isExpired;
    	}
    	public void setExpired(boolean isExpired) {
    		this.isExpired = isExpired;
    	}
    	
    }

以及缓存操作类CacheManager.java

package com.cams.Comet4jUtil;
    
    import java.util.Date;
    import java.util.HashMap;
    
    import org.comet4j.core.event.ConnectEvent;
    
    public class CacheManager {
    	
    	public static HashMap<ConnectEvent,Object> cacheMap = new HashMap<ConnectEvent,Object>(); 
        
        /**
         * This class is singleton so private constructor is used.
         */ 
        private CacheManager()
        { 
                super(); 
        } 
    
        /**
         * 获取缓存
         * returns cache item from hashmap
         * @param key
         * @return Cache
         */ 
        public synchronized static Cache getCache(ConnectEvent key)
        { 
                return (Cache)cacheMap.get(key); 
        } 
    
        /**
         * 缓存是否存在
         * Looks at the hashmap if a cache item exists or not
         * @param key
         * @return Cache
         */ 
        private synchronized static boolean hasCache(ConnectEvent key)
        { 
                return cacheMap.containsKey(key); 
        } 
    
        /**
         * 清除所有缓存
         * Invalidates all cache
         */ 
        public synchronized static void invalidateAll()
        { 
                cacheMap.clear(); 
        } 
    
        /**
         * 清除指定缓存
         * Invalidates a single cache item
         * @param key
         */ 
        public synchronized static void invalidate(ConnectEvent key)
        { 
                cacheMap.remove(key); 
        } 
    
        /**
         * 存入缓存
         * Adds new item to cache hashmap
         * @param key
         * @return Cache
         */ 
        private synchronized static void putCache(ConnectEvent key, Cache object)
        { 
           cacheMap.put(key, object); 
        } 
    
        
        /**
         * 获取缓存
         * Reads a cache item's content
         * @param key
         * @return
         */ 
        public static Cache getContent(ConnectEvent key)
        { 
                 if (hasCache(key)) { 
                        Cache cache = getCache(key); 
                        if (cacheExpired(cache)) { 
                                cache.setExpired(true); 
                        } 
                        return cache; 
                 } else { 
                         return null; 
                 } 
        } 
    
        /**
         * 存放设置时效的缓存
         * @param key
         * @param content
         * @param ttl
         */ 
        public static void putContent(ConnectEvent key, Object content, long ttl)
        { 
                Cache cache = new Cache(); 
                cache.setKey(key); 
                cache.setValue(content); 
                cache.setTimeOut(ttl + new Date().getTime()); 
                cache.setExpired(false); 
                putCache(key, cache); 
        } 
        
        public static void putContent(ConnectEvent key, Object content)
        { 
            Cache cache = new Cache(); 
            cache.setKey(key); 
            cache.setValue(content); 
            cache.setExpired(false); 
            putCache(key, cache); 
        } 
         
        /**
         * 判断缓存是否失效（过期）
         */ 
        public static boolean cacheExpired(Cache cache){ 
                if (cache == null) { 
                        return false; 
                } 
                long milisNow = new Date().getTime(); 
                long milisExpire = cache.getTimeOut(); 
                if (milisExpire < 0) {                // Cache never expires  
                        return false; 
                } else if (milisNow >= milisExpire) { 
                        return true; 
                } else { 
                        return false; 
                } 
        }
    }

来管理控制。简而言之，以每次的请求所产生的事件event（即参数ConnectEvent event）来做为key，以当前操作用户的id为value，设置时效，在每次的handleEvent中将其保存。然后在一个执行推送的线程中取循环这个CacheManager的存储cacheMap，在每一次的循环中推送数据。

class SendToClientThread implements Runnable {
        	private int userid;
        	private HttpSession session;
        	
            public void run() {
        		
                while (true) {  
                    try {
                    	//每15秒执行一次
                        Thread.sleep(15000);
                        
                        //因为迭代时不允许集合中的元素发生位置改变（即删除，添加等），所以创建一个list集合暂存要删除的元素
                        List<ConnectEvent> tempList = new ArrayList<>();
                        for (ConnectEvent  cEvent : CacheManager.cacheMap.keySet())
                        {
                     	   //判断缓存是否存在
                     	   if(CacheManager.getContent(cEvent) != null)
                     	   {
                     		   Cache cache = CacheManager.getContent(cEvent);
                     		   if(cache.isExpired())
                     		   {
                     			   //过期则删除该缓存
    //                 			   CacheManager.invalidate(cache.getKey());
                     			  tempList.add(cache.getKey());
                     		   }
                     		   else
                     		   {
                     			   //没有过期则推送
                     			   // CometEngine ： 引擎，负责管理和维持连接，并能够必要的发送服务  
                                   CometEngine engine = CometContext.getInstance().getEngine();
                                   String cacheValue = cache.getValue() == null ? "0" : cache.getValue().toString();
                     			   userid = Integer.parseInt(cacheValue);
                     			   final CometConnection cc = cEvent.getConn();
                     			   String preconnid = cc.getId();
                     			   session = cc.getRequest().getSession();
                     			   
                     			   if(session != null)
                     			   {
                     				   //获取各频道所需参数
            	                   		personInboxModel = (PersonMsgSearchModel)session.getAttribute("PersonInboxPar") == null ? new PersonMsgSearchModel() : (PersonMsgSearchModel)session.getAttribute("PersonInboxPar");
            	                   		
                               		
                               		   //获取 发送个人消息页面消息列表
                                       personMsgList = messageService.getMessages(userid, 1);
                                       //获取 个人收件箱列表
                                      
                                       
                                       //推送到指定用户和指定频道
                                       engine.sendTo(CHANNEL1,engine.getConnection(preconnid),JSON.toJSONString(personMsgList));
                                       
                     			   }
                     			   else
                     			   {
                     				  tempList.add(cache.getKey());
                     			   }
                     			   
                     		   }
                     	   }
                        }
                        //删除无效事件（即session为空的）
                        if(tempList.size() > 0){
    	                    for (int i = 0; i < tempList.size(); i++) {
    							CacheManager.invalidate(tempList.get(i));
    						}
                        }
                    } catch (Exception ex) {  
                        ex.printStackTrace();  
                    }  
               		
                    if(flag)
                    {
                    	//当flag=true,标识程序已终止
                    	break;
                    }
                    
                }  
            }  
        }

果不其然，这次目前算是解决问题了。（以上代码只粘贴部分，若有缺失不必见怪）

--------------------

# 3.总结 #

问题解决了，但因为项目暂时还没有正式投入使用，后期不确定还会不会出现新的问题。若有新问题产生，将后续编辑更新。希望能帮到有同样需求的朋友，也欢迎留言交流！！

> ### 本博客皆为学习、分享、探讨为本，欢迎各位朋友评论、点赞、收藏、关注，一起加油！ ###

![20201029165124734.gif][]

[Java_WebService]: https://blog.csdn.net/yy339452689/article/details/68061339
[Java_Spring MVC_Service_Service_null]: https://blog.csdn.net/yy339452689/article/details/73291777
[Java_Spring]: https://blog.csdn.net/yy339452689/article/details/76900650
[Java_log4j]: https://blog.csdn.net/yy339452689/article/details/77571360
[Java_Java_SerialPort]: https://blog.csdn.net/yy339452689/article/details/77772421
[Java_JavaMail_SMTP]: https://blog.csdn.net/yy339452689/article/details/78062727
[Java_Java_URL]: https://blog.csdn.net/yy339452689/article/details/80662094
[Java_List]: https://blog.csdn.net/yy339452689/article/details/105657691
[Java_Java_Mysql]: https://blog.csdn.net/yy339452689/article/details/106199652
[Java_Java_-Tag]: https://blog.csdn.net/yy339452689/article/details/78520692
[Java_0.3_0.6_0.9]: https://blog.csdn.net/yy339452689/article/details/108256478
[Link 1]: https://blog.csdn.net/yy339452689/article/details/109314083
[Comet4j]: https://blog.csdn.net/yy339452689/article/details/77850544
[20201029165124734.gif]: /images/20220608/bcf92e8e61864c25a4425b04a0084f2d.png