添加IM

patient-co-statistics/src/main/java/com/yihu/wlyy/statistics/job/message/HealthMessageJob.java

@@ -6,7 +6,9 @@ import com.yihu.wlyy.statistics.etl.storage.DBStorage;
import com.yihu.wlyy.statistics.model.job.QuartzJobLog;
import com.yihu.wlyy.statistics.model.signfamily.Message;
import com.yihu.wlyy.statistics.model.signfamily.SignFamily;
import com.yihu.wlyy.statistics.util.JsonUtil;
import com.yihu.wlyy.statistics.task.PushMsgTask;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
@@ -78,6 +80,7 @@ public class HealthMessageJob implements Job {
                    signFamilyMap.put(signFamilyTemp.getDoctor(),signFamilies);
                }
            }
            JSONArray jsonArray=new JSONArray();
            for(Map.Entry<String,List<SignFamily>> entry:signFamilyMap.entrySet()){
                 countSql="select count(id) from wlyy_sign_family " +
                        " where  status >0 and type='2' and expenses_status='1'  " +
@@ -92,18 +95,33 @@ public class HealthMessageJob implements Job {
                message.setSender("system");
                message.setCzrq(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(yesterday+" 12:00:00"));
                message.setState(1);
                message.setTitle("新增系统消息");
                SimpleDateFormat dateFormat=new SimpleDateFormat("MM月dd日");
                String date=  dateFormat.format(new Date());
                message.setContent(date+"新增"+entry.getValue().size()+"个签约居民待分配健管师，目前共"+allCount+"人待处理");
                String title="新增系统消息";
                message.setTitle(title);
                SimpleDateFormat dateFormat1=new SimpleDateFormat("yyyy-MM-dd");
                SimpleDateFormat dateFormat2=new SimpleDateFormat("MM月dd日");
                String content=dateFormat2.format(dateFormat1.parse(now))+"新增"+entry.getValue().size()+"个签约居民待分配健管师，目前共"+allCount+"人待处理";
                message.setContent(content);
                message.setCode(UUID.randomUUID().toString());
                message.setReceiver(entry.getKey());
                messageDao.save(message);
                // 异常通知
                JSONObject json = new JSONObject();
                json.put("receiver", entry.getKey());
                json.put("type", "D_NH_01");
                json.put("title", title);
                json.put("msg", content);
                json.put("data", "");
                jsonArray.add(json);
            }
            quartzJobLog.setJobEndTime(new Date());
            quartzJobLog.setJobContent("生成"+now+"的健康管理师消息成功");
            quartzJobLog.setJobType("1");
            dbStorage.saveLog(quartzJobLog);
            // 推送消息给医生
            PushMsgTask.getInstance().put(jsonArray);
        }catch (Exception e){
            e.printStackTrace();
        }

patient-co-statistics/src/main/java/com/yihu/wlyy/statistics/task/PushMsgTask.java

@@ -0,0 +1,173 @@
package com.yihu.wlyy.statistics.task;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.HttpURLConnection;
import java.net.URL;
import java.net.URLEncoder;
import java.util.concurrent.LinkedBlockingQueue;
public class PushMsgTask {
	private static Logger logger = LoggerFactory.getLogger(PushMsgTask.class);
	// 最大容量为50的数组堵塞队列
	private static LinkedBlockingQueue<JSONObject> queue = new LinkedBlockingQueue<JSONObject>();
	private static PushMsgTask instance;
	private static Object lock = new Object();
	@Value("${systemConfig.msg_push_server}")
	private String msgPushServer;
	public static PushMsgTask getInstance() {
		synchronized (lock) {
			if (instance == null) {
				instance = new PushMsgTask();
				instance.run();
			}
		}
		return instance;
	}
	/**
	 * 添加一条推送消息
	 * @param receiver 接收人
	 * @param type 消息类型
	 * @param title 消息标题
	 * @param msg 消息内容
	 * @param data 消息数据
	 */
	public void put(String receiver, String type, String title, String msg, String data) {
		try {
			JSONObject json = new JSONObject();
			json.put("receiver", receiver);
			json.put("type", type);
			json.put("title", title);
			json.put("msg", msg);
			json.put("data", data);
			queue.put(json);
		} catch (Exception e) {
			logger.error("添加到消息队列失败！", e);
			e.printStackTrace();
		}
	}
	public void put(JSONArray array) {
		if (array == null || array.size() == 0) {
			return;
		}
		for (int i = 0; i < array.size(); i++) {
			JSONObject json = array.getJSONObject(i);
			if (json == null) {
				continue;
			}
			try {
				queue.put(json);
			} catch (Exception e) {
				logger.error("批量添加到消息队列失败！", e);
			}
		}
	}
	private void run() {
		new Thread(new ConsumerTask()).start();
	}
	// 消费者
	class ConsumerTask implements Runnable {
		@Override
		public void run() {
			try {
				while (true) {
					// 如果queue为空，则当前线程会堵塞，直到有新数据加入
					JSONObject json = queue.take();
					// 推送平台消息
					String receiver = json.has("receiver") ? json.getString("receiver") : "";
					String type = json.has("type") ? json.getString("type") : "";
					String title = json.has("title") ? json.getString("title") : "";
					String msg = json.has("msg") ? json.getString("msg") : "";
					String data = json.has("data") ? json.getString("data") : "";
					boolean res = pushMessage(receiver, type, title, msg, data);
					if (res) {
						logger.info("消息推送成功！");
					} else {
						logger.error("消息推送失败！");
					}
				}
			} catch (Exception ex) {
				ex.printStackTrace();
			}
		}
	}
	/**
	 * 消息推送
	 *
	 * @param receiver 消息接收人
	 * @param msgType 消息类型
	 * @param title 消息标题
	 * @param msg 消息内容
	 * @param data 消息数据
	 */
	public static boolean pushMessage(String receiver, String msgType, String title, String msg, String data) {
		PrintWriter out = null;
		BufferedReader in = null;
		HttpURLConnection conn = null;
		try {
			//String server = getInstance().systemConfig.getMsgPushServer();
			String server = "http://127.0.0.1:3000/system/sendmsg.im";
			String url = server + "?to_uid=" + receiver + "&content=" + URLEncoder.encode(msg, "UTF-8") + "&type=" + msgType + "&title=" + URLEncoder.encode(title, "UTF-8") + "&data=" + data;
			URL realUrl = new URL(url);
			// 打开和URL之间的连接
			conn = (HttpURLConnection) realUrl.openConnection();
			conn.setRequestMethod("GET");
			conn.setDoOutput(true);
			conn.setDoInput(true);
			conn.setUseCaches(false);
			conn.setRequestProperty("Content-Type", "application/json");
			// 读取返回内容
			StringBuffer buffer = new StringBuffer();
			BufferedReader br = new BufferedReader(new InputStreamReader(conn.getInputStream(), "UTF-8"));
			String temp;
			while ((temp = br.readLine()) != null) {
				buffer.append(temp);
				buffer.append("\n");
			}
			System.out.println(buffer.toString());
			JSONObject json = new JSONObject().fromObject(buffer.toString());
			if (json.getInt("errno") == 0) {
				// 成功
				return true;
			} else {
				// 失败
				return false;
			}
		} catch (Exception e) {
			logger.error("push message error:", e);
		} finally {
			try {
				if (out != null) {
					out.close();
				}
				if (in != null) {
					in.close();
				}
			} catch (IOException ex) {
				ex.printStackTrace();
			}
		}
		return false;
	}
}