朱兆平 / dsly-hz-analysis · 提交

转到一个项目

作者 朱兆平
提交 1b681e84868707a322daf03d8e30a83ccd5e454d 1 个父辈 7c921571

优化重发校验

内嵌 并排对比
正在显示 4 个修改的文件 包含 73 行增加 和 29 行删除
... ... @@ -27,7 +27,7 @@ public class MessageTransApplication {
@Bean
public TaskScheduler taskScheduler() {
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.setPoolSize(10);
taskScheduler.setPoolSize(3);
return taskScheduler;
}
... ...
... ... @@ -3,6 +3,7 @@ package com.tianbo.messagebus.controller;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.tianbo.messagebus.service.MessageBusProcessor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
... ... @@ -16,6 +17,7 @@ import java.io.Serializable;
import java.util.Date;
@RestController()
@Slf4j
@RequestMapping("/")
public class HeartbeatController implements Serializable {
private static final long serialVersionUID = 1L;
... ... @@ -34,6 +36,7 @@ public class HeartbeatController implements Serializable {
@PostMapping("getmsg")
public void getmsg(){
log.error("for test");
messageBusDemo.getMsg();
}
... ...
... ... @@ -13,6 +13,8 @@ import org.springframework.stereotype.Service;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.util.StringUtils;
import org.springframework.web.client.HttpClientErrorException;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;
import javax.annotation.Resource;
... ... @@ -69,6 +71,10 @@ public class MessageBusProcessor {
*/
private static Boolean LOGIN_STATUS=false;
/**
* 失败重发请求次数
*/
private static final int RETRY_TIMES= 3;
/**
* HTTP请求框架
... ... @@ -175,7 +181,18 @@ public class MessageBusProcessor {
}
}
@Scheduled(fixedRate = 10000)
/**
* 获取返回信息的状态码
* @param response response对象
* @return
*/
private String getResponseDataCode(ResponseEntity<String> response){
JSONObject resJson = JSON.parseObject(response.getBody());
String code = resJson.getString("code");
return code;
}
@Scheduled(fixedDelay = 10000)
public void heartBit() {
if (!StringUtils.isEmpty(TOKEN) && LOGIN_STATUS){
/*
... ... @@ -200,10 +217,12 @@ public class MessageBusProcessor {
* 提交HTTP访问，获取返回信息
*/
ResponseEntity<String> response = restTemplate.postForEntity(HEARTBEAT_URL, request, String.class);
// 输出结果
System.out.println(response.getBody());
log.info(response.getBody());
if (response.getStatusCode().equals(HttpStatus.OK)) {
log.info("心跳成功");
log.info("心跳成功,token:[{}]",TOKEN);
} else {
log.error("心跳失败");
}
... ... @@ -217,6 +236,8 @@ public class MessageBusProcessor {
*/
public Boolean sendMsg(MSG msg) {
if (LOGIN_STATUS) {
try{
log.info("开始转发消息:{}",msg.toString());
/*
* 发起HTTP 登录请求
* 登录接口的请求头为application/json
... ... @@ -252,20 +273,22 @@ public class MessageBusProcessor {
HttpEntity<MSGS> request = new HttpEntity<MSGS>(msgs, headers);
ResponseEntity<String> response = restTemplate.postForEntity(SEND_MSG_URL, request, String.class);
if (response.getStatusCode().equals(HttpStatus.OK)) {
JSONObject resJson = JSON.parseObject(response.getBody());
String code = resJson.getString("code");
System.out.println(response.getBody());
if (response.getStatusCode().equals(HttpStatus.OK) && "200".equals(code)) {
if ("200".equals(code)) {
log.info("消息发送成功");
return true;
} else {
log.error("消息发送失败");
}
}
log.info("消息发送失败->{}",response.getBody());
}catch (Exception e){
log.info("消息发送失败->{},失败原因:{}",msg.toString(),e.toString());
log.error("消息发送失败->{},失败原因:{}",msg.toString(),e.toString());
return false;
}
}else {
log.info("未登陆，消息发送失败");
}
return false;
}
... ... @@ -276,8 +299,8 @@ public class MessageBusProcessor {
*
* @return
*/
@Async
@Scheduled(fixedRate = 300)
// @Async
@Scheduled(fixedRate = 1000)
public JSONArray getMsg() {
if(!LOGIN_STATUS){
login();
... ... @@ -287,6 +310,7 @@ public class MessageBusProcessor {
* 发起HTTP 登录请求
* 登录接口的请求头为application/json
*/
try{
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
headers.setBearerAuth(TOKEN);
... ... @@ -302,22 +326,24 @@ public class MessageBusProcessor {
*/
ResponseEntity<String> response = restTemplate.postForEntity(GET_MSG_URL, request, String.class);
// 输出结果
log.info("获取到消息返回---{}---",response.getBody());
if (response.getStatusCode().equals(HttpStatus.OK)) {
/*
* 从返回信息中确定是否获取到消息
*/
JSONObject resJson = JSON.parseObject(response.getBody());
String code = resJson.getString("code");
if (response.getStatusCode().equals(HttpStatus.OK) && "200".equals(code)) {
if ("200".equals(code)){
JSONArray data = resJson.getJSONArray("data");
log.info("消息接收成功，接收消息为>>>{}<<<",data.toString());
log.info("消息接收成功，接收消息数量>>>{}<<<",data.size());
for (int i = 0; i<data.size() ; i++) {
/*
取得是大数据小组的实体,他们的msg.body的封装是以对象实体object封装的。不是json字符串。
*/
String msg = data.getObject(i,String.class);
log.info("开始转发消息---{}---",msg);
JSONObject rootJson = JSON.parseObject(msg);
JSONObject msgJson = rootJson.getJSONObject("MSG");
JSONObject body = msgJson.getJSONObject("BODY");
... ... @@ -329,44 +355,54 @@ public class MessageBusProcessor {
transMsg.setHEADER(msgHeader);
transMsg.setBODY(transBody);
transMsg.toString();
/*
自定义对返回数据的处理
*/
log.info("开始转发消息");
Boolean sendResult = sendMsg(transMsg);
/**
* todo:消息失败处理
* todo:转发消息失败处理
*/
if(!sendResult){
log.error("!!!!!!消息--->{}<---转发失败!!!!!!,尝试重发",transMsg.toString());
//todo:消息备份或者重发？
reSend(transMsg);
}
}
return data;
}
} else {
log.error("消息获取失败");
return new JSONArray();
}
}catch (Exception httpE){
log.info("消息获取失败,失败原因:{}",httpE.toString());
log.error("消息获取失败,失败原因:{}",httpE.toString());
}
return new JSONArray();
}
/**
* 读取备份消息并消息重发
* @return
*/
public Boolean reSend(){
return false;
public void reSend(MSG msg){
log.error("***进入重发***");
for (int i = 0; i < RETRY_TIMES; i++) {
Boolean sendResult = sendMsg(msg);
if (sendResult){
log.error("***重发成功***");
break;
}
}
log.error("***已尝试重发三次，重发失败***");
}
}
/**
* 消息发送实体类
* 消息实体类
*/
class MSGS implements Serializable {
private MSG MSG;
... ... @@ -405,6 +441,11 @@ class MSG {
public void setBODY(String BODY) {
this.BODY = BODY;
}
@Override
public String toString() {
return this.BODY;
}
}
/**
... ...