守护线程保证主线程不断掉

àì夳堔傛蜴生んèń 2022-05-27 01:20 197阅读 0赞

1.主线程

package com.project.cache;
    
    import java.io.InputStream;
    import java.net.MalformedURLException;
    import java.net.URL;
    import java.util.Date;
    import java.util.List;
    import java.util.concurrent.Callable;
    import java.util.concurrent.FutureTask;
    
    import javax.annotation.PostConstruct;
    
    import org.apache.commons.lang3.StringUtils;
    import org.apache.log4j.Logger;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Configuration;
    
    import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.JSONObject;
    import com.project.entity.UserEntity;
    import com.project.entity.YeZhuEntity;
    import com.project.service.DaiLiEntityService;
    import com.project.service.LogEntityService;
    import com.project.service.UserEntityService;
    import com.project.service.YeZhuEntityService;
    import com.project.utils.HTTPUtils;
    
    @Configuration
    public class LieYData{
    	
    	private Logger logger = Logger.getLogger(LieYData.class);
    
    	@Autowired
    	private YeZhuEntityService yeZhuEntityServiceImpl;
    	
    	@Autowired
    	private UserEntityService userEntityServiceImpl;
    	
    	@Autowired
    	private DaiLiEntityService daiLiEntityServiceImpl;
    	
    	@Autowired
    	public LogEntityService logEntityService;
    	
    	@PostConstruct
    	public void init(){
    		FutureTask<String> task = new FutureTask<String>(new Callable<String>(){  
    	         @Override  
    	         public String call() throws Exception {  
    	        	  loadData(); //使用另一个线程来执行该方法，会避免占用Tomcat的启动时间  
    	              return "init ok";  
    	         }  
    	    });  
    	    Thread thread = new Thread(task);
    	    thread.start();
    	    //启动守护线程，监听该线程是否断掉
    	    Thread guardThread = new Thread(new GuardThread(thread));
    	    guardThread.setDaemon(true);
    	    guardThread.start();
    	}
    	
    	public void loadData() {
    		try {  
                //确定服务地址 
    			String token = getToken();
    			String urls = "http://111.11.111.11:8100/stream?cname=0a40ac1547044518b6cdd0b3ee1a9303&token=" + token;
                URL url = new URL(urls);  
                InputStream in=url.openStream();  
                int n = -1;  
                byte[] b = new byte[1024];  
                //从服务端读取数据并打印  
                StringBuffer sb = new StringBuffer();
                Date startDate = new Date();
                while((n=in.read(b))!=-1)
                {  
                	String s=new String(b,0,n, "UTF-8"); 
                	Date endDate = new Date();
                	if(endDate.getTime() - startDate.getTime() < 25*1000){
                		sb.append(s);
                	}else{
                		startDate = endDate;
                		logger.info("LieYData:"+sb.toString());
                		todo(sb.toString());
                		sb.setLength(0);
                		sb.append(s);
                	}
                }           
            } catch (Exception e) {  
                e.printStackTrace();
            }  
    	}
    	
    	private void todo(String datas) {
    		String[] datass = datas.split("\n");
    		for (String data : datass) {
    			JSONObject js = JSON.parseObject(data);
    			String contentStr = js.getString("content");
    
    			logger.info("猎鹰线程名" + Thread.currentThread().getName());
    			
    			if(StringUtils.isBlank(contentStr)){
    				continue;
    			}
    			JSONObject content = JSON.parseObject(contentStr);
    			if(StringUtils.equals(content.getString("method"), "addUser")){
    				//h5注册用户1
    				UserEntity record = content.getObject("data", UserEntity.class);
    				int ss = userEntityServiceImpl.addUserEntity(record);
    				log("000", "000", "000", 6, "addUser", "000",contentStr, ss > 0?true:false);
    			}else if(StringUtils.equals(content.getString("method"), "updateUser")){
    				//h5注册用户2
    				UserEntity record = content.getObject("data", UserEntity.class);
    				int ss = userEntityServiceImpl.updateUserEntityByKeyM(record);
    				log("000", "000", "000", 6, "updateUser", "000",contentStr, ss > 0?true:false);
    			}else if(StringUtils.equals(content.getString("method"), "addYeZhu")){
    				//h5添加业主
    				YeZhuEntity record = content.getObject("data", YeZhuEntity.class);
    				String userUuid = content.getString("userUuid");
    				int ss = yeZhuEntityServiceImpl.addYeZhuEntityM(record, userUuid);
    				log("000", "000", "000", 6, "addYeZhu", "000",contentStr, ss > 0?true:false);
    			}else if(StringUtils.equals(content.getString("method"), "updateYeZhu")){
    				//h5修改业主
    				YeZhuEntity record = content.getObject("data", YeZhuEntity.class);
    				String userUuid = content.getString("userUuid");
    				int ss = yeZhuEntityServiceImpl.updateYeZhuEntityByKeyM(record, userUuid);
    				log("000", "000", "000", 6, "updateYeZhu", "000",contentStr, ss > 0?true:false);
    			}else if(StringUtils.equals(content.getString("method"), "resetYzcode") || StringUtils.equals(content.getString("method"), "login") ){
    				//h5重置业主code
    				@SuppressWarnings("unchecked")
    				List<String> uuidIds = content.getObject("data", List.class);
    				String yezhucode = content.getString("yezhucode");
    				String zukecode = content.getString("zukecode");
    				boolean flag = yeZhuEntityServiceImpl.resetYzcodeM(uuidIds, yezhucode, zukecode);
    				log("000", "000", "000", 6, "resetYzcode", "000",contentStr, flag);
    			}
    		}
    	}
    	
    	//得到token
    	private String getToken() throws MalformedURLException{
    		String url = "http://111.11.11.11:11101/MicroRecon/1.2/datajoin_manage?action=get_token&vender_id=0a40ac1547044518b6cdd0b3ee1a9303&data_type=cbf1cfa898094c0ab9ff42c74318441d";
    		String ss = HTTPUtils.post(new URL(url),  new JSONObject().toJSONString());
    		if(StringUtils.isNotBlank(ss)){
    			JSONObject js = JSON.parseObject(ss);
    			JSONObject jsonResponse = js.getJSONObject("response");
    			return jsonResponse.getString("token");
    		}
    		return null;
    	}
    	
    	//日志
    	private void log(String ip , String userId,String realName,int type,String moduleId,String moduleName,String description,boolean result){
    		logEntityService.log(ip, userId, realName, type, moduleId, moduleName, description, result);
    	}
    }

2.守护线程

package com.project.cache;
    
    import java.util.concurrent.Callable;
    import java.util.concurrent.FutureTask;
    
    import org.apache.log4j.Logger;
    
    /*
     * 守护线程，监听从猎鹰平台读取数据的线程
     */
    public class GuardThread implements Runnable{
    
    	private Logger logger = Logger.getLogger(LieYData.class);
    	
    	private Thread thread;
    	
    	public GuardThread(Thread thread){
    		this.thread = thread;
    	}
    	
    	@Override
    	public void run() {
    		while (true) {
                monitor();
                try {
                    Thread.sleep(5*1000);
                    logger.info("守护线程："+Thread.currentThread().getName());
                } catch (InterruptedException e) {
                	logger.error("守护线程异常");
                    e.printStackTrace();
                }
            }
    	}
    	
    	/**
         * 监控主要逻辑
         */
        private void monitor() {
            Thread.State state = thread.getState();
            //如果被监控线程为终止状态,则重启被监控的线程
            if (Thread.State.TERMINATED.equals(state)) {
               logger.info(thread.getName() + "被监控线程已经终止,现在开始重启被监控的线程!");
               
                FutureTask<String> task = new FutureTask<String>(new Callable<String>(){  
    	   	         @Override  
    	   	         public String call() throws Exception {  
    	   	        	  new LieYData().loadData(); 
    	   	              return "init ok";  
    	   	         }  
    	   	    });  
    	   	    Thread thread = new Thread(task);
    	   	    thread.start();
    	   	    this.thread = thread;
            }
        }
    }