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;
        }
    }
}