环信聊天记录保存到数据库实体

摘要：发现环信的根据时间条件拉取历史消息接口已经停用就做了个通过导出聊天记录接口保存到数据库实体的功能分享一下大致思路通过环信的接口把前一天小时的数据压缩包下载到本地把下载后的文件解压读取处理写入到实体设置一个定时器定时执行通过环信接口拉取数据并

发现环信的根据时间条件拉取历史消息接口已经停用,就做了个通过导出聊天记录接口保存到数据库实体的功能,分享一下.

大致思路:

1.通过环信的接口,把前一天24小时的数据压缩包下载到本地
2.把下载后的文件解压读取处理,写入到实体
3.设置一个定时器,定时执行.

1.通过环信接口拉取数据,并解压读取
环信聊天记录下载接口

@Service
public class EaseMobService implements IEaseMobService{

    @Autowired
    private IChatMessageService chatMessageService;
    @Override
    public void saveChatMessages() {
        //下载文件保存路径
        String filePath = "/opt/apache/chatFiles/";
        //未加时间戳的请求地址
        //OrgInfo.ORG_NAME  环信org_name  OrgInfo.APP_NAME 环信app_name
        String requestUrl = "http://a1.easemob.com/"+ OrgInfo.ORG_NAME + "/" + OrgInfo.APP_NAME + "/chatmessages/";
        //获取前一天内的时间list
        List hourList = DateUtils.getOneDayHourList(DateUtils.getBeforeDayDate(new Date(), 1));
        //环信token 自己写一个工具类获取token
        String token = TokenUtil.getAccessToken();
        //获取下载地址
        for(String hour: hourList){
            try {
                String downloadUrl = HttpUtil.getEasemobChatMessageDownloadUrl(requestUrl + hour, token);
                if(!"fail".equals(downloadUrl)){
                    //下载压缩文件到指定文件夹
                    String fileName = hour + ".gz";
                    String downLoadResult = HttpUtil.downloadFileByUrls(downloadUrl, fileName,filePath);
                    //下载成功进行解压文件和读取文件
                    if("ok".equals(downLoadResult)){
                        //解压文件
                        String outPutFilePath = unZipFile(filePath + fileName);
                        //读取文件
                        if(outPutFilePath.length() >0) {
                            String content = readFile2String(outPutFilePath);
                            //处理文本内容,写入实体
                            if(content.length() > 0) {
                                chatMessageService.handleContent(content);
                            }
                        }
                    }
                }
                //延时执行,环信下载接口有限流
                Thread.sleep(5000);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 读取文件内容
    **/
    private String readFile2String(String outPutFilePath) {
        String content = "";
        String encoding = "UTF-8";
        File file = new File(outPutFilePath);
        Long fileLength = file.length();
        byte[] fileContent = new byte[fileLength.intValue()];
        try {
            FileInputStream in = new FileInputStream(file);
            in.read(fileContent);
            in.close();
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            content = new String(fileContent, encoding).trim();
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        return content;
    }

    /**
     * 解压文件并返回解压后的文件
    **/
    private String unZipFile(String filePath) {
        //解压gz压缩包
        String ouPutFile = "";
        try {
            //建立gzip压缩文件输入流
            FileInputStream fIn = new FileInputStream(filePath);
            //建立gzip解压工作流
            GZIPInputStream gzIn = new GZIPInputStream(fIn);
            //建立解压文件输出流
            ouPutFile = filePath.substring(0,filePath.lastIndexOf("."));
            FileOutputStream fOut = new FileOutputStream(ouPutFile);
            int num;
            byte[] buf=new byte[1024];

            while ((num = gzIn.read(buf,0,buf.length)) != -1)
            {
                fOut.write(buf,0,num);
            }
            gzIn.close();
            fOut.close();
            fIn.close();
        } catch (Exception e){
            e.printStackTrace();
        }
        return ouPutFile;
    }
}

DateUtils工具类方法

 /**
     * 获取指定日期的一天小时集合yyyyMMddHH
    **/
    public static List getOneDayHourList(Date date){
        List hourList = new ArrayList();
        SimpleDateFormat fmt = new SimpleDateFormat("yyyyMMdd");
        String dateString = fmt.format(date);
        for(int i = 0; i < 24; i++) {
            String hour = String.valueOf(i);
            if(i < 10){
                hour = "0" + hour;
            }
            hourList.add(dateString + hour);
        }
        return hourList;
    }
/**
     * 获取指定日期的前N天日期
    **/
public static Date getBeforeDayDate(Date date, int beforeDay)
    {
        Calendar a = Calendar.getInstance();
        a.setTime(date);
        a.add(Calendar.DATE, -beforeDay);
        return a.getTime();
    }

HttpUtil工具类

public class HttpUtil {

   private static Logger log = LoggerFactory.getLogger(HttpUtil.class);

    public static String getEasemobChatMessageDownloadUrl(String getUrl, String token) {
        String downloadUrl = "";
        try {
            URL url = new URL(getUrl);    //把字符串转换为URL请求地址
            HttpURLConnection connection = (HttpURLConnection) url.openConnection();// 打开连接
            //设置Head参数
            connection.setRequestProperty("Content-Type", " application/json");//设定 请求格式 json，也可以设定xml格式的
            connection.setRequestProperty("Accept-Charset", "utf-8");  //设置编码语言
            connection.setRequestProperty("Connection", "keep-alive");  //设置连接的状态
            connection.setRequestProperty("Authorization", token);
            connection.connect();// 连接会话
            // 获取输入流
            BufferedReader br = new BufferedReader(new InputStreamReader(connection.getInputStream(), "UTF-8"));
            String line;
            StringBuilder sb = new StringBuilder();
            while ((line = br.readLine()) != null) {// 循环读取流
                sb.append(line);
            }
            br.close();// 关闭流
            connection.disconnect();// 断开连接
            //返回结果处理
            JSONArray jsonArray = JSON.parseArray("[" + sb.toString() + "]");
            JSONObject jsonObject = (JSONObject) jsonArray.get(0);
            JSONArray urlJSON = JSON.parseArray(jsonObject.get("data").toString());
            downloadUrl = ((JSONObject) urlJSON.get(0)).get("url").toString();
        } catch (Exception e) {
            return "fail";
        }
        return downloadUrl;
    }

    /**
     * 通过url下载文件到本地
    **/
    public static String  downloadFileByUrls(String urlStr,String fileName,String savePath){
        try {
            URL url = new URL(urlStr);
            HttpURLConnection conn = (HttpURLConnection) url.openConnection();
            //设置超时间为3秒
            conn.setConnectTimeout(3 * 1000);
            //防止屏蔽程序抓取而返回403错误
            conn.setRequestProperty("User-Agent", "Mozilla/4.0 (compatible; MSIE 5.0; Windows NT; DigExt)");
            //得到输入流
            InputStream inputStream = conn.getInputStream();
            //获取自己数组
            byte[] getData = readInputStream(inputStream);
            //文件保存位置
            File saveDir = new File(savePath);
            if (!saveDir.exists()) {
                saveDir.mkdir();
            }
            File file = new File(saveDir + File.separator + fileName);
            FileOutputStream fos = new FileOutputStream(file);
            fos.write(getData);
            if (fos != null) {
                fos.close();
            }
            if (inputStream != null) {
                inputStream.close();
            }
            return "ok";
        }catch (Exception e){
            e.printStackTrace();
            return "fail";
        }
    }


    /**
     * 从输入流中获取字节数组
     */
    public static  byte[] readInputStream(InputStream inputStream) throws IOException {
        byte[] buffer = new byte[1024];
        int len = 0;
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        while((len = inputStream.read(buffer)) != -1) {
            bos.write(buffer, 0, len);
        }
        bos.close();
        return bos.toByteArray();
    }
}

2.数据库实体,及文本读取内容处理

chat_message表

SET FOREIGN_KEY_CHECKS=0;
-- ----------------------------
-- Table structure for chat_message
-- ----------------------------
DROP TABLE IF EXISTS `chat_message`;
CREATE TABLE `chat_message` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `msg_id` varchar(25) DEFAULT NULL,
  `timestamp` datetime DEFAULT NULL,
  `direction` varchar(50) DEFAULT NULL,
  `to_user` varchar(50) DEFAULT NULL,
  `from_user` varchar(50) DEFAULT NULL,
  `msg` varchar(255) DEFAULT NULL,
  `type` varchar(20) DEFAULT NULL,
  `url` varchar(100) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=11 DEFAULT CHARSET=utf8;

文本处理

/**
     * 处理环信返回的内容,写入实体
     *
     * @param content
     */
    @Override
    public void handleContent(String content) {
        JSONArray jsonArray = JSON.parseArray("[" + content + "]");
        List chatMessageList = new ArrayList();
        for(int i = 0; i < jsonArray.size(); i++){
            ChatMessage chatMessage = new ChatMessage();
            JSONObject jsonObject = (JSONObject) jsonArray.get(i);
            JSONArray bodyJsons = (JSONArray)((JSONObject) jsonObject.get("payload")).get("bodies");
            for(int j = 0; j < bodyJsons.size(); j ++) {
                JSONObject bodyJson = (JSONObject) bodyJsons.get(j);
                chatMessage.setMsgId(jsonObject.getString("msg_id"));
                chatMessage.setTimestamp(new Date(Long.parseLong(jsonObject.getString("timestamp"))));
                chatMessage.setDirection(jsonObject.getString("direction"));
                chatMessage.setToUser(jsonObject.getString("to"));
                chatMessage.setFromUser(jsonObject.getString("from"));
                chatMessage.setMsg(bodyJson.getString("msg"));
                chatMessage.setType(bodyJson.getString("type"));
                chatMessage.setUrl(bodyJson.getString("url"));
                chatMessageList.add(chatMessage);
            }
        }
        //批量插入到数据库
        getMapper().insertBatch(chatMessageList);
    }

用到了mybatis批量插入数据库,贴一下chatMessageMapper的这一段

    insert into chat_message (msg_id, timestamp, direction, to_user, from_user, msg, type, url)
    values
    
      (#{item.msgId,jdbcType=VARCHAR}, #{item.timestamp,jdbcType=TIMESTAMP},
      #{item.direction,jdbcType=VARCHAR},#{item.toUser,jdbcType=VARCHAR},#{item.fromUser,jdbcType=VARCHAR},
      #{item.msg,jdbcType=VARCHAR},#{item.type,jdbcType=VARCHAR},#{item.url,jdbcType=VARCHAR})
    
  

3.设置一个定时器定时执行service,还可以根据实际项目需求设置定时清理从环信下载的压缩包文件.

定时器

/**
 * 定时器实现
 *
 * @author Ray
 * @date 2018/1/27 10:35
 */
@Component
public class Timer implements ITimer{

    private static Logger log = LoggerFactory.getLogger(Timer.class);

    @Autowired
    IOrderService orderService;
    @Autowired
    ICouponService couponService;
    @Autowired
    IEaseMobService easeMobService;

    /*
1 Seconds (0-59)
2 Minutes (0-59)
3 Hours (0-23)
4 Day of month (1-31)
5 Month (1-12 or JAN-DEC)
6 Day of week (1-7 or SUN-SAT)
7 Year (1970-2099)
    取值：可以是单个值，如6；
    也可以是个范围，如9-12；
    也可以是个列表，如9,11,13
    也可以是任意取值，使用*
*/
    @Scheduled(cron = "0 0 12 * * ?")
    public void everyDay() {
        log.info("每日定时器执行");
        //1.检查订单自动收货
        orderService.checkReceiveConfirm();
        log.info("检查订单自动收货");
        //2.失效用户优惠券
        couponService.updateCouponUseStatusOnTime();
        log.info("失效用户优惠券");
        //3.保存前一天聊天记录
        easeMobService.saveChatMessages();
        log.info("保存前一天聊天记录");
    }
}