发现环信的根据时间条件拉取历史消息接口已经停用,就做了个通过导出聊天记录接口保存到数据库实体的功能,分享一下.
大致的思路:
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<String> 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<String> getOneDayHourList(Date date){
List<String> hourList = new ArrayList<String>();
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<ChatMessage> chatMessageList = new ArrayList<ChatMessage>();
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 id="insertBatch" parameterType="java.util.List"
useGeneratedKeys="true">
insert into chat_message (msg_id, timestamp, direction, to_user, from_user, msg, type, url)
values
<foreach collection="list" item="item" index="index" separator=",">
(#{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})
</foreach>
</insert>
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("保存前一天聊天记录");
}
}
网友评论