gfwleak/zx-dataserker-yb-flume-cus-sink-file
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(batch):

Browse Source 
1. 支持单条与批量发送接口。-涉及所有各个Topic 的conf 文件，需要修改总线的文件路径
2. 解决总线负载过高，发送重定向URL，对内容进行缓存，造成内存溢出问题。
3. 增加批量回调接口与单条进行区分，单条目前支持缓存重定向后的错误数据进行重发，目前支持最大缓存10w条。
This commit is contained in:
doufenghu
2020-10-27 17:06:02 +08:00
parent 5e26e421ce
commit 2273acd47c
16 changed files with 443 additions and 1247 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/CommonUtils/HttpClientTest.java
View File

@@ -1,30 +1,22 @@
package cn.ac.iie.cusflume.sink.CommonUtils;
import cn.ac.iie.cusflume.sink.HttpAsyncUtils.HttpClientFactory;
import cn.ac.iie.cusflume.sink.HttpAsyncUtils.HttpClientUtil;
import cn.ac.iie.cusflume.sink.HttpAsyncUtils.msgCallBack.AsyncPostMsgCallBack;
import cn.ac.iie.cusflume.sink.avroUtils.MD5Utils;
import com.alibaba.fastjson.JSON;
import org.apache.http.HttpHeaders;
import org.apache.http.HttpResponse;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.ResponseHandler;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.methods.HttpRequestBase;
import org.apache.http.entity.ByteArrayEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.protocol.HTTP;
import org.apache.http.util.EntityUtils;
import org.apache.log4j.Logger;
import org.mortbay.log.Log;
import scala.util.Try;
import java.io.IOException;
import java.net.MalformedURLException;
public class HttpClientTest {

232
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/CommonUtils/SinkHttpClientUtil.java
View File

@@ -1,232 +0,0 @@
package cn.ac.iie.cusflume.sink.CommonUtils;
import cn.ac.iie.cusflume.sink.YbHttpAvroSinkFile;
import cn.ac.iie.cusflume.sink.avroUtils.AvroMonitorTimerTask;
import cn.ac.iie.cusflume.sink.bean.redirectBean.ResRedirBody;
import cn.ac.iie.cusflume.sink.daoUtils.RealtimeCountConfig;
import com.alibaba.fastjson.JSONObject;
import com.zdjizhi.utils.StringUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.http.*;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.HttpRequestRetryHandler;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.protocol.HttpClientContext;
import org.apache.http.conn.ConnectTimeoutException;
import org.apache.http.conn.ConnectionKeepAliveStrategy;
import org.apache.http.entity.ByteArrayEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.message.BasicHeaderElementIterator;
import org.apache.http.protocol.HTTP;
import org.apache.http.util.EntityUtils;
import org.apache.log4j.Logger;
import org.mortbay.log.Log;
import javax.net.ssl.SSLException;
import javax.net.ssl.SSLHandshakeException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.UnknownHostException;
public class SinkHttpClientUtil {
private static Logger LOG = Logger.getLogger(SinkHttpClientUtil.class);
/** 全局连接池对象 */
private static final PoolingHttpClientConnectionManager connManager =
new PoolingHttpClientConnectionManager();
public static final String DEFAULT_CHARSET = "utf-8";
private static int socketTimeout = RealtimeCountConfig.HTTP_ASYNC_SOCKETTIMEOUT;//设置等待数据超时时间60秒钟 根据业务调整
private static int connectTimeout = RealtimeCountConfig.HTTP_ASYNC_CONNECTTIMEOUT;//连接超时
private static int poolSize = RealtimeCountConfig.HTTP_ASYNC_POOLSIZE;//连接池最大连接数
private static int maxPerRoute = RealtimeCountConfig.HTTP_ASYNC_MAXPERROUTE;//每个主机的并发最多只有1500
private static int connectionRequestTimeout = RealtimeCountConfig.HTTP_ASYNC_CONNECTIONREQUESTTIMEOUT; //从连接池中后去连接的timeout时间
static {
connManager.setMaxTotal(poolSize);
connManager.setDefaultMaxPerRoute(maxPerRoute);
}
public static CloseableHttpClient getHttpClient() {
RequestConfig requestConfig = RequestConfig.custom()
// 获取连接超时时间
.setConnectionRequestTimeout(connectionRequestTimeout)
// 请求超时时间
.setConnectTimeout(connectTimeout)
// 响应超时时间
.setSocketTimeout(socketTimeout)
.build();
/**
* 测出超时重试机制为了防止超时不生效而设置
* 如果直接放回false,不重试
* 这里会根据情况进行判断是否重试
*/
HttpRequestRetryHandler retry = (exception, executionCount, context) -> {
if (executionCount >= 3) {// 如果已经重试了3次，就放弃
return false;
}
if (exception instanceof NoHttpResponseException) {// 如果服务器丢掉了连接，那么就重试
return true;
}
if (exception instanceof SSLHandshakeException) {// 不要重试SSL握手异常
return false;
}
if (exception instanceof InterruptedIOException) {// 超时
return true;
}
if (exception instanceof UnknownHostException) {// 目标服务器不可达
return false;
}
if (exception instanceof ConnectTimeoutException) {// 连接被拒绝
return false;
}
if (exception instanceof SSLException) {// ssl握手异常
return false;
}
HttpClientContext clientContext = HttpClientContext.adapt(context);
HttpRequest request = clientContext.getRequest();
// 如果请求是幂等的，就再次尝试
if (!(request instanceof HttpEntityEnclosingRequest)) {
return true;
}
return false;
};
ConnectionKeepAliveStrategy myStrategy = (response, context) -> {
HeaderElementIterator it = new BasicHeaderElementIterator
(response.headerIterator(HTTP.CONN_KEEP_ALIVE));
while (it.hasNext()) {
HeaderElement he = it.nextElement();
String param = he.getName();
String value = he.getValue();
if (value != null && param.equalsIgnoreCase("timeout")) {
return Long.parseLong(value) * 1000;
}
}
return 60 * 1000;//如果没有约定，则默认定义时长为60s
};
// 创建httpClient
return HttpClients.custom()
// 把请求相关的超时信息设置到连接客户端
.setDefaultRequestConfig(requestConfig)
// 把请求重试设置到连接客户端
.setRetryHandler(retry)
.setKeepAliveStrategy(myStrategy)
// 配置连接池管理对象
.setConnectionManager(connManager)
.build();
}
public static void httpPost(String url, byte[] requestBody, int batchSize, Header... headers) {
String msg = "-1";
// 获取客户端连接对象
CloseableHttpClient httpClient = getHttpClient();
// 创建POST请求对象
CloseableHttpResponse response = null;
try {
HttpPost httpPost = new HttpPost(url);
httpPost.setHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE);
httpPost.setHeader("Content-Type", "application/json");
if (StringUtil.isNotEmpty(headers)) {
for (Header h : headers) {
httpPost.addHeader(h);
}
}
ByteArrayEntity payload = new ByteArrayEntity(requestBody);
payload.setContentEncoding("utf-8");
httpPost.setEntity(payload);
long startTime = System.currentTimeMillis();
response = httpClient.execute(httpPost);
int statusCode = response.getStatusLine().getStatusCode();
// 获取响应实体
HttpEntity entity = response.getEntity();
// 获取响应信息
msg = EntityUtils.toString(entity, "UTF-8");
Log.warn(Thread.currentThread().getName() + "，请求响应用时（ms）：" + (System.currentTimeMillis() - startTime));
if (statusCode != HttpStatus.SC_OK ) {
AvroMonitorTimerTask.addSuccessNum(batchSize);
LOG.info("数据总线响应内容：" + msg);
} else {
AvroMonitorTimerTask.addFailedNum(batchSize);
LOG.error("数据总线反馈出错：" + statusCode + "，msg" + msg);
switch (statusCode) {
case 300:
LOG.info("AsyncPostMsgCallBack==>重定向响应体-redirect-ret==>"
+ ",resRedirBodyCode:300,当前服务节点负载过高,将向其他通信地址发送请求.");
//若不包含对应字段，则不进行对象转换，减少报错
if (msg.contains("redirect")) {
ResRedirBody resRedirBody = JSONObject.parseObject(msg, ResRedirBody.class);
String redirectUrl = resRedirBody.getData().getRedirect();
if (StringUtils.isNotBlank(redirectUrl)) {
YbHttpAvroSinkFile.changeUrl(redirectUrl);
}
} else {
LOG.error("AsyncPostMsgCallBack==>服务端响应体中ResRedirBody.data.redirect字段不存在或格式不正确!!!");
}
break;
case 301:
LOG.error("AsyncPostMsgCallBack==>Status Code:" + statusCode +
",resRedirBodyCode:301,当前所有服务端节点均负载过高,暂无可用资源,请等待.");
break;
case 410:
LOG.error("AsyncPostMsgCallBack==>Status Code:" + statusCode
+ "服务端响应时间(ms)==>" + "<==,"
+ ",resRedirBodyCode:410,Cookie已过期或Cookie错误,将开始更新Cookie.");
YbHttpAvroSinkFile.updateCookie();
break;
case 500:
LOG.error("AsyncPostMsgCallBack==>Status Code:" + statusCode
+ "服务端响应时间(ms)==>" + "<==,"
+ ",resRedirBodyCode:500,处理请求过程出现系统错误.");
YbHttpAvroSinkFile.updateCookie();
break;
default:
LOG.error("AsyncPostMsgCallBack==>数据加载失败,响应体:" + "<==,"
+ "---Status Code:" + statusCode );
break;
}
}
} catch (ClientProtocolException e) {
LOG.error("协议错误: {}", e);
} catch (ParseException e) {
LOG.error("解析错误: {}", e);
} catch (IOException e) {
LOG.error("IO错误: {}", e);
} finally {
if (null != response) {
try {
EntityUtils.consumeQuietly(response.getEntity());
response.close();
} catch (IOException e) {
LOG.error("释放链接错误: {}", e);
}
}
}
}
}

2
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/HttpAsyncUtils/AsyncHttpClientGetFileCallback.java
View File

@@ -52,7 +52,7 @@ public class AsyncHttpClientGetFileCallback implements FutureCallback<HttpRespon
public AsyncHttpClientGetFileCallback(ConfigInfo configInfo, String getFileUrl, String sendMsg, AsyncHttpClientPostFileCallback asyncHttpClientPostFileCallback, int count) {
this.configInfo = configInfo;
this.postFileUrl = configInfo.getPostFileUrl();//通过configInfo赋值
this.postMsgUrl = configInfo.getPostMsgUrl();//通过configInfo赋值
this.postMsgUrl = configInfo.getSingleProduceUrl();//通过configInfo赋值
this.getFileUrl = getFileUrl;
this.sendMsg = sendMsg;
this.asyncHttpClientPostFileCallback = asyncHttpClientPostFileCallback;

2
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/HttpAsyncUtils/AsyncHttpClientPostFileCallback.java
View File

@@ -58,7 +58,7 @@ public class AsyncHttpClientPostFileCallback implements FutureCallback<HttpRespo
public AsyncHttpClientPostFileCallback(ConfigInfo configInfo, String sendMsg, int count) {
this.configInfo = configInfo;
this.postFileUrl = configInfo.getPostFileUrl();
this.postMsgUrl = configInfo.getPostMsgUrl();
this.postMsgUrl = configInfo.getSingleProduceUrl();
this.sendMsg = sendMsg;//存放对应于url的数据
this.count = count;

3
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/HttpAsyncUtils/HttpAsyncClient.java
View File

@@ -68,6 +68,7 @@ public class HttpAsyncClient {
private CloseableHttpAsyncClient proxyAsyncHttpClient;
public HttpAsyncClient() {
try {
this.asyncHttpClient = createAsyncClient(false);
this.proxyAsyncHttpClient = createAsyncClient(true);
@@ -131,7 +132,7 @@ public class HttpAsyncClient {
.register("https", new SSLIOSessionStrategy(sslcontext, NoopHostnameVerifier.INSTANCE))
.build();
// 配置io线程\
// 配置io线程
IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
.setSoKeepAlive(true)
.setTcpNoDelay(true)

633
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/HttpAsyncUtils/HttpClientUtil.java
View File

@@ -4,8 +4,8 @@ import cn.ac.iie.cusflume.sink.CommonUtils.GetDataDictionaryCodeByTopicUtils;
import cn.ac.iie.cusflume.sink.CommonUtils.GetFilePathByTopicUtils;
import cn.ac.iie.cusflume.sink.CommonUtils.HttpClientTest;
import cn.ac.iie.cusflume.sink.HttpAsyncUtils.mail.AsyncPostMailFilesCallback;
import cn.ac.iie.cusflume.sink.HttpAsyncUtils.msgCallBack.AsyncPostMsgCallBack;
import cn.ac.iie.cusflume.sink.avroUtils.AvroMonitorTimerTask;
import cn.ac.iie.cusflume.sink.HttpAsyncUtils.msgCallBack.AsyncBatchMsgCallBack;
import cn.ac.iie.cusflume.sink.HttpAsyncUtils.msgCallBack.AsyncSingleMsgCallBack;
import cn.ac.iie.cusflume.sink.avroUtils.MD5Utils;
import cn.ac.iie.cusflume.sink.avroUtils.avroRecord.GetAvroRecordByTopicUtils;
import cn.ac.iie.cusflume.sink.avroUtils.avroSchema.GetAvroSchemaByTopicUtils;
@@ -21,7 +21,6 @@ import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
@@ -42,11 +41,12 @@ import org.apache.log4j.Logger;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.MalformedURLException;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -55,347 +55,22 @@ import java.util.concurrent.Executors;
* 执行请求发送
*/
public class HttpClientUtil {
private static Logger LOG = Logger.getLogger(HttpClientUtil.class);
private static HashMap<String, Schema> schemaHashMap = new HashMap<String, Schema>();//用于存放Schma
private static Logger logger = Logger.getLogger(HttpClientUtil.class);
protected static ExecutorService pool = Executors.newFixedThreadPool(RealtimeCountConfig.HTTP_ASYNC_PARALLELISM);//线程池
private final static BlockingQueue<String> retryQueue = new ArrayBlockingQueue(100000);
private static String utf8Charset = "utf-8";
private static final MediaType JSON
= MediaType.parse("application/json; charset=utf-8");
/**
* 向指定的url发送一次post请求,参数是List<NameValuePair>
*
* @param baseUrl 请求地址
* @param list 请求参数,格式是List<NameValuePair>
* @return 返回结果, 请求失败时返回null
* @apiNote http接口处用 @RequestParam接收参数
*/
public static String httpSyncPost(String baseUrl, List<BasicNameValuePair> list) {
CloseableHttpClient httpClient = HttpClientFactory.getInstance().getHttpSyncClientPool().getHttpClient();
HttpPost httpPost = new HttpPost(baseUrl);
//Parameters
LOG.info("==== Parameters ======" + list);
CloseableHttpResponse response = null;
try {
httpPost.setEntity(new UrlEncodedFormEntity(list));
// httpPost.setHeader("Connection","close");
response = httpClient.execute(httpPost);
LOG.info("========HttpResponseProxy：========" + response.getStatusLine());
HttpEntity entity = response.getEntity();
String result = null;
if (entity != null) {
result = EntityUtils.toString(entity, "UTF-8");
LOG.info("========Response=======" + result);
}
EntityUtils.consume(entity);
return result;
} catch (Exception e) {
e.printStackTrace();
} finally {
if (response != null) {
try {
response.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
return null;
}
/**
* 向指定的url发送一次post请求,参数是字符串
*
* @param baseUrl 请求地址
* @param postString 请求参数,格式是json.toString()
* @return 返回结果, 请求失败时返回null
* @apiNote http接口处用 @RequestBody接收参数
*/
public static String httpSyncPost(String baseUrl, String postString) {
CloseableHttpClient httpClient = HttpClientFactory.getInstance().getHttpSyncClientPool().getHttpClient();
HttpPost httpPost = new HttpPost(baseUrl);
//parameters
LOG.warn("==== Parameters ======" + postString);
CloseableHttpResponse response = null;
try {
if (postString == null || "".equals(postString)) {
throw new Exception("missing post String");
}
StringEntity stringEntity = new StringEntity(postString.toString(), utf8Charset);
stringEntity.setContentEncoding("UTF-8");
stringEntity.setContentType("application/json");
httpPost.setEntity(stringEntity);
response = httpClient.execute(httpPost);
LOG.warn("========HttpResponseProxy：========" + response.getStatusLine());
HttpEntity entity = response.getEntity();
String result = null;
if (entity != null) {
result = EntityUtils.toString(entity, "UTF-8");
LOG.warn("========Response=======" + result);
}
EntityUtils.consume(entity);
return result;
} catch (Exception e) {
e.printStackTrace();
} finally {
if (response != null) {
try {
response.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
return null;
}
/**
* 向指定的url发送一次get请求,参数是List<NameValuePair>
*
* @param baseUrl 请求地址
* @param list 请求参数,格式是List<NameValuePair>
* @return 返回结果, 请求失败时返回null
* @apiNote http接口处用 @RequestParam接收参数
*/
public static String httpSyncGet(String baseUrl, List<BasicNameValuePair> list) {
CloseableHttpClient httpClient = HttpClientFactory.getInstance().getHttpSyncClientPool().getHttpClient();
HttpGet httpGet = new HttpGet(baseUrl);
//Parameters
LOG.warn("==== Parameters ======" + list);
CloseableHttpResponse response = null;
try {
if (list != null) {
String getUrl = EntityUtils
.toString(new UrlEncodedFormEntity(list));
httpGet.setURI(new URI(httpGet.getURI().toString()
+ "?" + getUrl));
} else {
httpGet.setURI(new URI(httpGet.getURI().toString()));
}
response = httpClient.execute(httpGet);
LOG.warn("========HttpResponseProxy：========" + response.getStatusLine());
HttpEntity entity = response.getEntity();
String result = null;
if (entity != null) {
result = EntityUtils.toString(entity, "UTF-8");
LOG.warn("========Response=======" + result);
}
EntityUtils.consume(entity);
return result;
} catch (Exception e) {
e.printStackTrace();
} finally {
if (response != null) {
try {
response.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
return null;
}
/**
* 向指定的url发送一次get请求,参数是字符串
*
* @param baseUrl 请求地址
* @param urlParams 请求参数,格式是String
* @return 返回结果, 请求失败时返回null
* @apiNote http接口处用 @RequestParam接收参数
*/
public static String httpSyncGet(String baseUrl, String urlParams) {
CloseableHttpClient httpClient = HttpClientFactory.getInstance().getHttpSyncClientPool().getHttpClient();
HttpGet httpGet = new HttpGet(baseUrl);
//Parameters
LOG.warn("==== Parameters ======" + urlParams);
CloseableHttpResponse response = null;
try {
if (null != urlParams || "".equals(urlParams)) {
httpGet.setURI(new URI(httpGet.getURI().toString()
+ "?" + urlParams));
} else {
httpGet.setURI(new URI(httpGet.getURI().toString()));
}
response = httpClient.execute(httpGet);
LOG.warn("========HttpResponseProxy：========" + response.getStatusLine());
HttpEntity entity = response.getEntity();
String result = null;
if (entity != null) {
result = EntityUtils.toString(entity, "UTF-8");
LOG.warn("========Response=======" + result);
}
EntityUtils.consume(entity);
return result;
} catch (Exception e) {
e.printStackTrace();
} finally {
if (response != null) {
try {
response.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
return null;
}
/**
* 向指定的url发送一次get请求,参数是字符串
*
* @param baseUrl 请求地址
* @return 返回结果, 请求失败时返回null
* @apiNote http接口处用 @RequestParam接收参数
*/
public static String httpSyncGet(String baseUrl) {
CloseableHttpClient httpClient = HttpClientFactory.getInstance().getHttpSyncClientPool().getHttpClient();
HttpGet httpGet = new HttpGet(baseUrl);
CloseableHttpResponse response = null;
try {
httpGet.setURI(new URI(httpGet.getURI().toString()));
response = httpClient.execute(httpGet);
LOG.warn("========HttpResponseProxy：========" + response.getStatusLine());
HttpEntity entity = response.getEntity();
String result = null;
if (entity != null) {
result = EntityUtils.toString(entity, "UTF-8");
LOG.warn("========Response=======" + result);
}
EntityUtils.consume(entity);
return result;
} catch (Exception e) {
e.printStackTrace();
} finally {
if (response != null) {
try {
response.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
return null;
}
/**
* 向指定的url发送一次异步post请求,参数是字符串
*
* @param baseUrl 请求地址
* @param postString 请求参数,格式是json.toString()
* @param urlParams 请求参数,格式是String
* @param callback 回调方法,格式是FutureCallback
* @return 返回结果, 请求失败时返回null
* @apiNote http接口处用 @RequestParam接收参数
*/
public static void httpAsyncPost(String baseUrl, String postString,
String urlParams, FutureCallback callback) throws Exception {
if (baseUrl == null || "".equals(baseUrl)) {
LOG.warn("we don't have base url, check config");
throw new Exception("missing base url");
}
CloseableHttpAsyncClient hc = HttpClientFactory.getInstance().getHttpAsyncClientPool()
.getAsyncHttpClient();
try {
hc.start();
HttpPost httpPost = new HttpPost(baseUrl);
// httpPost.setHeader("Connection","close");
if (null != postString) {
LOG.debug("exeAsyncReq post postBody={" + postString + "}");
StringEntity entity = new StringEntity(postString.toString(), utf8Charset);
entity.setContentEncoding("UTF-8");
entity.setContentType("application/json");
httpPost.setEntity(entity);
}
if (null != urlParams) {
httpPost.setURI(new URI(httpPost.getURI().toString()
+ "?" + urlParams));
}
LOG.warn("exeAsyncReq getparams:" + httpPost.getURI());
hc.execute(httpPost, callback);
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 向指定的url发送一次异步post请求,参数是字符串
*
* @param baseUrl 请求地址
* @param urlParams 请求参数,格式是List<BasicNameValuePair>
* @param callback 回调方法,格式是FutureCallback
* @return 返回结果, 请求失败时返回null
* @apiNote http接口处用 @RequestParam接收参数
*/
public static void httpAsyncPost(String baseUrl, List<BasicNameValuePair> postBody,
List<BasicNameValuePair> urlParams, FutureCallback callback) throws Exception {
if (baseUrl == null || "".equals(baseUrl)) {
LOG.warn("we don't have base url, check config");
throw new Exception("missing base url");
}
try {
CloseableHttpAsyncClient hc = HttpClientFactory.getInstance().getHttpAsyncClientPool()
.getAsyncHttpClient();
hc.start();
HttpPost httpPost = new HttpPost(baseUrl);
// httpPost.setHeader("Connection","close");
if (null != postBody) {
LOG.debug("exeAsyncReq post postBody={" + postBody + "}");
UrlEncodedFormEntity entity = new UrlEncodedFormEntity(
postBody, "UTF-8");
httpPost.setEntity(entity);
}
if (null != urlParams) {
String getUrl = EntityUtils
.toString(new UrlEncodedFormEntity(urlParams));
httpPost.setURI(new URI(httpPost.getURI().toString()
+ "?" + getUrl));
}
LOG.warn("exeAsyncReq getparams:" + httpPost.getURI());
hc.execute(httpPost, callback);
} catch (Exception e) {
e.printStackTrace();
}
public static BlockingQueue<String> getRetryQueue() {
return retryQueue;
}
/**
@@ -409,7 +84,7 @@ public class HttpClientUtil {
*/
public static void httpAsyncPostFileToZx(String baseUrl, byte[] fileIs, AsyncHttpClientPostFileCallback callback) throws Exception {
if (baseUrl == null || "".equals(baseUrl)) {
LOG.warn("we don't have base url, check config");
logger.warn("we don't have base url, check config");
throw new Exception("missing base url");
}
@@ -433,7 +108,7 @@ public class HttpClientUtil {
if (null != fileIs) {
httpPut.setEntity(new ByteArrayEntity(fileIs));
}
LOG.info("File httpPut.getURI()===>>>" + httpPut.getURI());
logger.info("File httpPut.getURI()===>>>" + httpPut.getURI());
localContext.setAttribute(HttpClientContext.COOKIE_STORE, cookieStore);
@@ -454,7 +129,7 @@ public class HttpClientUtil {
*/
public static void httpAsyncPostFileToZx(String baseUrl, byte[] fileIs, AsyncPostMailFilesCallback callback) throws Exception {
if (baseUrl == null || "".equals(baseUrl)) {
LOG.warn("we don't have base url, check config");
logger.warn("we don't have base url, check config");
throw new Exception("missing base url");
}
@@ -478,7 +153,7 @@ public class HttpClientUtil {
if (null != fileIs) {
httpPut.setEntity(new ByteArrayEntity(fileIs));
}
LOG.info("File httpPut.getURI()===>>>" + httpPut.getURI());
logger.info("File httpPut.getURI()===>>>" + httpPut.getURI());
localContext.setAttribute(HttpClientContext.COOKIE_STORE, cookieStore);
@@ -503,7 +178,7 @@ public class HttpClientUtil {
public static void httpAsyncGetFile(String baseUrl, FutureCallback callback) throws Exception {
if (baseUrl == null || "".equals(baseUrl)) {
LOG.warn("we don't have base url, check config");
logger.warn("we don't have base url, check config");
throw new Exception("missing base url");
}
CloseableHttpAsyncClient hc = HttpClientFactory.getInstance().getHttpAsyncClientPool().getAsyncHttpClient();
@@ -523,7 +198,7 @@ public class HttpClientUtil {
// }
// LOG.info("exeAsyncReq getparams:" + httpGet.getURI());
LOG.info("File httpGet.getURI()===>>>" + httpGet.getURI());
logger.info("File httpGet.getURI()===>>>" + httpGet.getURI());
hc.execute(httpGet, callback);
} catch (Exception e) {
e.printStackTrace();
@@ -545,7 +220,7 @@ public class HttpClientUtil {
public static void httpAsyncGet(String baseUrl, String urlParams, FutureCallback callback) throws Exception {
if (baseUrl == null || "".equals(baseUrl)) {
LOG.warn("we don't have base url, check config");
logger.warn("we don't have base url, check config");
throw new Exception("missing base url");
}
CloseableHttpAsyncClient hc = HttpClientFactory.getInstance().getHttpAsyncClientPool()
@@ -567,7 +242,7 @@ public class HttpClientUtil {
httpGet.setURI(new URI(httpGet.getURI().toString()));
}
LOG.warn("exeAsyncReq getparams:" + httpGet.getURI());
logger.warn("exeAsyncReq getparams:" + httpGet.getURI());
hc.execute(httpGet, callback);
@@ -590,7 +265,7 @@ public class HttpClientUtil {
*/
public static void httpAsyncGet(String baseUrl, List<BasicNameValuePair> urlParams, FutureCallback callback) throws Exception {
if (baseUrl == null || "".equals(baseUrl)) {
LOG.warn("we don't have base url, check config");
logger.warn("we don't have base url, check config");
throw new Exception("missing base url");
}
@@ -613,7 +288,7 @@ public class HttpClientUtil {
+ "?" + getUrl));
}
LOG.warn("exeAsyncReq getparams:" + httpGet.getURI());
logger.warn("exeAsyncReq getparams:" + httpGet.getURI());
hc.execute(httpGet, callback);
@@ -624,126 +299,6 @@ public class HttpClientUtil {
}
public static String OkSyncPost(String url, String json) throws IOException {
OkHttpClient okClient = HttpClientFactory.getInstance().getOkClientPool().getHttpClient();
RequestBody body = RequestBody.create(JSON, json);
Request request = new Request.Builder()
.url(url)
.post(body)
.build();
try (Response response = okClient.newCall(request).execute()) {
return response.body().string();
}
}
public static void OkAsyncPost(String url, String json) throws IOException {
OkHttpClient okClient = HttpClientFactory.getInstance().getOkClientPool().getHttpClient();
RequestBody body = RequestBody.create(JSON, json);
Request request = new Request.Builder()
.url(url)
.post(body)
.build();
Call call = okClient.newCall(request);
call.enqueue(new Callback() {
@Override
public void onFailure(Call call, IOException e) {
e.printStackTrace();
}
@Override
public void onResponse(Call call, Response response) throws IOException {
LOG.warn("OkAsyncPost回调:" + response.body().string());
}
});
}
public static void OkAsyncPost(String url, Map<String, String> map) throws IOException {
OkHttpClient okClient = HttpClientFactory.getInstance().getOkClientPool().getHttpClient();
FormBody.Builder formBodyBuilder = new FormBody.Builder();
for (Map.Entry<String, String> entry : map.entrySet()) {
formBodyBuilder.add(entry.getKey(), entry.getValue());
}
Request request = new Request.Builder()
.url(url)
.post(formBodyBuilder.build())
.build();
Call call = okClient.newCall(request);
call.enqueue(new Callback() {
@Override
public void onFailure(Call call, IOException e) {
e.printStackTrace();
}
@Override
public void onResponse(Call call, Response response) throws IOException {
LOG.warn("OkAsyncPost回调:" + response.body().string());
}
});
}
public static void OkAsyncPost(String url, Map<String, String> map, Callback callback) throws IOException {
OkHttpClient okClient = HttpClientFactory.getInstance().getOkClientPool().getHttpClient();
FormBody.Builder formBodyBuilder = new FormBody.Builder();
for (Map.Entry<String, String> entry : map.entrySet()) {
formBodyBuilder.add(entry.getKey(), entry.getValue());
}
Request request = new Request.Builder()
.url(url)
.post(formBodyBuilder.build())
.build();
Call call = okClient.newCall(request);
call.enqueue(callback);
}
public static String OkSyncGet(String url) throws IOException {
OkHttpClient okClient = HttpClientFactory.getInstance().getOkClientPool().getHttpClient();
Request request = new Request.Builder()
.url(url)
.build();
try (Response response = okClient.newCall(request).execute()) {
return response.body().string();
}
}
public static void OkAsyncGet(String url) throws IOException {
OkHttpClient okClient = HttpClientFactory.getInstance().getOkClientPool().getHttpClient();
Request request = new Request.Builder()
.url(url)
.build();
Call call = okClient.newCall(request);
call.enqueue(new Callback() {
@Override
public void onFailure(Call call, IOException e) {
e.printStackTrace();
}
@Override
public void onResponse(Call call, Response response) throws IOException {
LOG.warn("OkAsyncGet回调:" + response.body().string());
}
});
}
/**
* 获取单个文件并传递单个文件时使用
*
@@ -759,15 +314,17 @@ public class HttpClientUtil {
httpAsyncGetFile(getFileUrl, callback);
// LOG.info(Thread.currentThread().getName() + "===>run====>>>" + "success,now time is" + System.currentTimeMillis());
} catch (Exception e) {
LOG.error("getFileAndPostFile multithreading is error===>" + e);
logger.error("getFileAndPostFile multithreading is error===>" + e);
}
}
});
} catch (Exception e) {
LOG.error("getFileAndPostFile method is error===>" + e);
logger.error("getFileAndPostFile method is error===>" + e);
}
}
/**
* 对消息数据进行批量组装
* @param contents
@@ -815,90 +372,66 @@ public class HttpClientUtil {
}
public static void asyncProducerAvroMessageToZX(String urlProducer,String topic, byte[] results, String dataJson, String userAgent, String msgSessionCookie, int count, long postTime) throws IOException {
/**
* 用于测试总线的同步发送客户端接口，不作为生产环境使用
* @param urlProducer
* @param topic
* @param results
* @param dataJson
* @param userAgent
* @param msgSessionCookie
* @param count
* @param postTime
* @throws IOException
*/
public static void syncProducerAvroMessageTest(String urlProducer,String topic, byte[] results, String dataJson, String userAgent, String msgSessionCookie, int count, long postTime) throws IOException {
HttpClientTest.producerAvroMessageToZX(urlProducer, topic, results, dataJson, userAgent, msgSessionCookie, count, postTime);
}
public static void asyncProducerAvroMessageToZX_toBatch(String urlProducer,String topic, byte[] results, String dataJson, String userAgent, String msgSessionCookie, int count, long postTime) {
HttpPost httpPost = null;
urlProducer = urlProducer.trim();
CloseableHttpAsyncClient httpClient = null;
try {
httpClient = HttpClientFactory.getInstance().getHttpAsyncClientPool().getAsyncHttpClient();
httpClient.start();
httpPost = new HttpPost(urlProducer);
httpPost.addHeader("User-Agent", userAgent);
httpPost.setHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE);
httpPost.addHeader("Cookie", msgSessionCookie);
String md5Avro = MD5Utils.md5Encode(results);
httpPost.addHeader("Checksum", md5Avro);
logger.info("批量发送body Checksum MD5 为:" + md5Avro);
httpPost.addHeader("Content-Type", "binary/octet-stream");
httpPost.addHeader("X-Tag", "getXTAG(dataJson, topic)");
ByteArrayEntity payload = new ByteArrayEntity(results);
payload.setContentEncoding("utf-8");
httpPost.setEntity(payload);
AsyncPostMsgCallBack asyncPostMsgCallBack = new AsyncPostMsgCallBack(urlProducer, topic, dataJson, userAgent, msgSessionCookie, count, postTime);
httpClient.execute(httpPost, asyncPostMsgCallBack);
logger.info("当前Thread number ID ：" + Thread.currentThread().getId());
} catch (MalformedURLException e) {
//执行URL url = new URL()的异常
e.printStackTrace();
} catch (ClientProtocolException e) {
// 执行httpClient.execute(httpGet)的异常
e.printStackTrace();
} catch (IOException e) {
// 执行httpClient.execute(httpGet)的异常
logger.error("asyncProducerAvroToZX is IOException===>>>" + e.getMessage() + "<<<===IOException Message is==>" + dataJson + "<==");//进一步细化日志
e.printStackTrace();
} catch (Exception e) {
//handle response here... try other servers
logger.error("asyncProducerAvroToZX is Exception===>>>" + e.getMessage() + "<<<===Exception Message is==>" + dataJson + "<==");//进一步细化日志
e.printStackTrace();
}
}
/**
* 批量生产消息-总线
* 异步批量发送接口
* @param batchProduceUrl
* @param topic
* @param results
* @param xTag
* @param userAgent
* @param msgSessionCookie
* @param count
* @param postTime
*/
public static void asyncProducerAvroMessageToZX_bk(String urlProducer,String topic, byte[] results, String dataJson, String userAgent, String msgSessionCookie, int count, long postTime) {
public static void batchAsyncProduceMessage(String batchProduceUrl, String topic,
byte[] results,
String xTag, String userAgent,
String msgSessionCookie, int count, long postTime) {
HttpPost httpPost = null;
urlProducer = urlProducer.trim();
batchProduceUrl = batchProduceUrl.trim();
CloseableHttpAsyncClient httpClient = null;
try {
httpClient = HttpClientFactory.getInstance().getHttpAsyncClientPool().getAsyncHttpClient();
httpClient.start();
httpPost = new HttpPost(urlProducer);
httpPost = new HttpPost(batchProduceUrl);
httpPost.addHeader("User-Agent", userAgent);
httpPost.setHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE);
httpPost.addHeader("Cookie", msgSessionCookie);
String md5Avro = MD5Utils.md5Encode(results);
httpPost.addHeader("Checksum", md5Avro);
logger.info("批量发送body Checksum MD5 为:" + md5Avro);
String checksum = MD5Utils.md5Encode(results);
httpPost.addHeader("Checksum", checksum);
httpPost.addHeader("Content-Type", "binary/octet-stream");
httpPost.addHeader("X-Tag", getXTAG(dataJson, topic));
httpPost.addHeader("X-Tag", xTag);
ByteArrayEntity payload = new ByteArrayEntity(results);
payload.setContentEncoding("utf-8");
httpPost.setEntity(payload);
AsyncPostMsgCallBack asyncPostMsgCallBack = new AsyncPostMsgCallBack(urlProducer, topic, dataJson, userAgent, msgSessionCookie, count, postTime);
httpClient.execute(httpPost, asyncPostMsgCallBack);
logger.info("当前Thread number ID ：" + Thread.currentThread().getId());
AsyncBatchMsgCallBack asyncBatchMsgCallBack = new AsyncBatchMsgCallBack(batchProduceUrl, topic, checksum, count, postTime);
httpClient.execute(httpPost, asyncBatchMsgCallBack);
} catch (MalformedURLException e) {
//执行URL url = new URL()的异常
e.printStackTrace();
} catch (ClientProtocolException e) {
// 执行httpClient.execute(httpGet)的异常
e.printStackTrace();
} catch (IOException e) {
// 执行httpClient.execute(httpGet)的异常
logger.error("asyncProducerAvroToZX is IOException===>>>" + e.getMessage() + "<<<===IOException Message is==>" + dataJson + "<==");//进一步细化日志
logger.error("IOException===>>>" + e.getMessage() + "<<<===IOException Message is==>" + batchProduceUrl + "<==");
e.printStackTrace();
} catch (Exception e) {
//handle response here... try other servers
logger.error("asyncProducerAvroToZX is Exception===>>>" + e.getMessage() + "<<<===Exception Message is==>" + dataJson + "<==");//进一步细化日志
logger.error("Other Exception===>>>" + e.getMessage() + "<<<===Exception Message is==>" + batchProduceUrl + "<==");
e.printStackTrace();
}
@@ -909,16 +442,18 @@ public class HttpClientUtil {
* 生产AVRO数据入ZX(单条)--数据不包含schema
* 静态,适用于异步与多线程的版本
*
* @param urlProducer
* @param singleProduceUrl
* @param topic
* @param dataJson
* @param userAgent
* @param msgSessionCookie
* @return
*/
public static void asyncProducerAvroToZX(String urlProducer, String topic, String dataJson, String userAgent, String msgSessionCookie, int count, long postTime) {
public static void singleAsyncProduceMessage(String singleProduceUrl, String topic,
String dataJson, String userAgent,
String msgSessionCookie, int count, long postTime) {
HttpPost httpPost = null;
urlProducer = urlProducer.trim();
singleProduceUrl = singleProduceUrl.trim();
byte[] resultArray = null;
CloseableHttpAsyncClient httpClient = null;
try {
@@ -961,7 +496,7 @@ public class HttpClientUtil {
resultArray = alreadyGetFileTagRecordSoOnlyGetMergeAllArray(topic, resultArray);
}
httpPost = new HttpPost(urlProducer);
httpPost = new HttpPost(singleProduceUrl);
// set header
httpPost.addHeader("User-Agent", userAgent);
@@ -974,16 +509,11 @@ public class HttpClientUtil {
httpPost.addHeader("Cookie", msgSessionCookie);//不设置Cookie时,广东测试出现报错,打开广东的Cookie设置测试一下,经测试,可用
try {
String md5Avro = MD5Utils.md5Encode(resultArray);
httpPost.addHeader("Checksum", md5Avro);
logger.info("请求端Checksum MD5 avro 加密为:" + md5Avro);
logger.debug("原始数据==>" + dataJson + "<==," +
"原始数据checksum==>" + MD5Utils.md5Encode(dataJson) + "<==," +
"对应请求端Checksum MD5 avro 加密为:" + md5Avro);//20200521新增,自证清白
} catch (Exception e) {
logger.error("MD5Utils.md5Encode Method is error===>>> " + e);
}
String md5Avro = MD5Utils.md5Encode(resultArray);
httpPost.addHeader("Checksum", md5Avro);
// httpPost.addHeader("Content-Type", "application/avro+json;charset=UTF-8");
httpPost.addHeader("Content-Type", "binary/octet-stream");
@@ -999,22 +529,16 @@ public class HttpClientUtil {
ByteArrayEntity payload = new ByteArrayEntity(resultArray);
payload.setContentEncoding("utf-8");
//payload.setContentType("text/xml; charset=UTF-8");
// anti avro
httpPost.setEntity(payload);
logger.info("最终加载内容字节数组长度: " + resultArray.length);
// logger.debug("封装数据==>" + dataJson + "<==最终加载内容字节数组长度: " + resultArray.length);//20200428进一步细化日志
logger.debug("原始数据==>" + dataJson + "<==," +
// logger.debug("封装数据==>" + dataJson + "<==最终加载内容字节数组长度: " + resultArray.length);//20200428进一步细化日志
logger.info("原始数据==>" + dataJson + "<==," +
"原始数据checksum==>" + MD5Utils.md5Encode(dataJson) + "<==," +
"数据处理时间handleTime==>" + (System.currentTimeMillis() - postTime) + "<==," +
"最终加载内容字节数组长度: " + resultArray.length + "," +
"最终加载内容字节数组:" + Arrays.toString(resultArray));//20200521进一步细化日志
"对应请求端Checksum MD5 avro 加密为:" + md5Avro);//20200521进一步细化日志
//执行请求
AsyncPostMsgCallBack asyncPostMsgCallBack = new AsyncPostMsgCallBack(urlProducer, topic, dataJson, userAgent, msgSessionCookie, count, postTime);
AsyncSingleMsgCallBack asyncPostMsgCallBack = new AsyncSingleMsgCallBack(singleProduceUrl, topic, dataJson, userAgent, msgSessionCookie, count, postTime);
httpClient.execute(httpPost, asyncPostMsgCallBack);
} catch (MalformedURLException e) {
//执行URL url = new URL()的异常
e.printStackTrace();
@@ -1023,11 +547,13 @@ public class HttpClientUtil {
e.printStackTrace();
} catch (IOException e) {
// 执行httpClient.execute(httpGet)的异常
logger.error("asyncProducerAvroToZX is IOException===>>>" + e.getMessage() + "<<<===IOException Message is==>" + dataJson + "<==");//进一步细化日志
logger.error("asyncProducerAvroToZX is IOException===>>>" + e.getMessage()
+ "<<<===IOException Message is==>" + dataJson + "<==");//进一步细化日志
e.printStackTrace();
} catch (Exception e) {
//handle response here... try other servers
logger.error("asyncProducerAvroToZX is Exception===>>>" + e.getMessage() + "<<<===Exception Message is==>" + dataJson + "<==");//进一步细化日志
logger.error("asyncProducerAvroToZX is Exception===>>>"
+ e.getMessage() + "<<<===Exception Message is==>" + dataJson + "<==");//进一步细化日志
e.printStackTrace();
}
}
@@ -1049,8 +575,7 @@ public class HttpClientUtil {
}
}
private static HashMap<String, Schema> schemaHashMap = new HashMap<String, Schema>();//用于存放Schema
private static Logger logger = Logger.getLogger(HttpClientUtil.class);
/**
* 获取数据中的日志标签并将所有相关数据字节数组化后拼接返回

2
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/HttpAsyncUtils/mail/AsyncGetMailFilesCallback.java
View File

@@ -52,7 +52,7 @@ public class AsyncGetMailFilesCallback implements FutureCallback<HttpResponse> {
public AsyncGetMailFilesCallback(ConfigInfo configInfo, String getFileUrl, String sendMsg, AsyncPostMailFilesCallback asyncPostMailFilesCallback, int count, int urlCount, int dealUrlCount, LinkedList<String> attachmentsUrl) {
this.configInfo = configInfo;
this.postFileUrl = configInfo.getPostFileUrl();//通过configInfo赋值
this.postMsgUrl = configInfo.getPostMsgUrl();//通过configInfo赋值
this.postMsgUrl = configInfo.getSingleProduceUrl();//通过configInfo赋值
this.getFileUrl = getFileUrl;//初次存储的是eml_file_url,后续每次存储的都是当次需要请求的文件路径
this.sendMsg = sendMsg;
this.asyncPostMailFilesCallback = asyncPostMailFilesCallback;

4
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/HttpAsyncUtils/mail/AsyncPostMailFilesCallback.java
View File

@@ -65,7 +65,7 @@ public class AsyncPostMailFilesCallback implements FutureCallback<HttpResponse>
public AsyncPostMailFilesCallback(ConfigInfo configInfo, String sendMsg, int count, int isCount, int dealIsCount, LinkedList<String> attachmentsIdList) {
this.configInfo = configInfo;
this.postFileUrl = configInfo.getPostFileUrl();
this.postMsgUrl = configInfo.getPostMsgUrl();
this.postMsgUrl = configInfo.getSingleProduceUrl();
this.sendMsg = sendMsg;//存放对应于url的数据
this.count = count;//初始为0
this.isCount = isCount;//初始总数为url总数
@@ -275,7 +275,7 @@ public class AsyncPostMailFilesCallback implements FutureCallback<HttpResponse>
AvroMonitorTimerTask.msgReadyPostSum++;//多个文件对应一条消息
//开始推送消息进入总线
ProResBody proResBody = dclAsyncPost.avroDataLoad(configInfo.getPostMsgUrl(), configInfo.getTopicName(), sendMsg, configInfo.getBatchSize(), configInfo.getUserAgent(), configInfo.getMsgSessionCookie());
ProResBody proResBody = dclAsyncPost.avroDataLoad(configInfo.getSingleProduceUrl(), configInfo.getTopicName(), sendMsg, configInfo.getBatchSize(), configInfo.getUserAgent(), configInfo.getMsgSessionCookie());
LOG.info("Send message with many fileId to zx over,this responseBody is===>" + proResBody.toString());
} else if (dealIsCount < isCount) {//继续发送余下文件流获取id
// 注意此处dealIsCount必然是>=1的,所以不需要考虑eml_file,因为eml_file必然已经完成,直接发送attachmentsResultIsList中的流获取id

206
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/HttpAsyncUtils/msgCallBack/AsyncBatchMsgCallBack.java Normal file
View File

@@ -0,0 +1,206 @@
package cn.ac.iie.cusflume.sink.HttpAsyncUtils.msgCallBack;
import cn.ac.iie.cusflume.sink.HttpAsyncUtils.HttpClientUtil;
import cn.ac.iie.cusflume.sink.YbHttpAvroSinkFile;
import cn.ac.iie.cusflume.sink.avroUtils.AvroMonitorTimerTask;
import cn.ac.iie.cusflume.sink.bean.redirectBean.ResRedirBody;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.utils.HttpClientUtils;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.util.EntityUtils;
import org.apache.log4j.Logger;
import java.io.IOException;
import java.util.Map;
/**
* 被回调的对象，给异步的httpclient使用
*/
public class AsyncBatchMsgCallBack implements FutureCallback<HttpResponse> {
private static Logger logger = Logger.getLogger(AsyncBatchMsgCallBack.class);
private String batchProduceUrl;
private String topicName;
private String checksum;
private int count;
private long postTime;
public AsyncBatchMsgCallBack(String batchProduceUrl, String topicName,
String checksum, int count, long postTime) {
this.batchProduceUrl = batchProduceUrl;
this.topicName = topicName;
this.count = count;
this.postTime = postTime;
this.checksum = checksum;
}
public String getBatchProduceUrl() {
return batchProduceUrl;
}
public void setBatchProduceUrl(String batchProduceUrl) {
this.batchProduceUrl = batchProduceUrl;
}
public String getChecksum() {
return checksum;
}
public void setChecksum(String checksum) {
this.checksum = checksum;
}
public String getTopicName() {
return topicName;
}
public void setTopicName(String topicName) {
this.topicName = topicName;
}
public int getCount() {
return count;
}
public void setCount(int count) {
this.count = count;
}
public long getPostTime() {
return postTime;
}
public void setPostTime(long postTime) {
this.postTime = postTime;
}
/**
* 请求完成后调用该函数
*/
@Override
public void completed(HttpResponse response) {
try {
int statusCode = response.getStatusLine().getStatusCode();
HttpEntity entity = response.getEntity();
String result = null;
if (entity != null) {
result = EntityUtils.toString(entity, "UTF-8");
}
/**
* 不直接进行对象转换，除非数据加载不成功
*/
Map map = JSONObject.parseObject(result, Map.class);
int resRedirBodyCode = 0;
if (map != null) {
resRedirBodyCode = (int) map.get("code");
}
/**
* 20200818-接口细化响应码
*/
if (statusCode == 200 && resRedirBodyCode == 200) {
logger.info(
"数据加载成功,返回码: " + statusCode +
"生产数据checksum==>" + checksum + "<==," +
"服务端响应时间(ms)==>" + (System.currentTimeMillis() - postTime) + "<==," +
"数据加载成功,返回码: " + statusCode);
AvroMonitorTimerTask.addSuccessNum(count);
} else {
AvroMonitorTimerTask.addFailedNum(count);
switch (resRedirBodyCode) {
case 300:
logger.error("AsyncPostBatchMsgCallBack==>重定向响应体-redirect-ret==>" + result
+ "总线批量发送URL==>" + batchProduceUrl
+ "服务端响应时间(ms)==>" + (System.currentTimeMillis() - postTime) + "<==,"
+ "<==,Status Code:" + statusCode + ",resRedirBodyCode:300,当前服务节点负载过高,将向其他通信地址发送请求.");
//若不包含对应字段，则不进行对象转换，减少报错
if (result.contains("redirect")) {
ResRedirBody resRedirBody = JSONObject.parseObject(result, ResRedirBody.class);
String redirectUrl = resRedirBody.getData().getRedirect();
if (StringUtils.isNotBlank(redirectUrl)) {
YbHttpAvroSinkFile.changeUrl(redirectUrl);
}
} else {
logger.error("AsyncPostBatchMsgCallBack==>服务端响应体中ResRedirBody.data.redirect字段不存在或格式不正确!!!");
}
break;
case 301:
logger.error("AsyncPostBatchMsgCallBack==>Status Code:" + statusCode
+ "总线批量发送URL==>" + batchProduceUrl
+ "服务端响应时间(ms)==>" + (System.currentTimeMillis() - postTime) + "<==,"
+ ",resRedirBodyCode:301,当前所有服务端节点均负载过高,暂无可用资源,请等待.");
break;
case 410:
logger.error("AsyncPostBatchMsgCallBack==>Status Code:" + statusCode
+ "总线批量发送URL==>" + batchProduceUrl
+ "服务端响应时间(ms)==>" + (System.currentTimeMillis() - postTime) + "<==,"
+ ",resRedirBodyCode:410,Cookie已过期或Cookie错误,将开始更新Cookie.");
YbHttpAvroSinkFile.updateCookie();
break;
case 500:
logger.error("AsyncPostBatchMsgCallBack==>Status Code:" + statusCode
+ "总线批量发送URL==>" + batchProduceUrl
+ "服务端响应时间(ms)==>" + (System.currentTimeMillis() - postTime) + "<==,"
+ ",resRedirBodyCode:500,处理请求过程出现系统错误.");
YbHttpAvroSinkFile.updateCookie();
break;
default:
logger.error("AsyncPostBatchMsgCallBack==>数据加载失败,响应体:" + result
+ "总线批量发送URL==>" + batchProduceUrl
+ "服务端响应时间(ms)==>" + (System.currentTimeMillis() - postTime) + "<==,"
+ "---Status Code:" + statusCode + "---resRedirBodyCode:" + resRedirBodyCode);
break;
}
}
if (entity != null) {
try {
EntityUtils.consume(entity);
} catch (final IOException ex) {
logger.error("IOException : " + ex);
}
}
} catch (Exception e) {
logger.error("AsyncPostBatchMsgCallBack Get response from ZX is error===>>>"
+ e.getMessage());
e.printStackTrace();
} finally {
HttpClientUtils.closeQuietly(response);
}
}
/**
* 请求取消后调用该函数
*/
@Override
public void cancelled() {
logger.error("AsyncPostMagCallBack Request is cancelled");
}
/**
* 请求失败后调用该函数
*/
@Override
public void failed(Exception e) {
logger.error("AsyncPostMagCallBack Request is Failed,This Failed data checksum is ==>"+ checksum);
}
}

47
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/HttpAsyncUtils/msgCallBack/AsyncPostMsgCallBack.java → yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/HttpAsyncUtils/msgCallBack/AsyncSingleMsgCallBack.java
View File

@@ -15,14 +15,13 @@ import org.apache.http.util.EntityUtils;
import org.apache.log4j.Logger;
import java.io.IOException;
import java.util.Arrays;
import java.util.Map;
/**
* 被回调的对象，给异步的httpclient使用
*/
public class AsyncPostMsgCallBack implements FutureCallback<HttpResponse> {
private static Logger logger = Logger.getLogger(AsyncPostMsgCallBack.class);
public class AsyncSingleMsgCallBack implements FutureCallback<HttpResponse> {
private static Logger logger = Logger.getLogger(AsyncSingleMsgCallBack.class);
private String postMsgUrl;
private String topicName;
@@ -32,7 +31,7 @@ public class AsyncPostMsgCallBack implements FutureCallback<HttpResponse> {
private int count;
private long postTime;
public AsyncPostMsgCallBack(String postMsgUrl, String topicName, String dataJson, String userAgent, String msgSessionCookie, int count, long postTime) {
public AsyncSingleMsgCallBack(String postMsgUrl, String topicName, String dataJson, String userAgent, String msgSessionCookie, int count, long postTime) {
this.postMsgUrl = postMsgUrl;
this.topicName = topicName;
this.dataJson = dataJson;
@@ -103,7 +102,6 @@ public class AsyncPostMsgCallBack implements FutureCallback<HttpResponse> {
*/
@Override
public void completed(HttpResponse response) {
// ProResBody proResBody = null;
try {
int statusCode = response.getStatusLine().getStatusCode();
HttpEntity entity = response.getEntity();
@@ -121,11 +119,7 @@ public class AsyncPostMsgCallBack implements FutureCallback<HttpResponse> {
if (map != null) {
resRedirBodyCode = (int) map.get("code");
}
// int resRedirBodyCode = resRedirBody.getCode();
/* logger.debug("生产数据==>" + dataJson + "<==," +
"生产数据checksum==>" + MD5Utils.md5Encode(dataJson) + "<==," +
"返回的生产原始响应体String数据为:" + ret);*/
/**
* 20200818-接口细化响应码
*/
@@ -137,12 +131,11 @@ public class AsyncPostMsgCallBack implements FutureCallback<HttpResponse> {
"数据加载成功,返回码: " + statusCode);
AvroMonitorTimerTask.addSuccessNum(count);
} else {
AvroMonitorTimerTask.addFailedNum(count);
switch (resRedirBodyCode) {
case 300:
logger.info("AsyncPostMsgCallBack==>重定向响应体-redirect-ret==>" + ret
logger.error("AsyncSingleMsgCallBack==>重定向响应体-redirect-ret==>" + ret
+ "服务端响应时间(ms)==>" + (System.currentTimeMillis() - postTime) + "<==,"
+ "<==,Status Code:" + statusCode + ",resRedirBodyCode:300,当前服务节点负载过高,将向其他通信地址发送请求.");
//若不包含对应字段，则不进行对象转换，减少报错
@@ -153,33 +146,35 @@ public class AsyncPostMsgCallBack implements FutureCallback<HttpResponse> {
YbHttpAvroSinkFile.changeUrl(redirectUrl);
}
} else {
logger.error("AsyncPostMsgCallBack==>服务端响应体中ResRedirBody.data.redirect字段不存在或格式不正确!!!");
logger.error("AsyncSingleMsgCallBack==>服务端响应体中ResRedirBody.data.redirect字段不存在或格式不正确!!!");
}
YbHttpAvroSinkFile.redirectContents.add(dataJson);
HttpClientUtil.getRetryQueue().put(dataJson);
break;
case 301:
logger.info("AsyncPostMsgCallBack==>Status Code:" + statusCode
logger.error("AsyncSingleMsgCallBack==>Status Code:" + statusCode
+ "服务端响应时间(ms)==>" + (System.currentTimeMillis() - postTime) + "<==,"
+ ",resRedirBodyCode:301,当前所有服务端节点均负载过高,暂无可用资源,请等待.");
YbHttpAvroSinkFile.redirectContents.add(dataJson);
HttpClientUtil.getRetryQueue().put(dataJson);
break;
case 410:
logger.info("AsyncPostMsgCallBack==>Status Code:" + statusCode
logger.error("AsyncSingleMsgCallBack==>Status Code:" + statusCode
+ "服务端响应时间(ms)==>" + (System.currentTimeMillis() - postTime) + "<==,"
+ ",resRedirBodyCode:410,Cookie已过期或Cookie错误,将开始更新Cookie.");
YbHttpAvroSinkFile.updateCookie();
break;
case 500:
logger.info("AsyncPostMsgCallBack==>Status Code:" + statusCode
logger.error("AsyncSingleMsgCallBack==>Status Code:" + statusCode
+ "服务端响应时间(ms)==>" + (System.currentTimeMillis() - postTime) + "<==,"
+ ",resRedirBodyCode:500,处理请求过程出现系统错误.");
YbHttpAvroSinkFile.updateCookie();
break;
default:
logger.error("AsyncPostMsgCallBack==>数据加载失败,响应体:" + ret
logger.error("AsyncSingleMsgCallBack==>数据加载失败,响应体:" + ret
+ "服务端响应时间(ms)==>" + (System.currentTimeMillis() - postTime) + "<==,"
+ "---Status Code:" + statusCode + "---resRedirBodyCode:" + resRedirBodyCode);
break;
}
}
@@ -188,11 +183,12 @@ public class AsyncPostMsgCallBack implements FutureCallback<HttpResponse> {
try {
EntityUtils.consume(entity);
} catch (final IOException ex) {
logger.error("IOException : " + ex);
}
}
} catch (Exception e) {
logger.error("AsyncPostMsgCallBack Get response from ZX is error===>>>" + e.getMessage() + "<<<===Message is==>" + dataJson + "<==");//细化日志
logger.error("AsyncSingleMsgCallBack Get response from ZX is error===>>>" + e.getMessage() + "<<<===Message is==>" + dataJson + "<==");//细化日志
e.printStackTrace();
} finally {
HttpClientUtils.closeQuietly(response);
@@ -204,7 +200,7 @@ public class AsyncPostMsgCallBack implements FutureCallback<HttpResponse> {
*/
@Override
public void cancelled() {
logger.error("AsyncPostMagCallBack Request is cancelled");
logger.error("AsyncSingleMsgCallBack Request is cancelled");
}
/**
@@ -212,15 +208,8 @@ public class AsyncPostMsgCallBack implements FutureCallback<HttpResponse> {
*/
@Override
public void failed(Exception e) {
count++;
logger.info("AsyncPostMagCallBack Request is Failed,This Failed data is ==>" + dataJson + "<==,Retry count=" + count);
if (count > 1) {
AvroMonitorTimerTask.addFailedNum(1);
logger.error("dataJson:" + dataJson + " send failed finally,error:" + e.toString());
} else {
HttpClientUtil.asyncProducerAvroToZX(postMsgUrl, topicName, dataJson, userAgent, msgSessionCookie, count, System.currentTimeMillis());//failed失败时重试
}
logger.error("AsyncSingleMsgCallBack Request is Failed,This Failed data is ==>"
+ dataJson );
}
}

139
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/SinkService.java
View File

@@ -1,139 +0,0 @@
package cn.ac.iie.cusflume.sink;
import cn.ac.iie.cusflume.sink.CommonUtils.SinkHttpClientUtil;
import cn.ac.iie.cusflume.sink.avroUtils.AvroMonitorTimerTask;
import cn.ac.iie.cusflume.sink.avroUtils.MD5Utils;
import cn.ac.iie.cusflume.sink.bean.redirectBean.ResRedirBody;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.http.HttpEntity;
import org.apache.http.HttpStatus;
import org.apache.http.ParseException;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ByteArrayEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.protocol.HTTP;
import org.apache.http.util.EntityUtils;
import org.apache.log4j.Logger;
import org.mortbay.log.Log;
import java.io.IOException;
public class SinkService {
private static final SinkService sinkService = new SinkService();
private SinkService() {
}
//SinkHttpClientUtil.getHttpClient();
public static CloseableHttpClient httpClient = null;
private static Logger LOG = Logger.getLogger(SinkHttpClientUtil.class);
public static SinkService getInstance() {
return sinkService;
}
public void producerAvroMessageToBus(
String urlProducer, String topic, byte[] results,
String dataJson, String userAgent, String msgSessionCookie, int batchSize) {
httpClient = SinkHttpClientUtil.getHttpClient();
HttpPost httpPost = null;
urlProducer = urlProducer.trim();
String msg = "-1";
// 创建POST请求对象
CloseableHttpResponse response = null;
try {
httpPost = new HttpPost(urlProducer);
httpPost.addHeader("User-Agent", userAgent);
httpPost.setHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE);
httpPost.addHeader("Cookie", msgSessionCookie);
String md5Avro = MD5Utils.md5Encode(results);
httpPost.addHeader("Checksum", md5Avro);
LOG.info("批量发送body Checksum MD5 为:" + md5Avro);
httpPost.addHeader("Content-Type", "binary/octet-stream");
httpPost.addHeader("X-Tag", cn.ac.iie.cusflume.sink.HttpAsyncUtils.HttpClientUtil.getXTAG(dataJson, topic));
ByteArrayEntity payload = new ByteArrayEntity(results);
payload.setContentEncoding("utf-8");
httpPost.setEntity(payload);
long startTime = System.currentTimeMillis();
response = httpClient.execute(httpPost);
int statusCode = response.getStatusLine().getStatusCode();
// 获取响应实体
HttpEntity entity = response.getEntity();
// 获取响应信息
msg = EntityUtils.toString(entity, "UTF-8");
Log.warn(Thread.currentThread().getName() + "，请求响应用时（ms）：" + (System.currentTimeMillis() - startTime));
if (statusCode != HttpStatus.SC_OK) {
AvroMonitorTimerTask.addSuccessNum(batchSize);
LOG.info("数据总线响应内容：" + msg);
} else {
AvroMonitorTimerTask.addFailedNum(batchSize);
LOG.error("数据总线反馈出错：" + statusCode + "，msg" + msg);
switch (statusCode) {
case 300:
LOG.info("AsyncPostMsgCallBack==>重定向响应体-redirect-ret==>"
+ ",resRedirBodyCode:300,当前服务节点负载过高,将向其他通信地址发送请求.");
//若不包含对应字段，则不进行对象转换，减少报错
if (msg.contains("redirect")) {
ResRedirBody resRedirBody = JSONObject.parseObject(msg, ResRedirBody.class);
String redirectUrl = resRedirBody.getData().getRedirect();
if (StringUtils.isNotBlank(redirectUrl)) {
YbHttpAvroSinkFile.changeUrl(redirectUrl);
}
} else {
LOG.error("AsyncPostMsgCallBack==>服务端响应体中ResRedirBody.data.redirect字段不存在或格式不正确!!!");
}
break;
case 301:
LOG.error("AsyncPostMsgCallBack==>Status Code:" + statusCode +
",resRedirBodyCode:301,当前所有服务端节点均负载过高,暂无可用资源,请等待.");
break;
case 410:
LOG.error("AsyncPostMsgCallBack==>Status Code:" + statusCode
+ "服务端响应时间(ms)==>" + "<==,"
+ ",resRedirBodyCode:410,Cookie已过期或Cookie错误,将开始更新Cookie.");
YbHttpAvroSinkFile.updateCookie();
break;
case 500:
LOG.error("AsyncPostMsgCallBack==>Status Code:" + statusCode
+ "服务端响应时间(ms)==>" + "<==,"
+ ",resRedirBodyCode:500,处理请求过程出现系统错误.");
YbHttpAvroSinkFile.updateCookie();
break;
default:
LOG.error("AsyncPostMsgCallBack==>数据加载失败,响应体:" + "<==,"
+ "---Status Code:" + statusCode);
break;
}
}
} catch (ClientProtocolException e) {
LOG.error("协议错误: {}", e);
} catch (ParseException e) {
LOG.error("解析错误: {}", e);
} catch (IOException e) {
LOG.error("IO错误: {}", e);
} catch (Exception e) {
LOG.error("其它错误: {}", e);
} finally {
if (null != response) {
try {
EntityUtils.consumeQuietly(response.getEntity());
response.close();
} catch (IOException e) {
LOG.error("释放链接错误: {}", e);
}
}
}
}
}

351
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/YbHttpAvroSinkFile.java
View File

@@ -16,20 +16,16 @@ import org.apache.flume.sink.AbstractSink;
import org.apache.log4j.Logger;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.*;
public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
private static Logger logger = Logger.getLogger(YbHttpAvroSinkFile.class);
protected static ExecutorService pool = Executors.newFixedThreadPool(RealtimeCountConfig.HTTP_ASYNC_PARALLELISM);
private static ConcurrentLinkedQueue<byte[]> concurrentLinkedQueue = new ConcurrentLinkedQueue<byte[]>();
private static DataCenterLoad dcl;
private static String postMsgUrl;//发送消息路径,配置文件获取,发送文件与发送消息皆需要
private static String singleProduceUrl;//单条发送接口
private static String batchProduceUrl; //批量发送接口
private String postFileUrl;//发送文件路径,配置文件获取,仅发送文件时需要---若只发送消息,则此路径与postMsgUrl设置相同即可
@@ -61,14 +57,9 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
private static boolean checkTimerStart = false;//定时获取Cookie启动器
private static boolean redirectContentsPostStart = false;//定时post重定向数据集合
private static int batchInsertNum = RealtimeCountConfig.BATCH_INSERT_NUM;
/**
* 用于存储由于服务器资源不足所造成的未发送数据
*/
public static List<String> redirectContents;
/**
* 用于存放验证以及连接的url的各组成部分，方便调取
@@ -84,35 +75,45 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
super.start();
dcl = new DataCenterLoad();
redirectContents = new ArrayList<>();//初始化
/**
* 拆解初始化获取的url后缀，用于填充urlToolHm，用于后续动态负载均衡中的url变更
*/
if (StringUtils.isNotBlank(checkMsgUrl) && StringUtils.isNotBlank(postMsgUrl)) {
if (StringUtils.isNotBlank(checkMsgUrl)
&& StringUtils.isNotBlank(singleProduceUrl)
&& StringUtils.isNotBlank(batchProduceUrl)) {
urlToolHm = new HashMap<>();
makeUrlSplitMap(checkMsgUrl, "check");
makeUrlSplitMap(postMsgUrl, "post");
makeUrlSplitMap(singleProduceUrl, "post_single");
makeUrlSplitMap(batchProduceUrl, "post_batch");
} else {
logger.error("Starting YbHttpAvroSinkFile is error==>checkMsgUrl and postMsgUrl can not be null!!!!");
}
logger.warn("启动Sink File 执行程序 ==============");
//new Thread(new Consumer()).start();
logger.warn("开启多线程消费队列数据==================");
new Thread(new RetrySendMessage()).start();
logger.warn("Starting YbHttpAvroSinkFile ... ...");
}
@Override
public void configure(Context context) {
try {
postMsgUrl = context.getString("postMsgUrl", "");
Preconditions.checkNotNull("".equals(postMsgUrl), "postMsgUrl must be set!!");
logger.info("Read Post Message URL from configuration : " + postMsgUrl);
singleProduceUrl = context.getString("singleProduceUrl", "");
Preconditions.checkNotNull("".equals(singleProduceUrl), "singleProduceUrl must be set!!");
logger.info("Read Post Message URL from configuration : " + singleProduceUrl);
} catch (IllegalArgumentException e) {
throw new IllegalArgumentException("Endpoint Message URL invalid", e);
} catch (Exception e) {
logger.error("Get postMsgUrl is error : " + e);
logger.error("Get singleProduceUrl is error : " + e);
}
try {
batchProduceUrl = context.getString("batchProduceUrl", "");
Preconditions.checkNotNull("".equals(batchProduceUrl), "batchProduceUrl must be set!!");
logger.info("Read Post Message URL from configuration : " + batchProduceUrl);
} catch (IllegalArgumentException e) {
throw new IllegalArgumentException("Endpoint Message URL invalid", e);
} catch (Exception e) {
logger.error("Get batchProduceUrl is error : " + e);
}
try {
@@ -178,7 +179,7 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
public Status process() throws EventDeliveryException {
Status result = Status.READY;
Channel channel = getChannel();
Transaction transaction = null;
Transaction transaction = null;
try {
logger.debug("Current Process Thread number ID ：" + Thread.currentThread().getId());
transaction = channel.getTransaction();
@@ -200,26 +201,23 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
switch (topicName) {
//作为单条发送-新分类-20191219
/**
* 非文件消息
* 非文件消息：NTC-CONN-RECORD-LOG,NTC-COLLECT-DNS-LOG,NTC-COLLECT-SSL-LOG
* 文件类消息：NTC-COLLECT-FILE-LOG（有独立文件标签）、
* NTC-COLLECT-HTTP-DOC-LOG、NTC-COLLECT-HTTP-AV-LOG、NTC-COLLECT-FTP-DOC-LOG、
* NTC-COLLECT-MAIL-LOG、NTC-COLLECT-TELNET-LOG
* 状态消息：INFLUX-SAPP-BPS-STAT-LOG（读取回写的influxDB合计数据用作状态上传）
*/
case "NTC-CONN-RECORD-LOG":
case "NTC-COLLECT-DNS-LOG":
case "NTC-COLLECT-SSL-LOG":
/**
* 文件消息
*/
case "NTC-COLLECT-FILE-LOG"://发送独立出来的文件标签
case "NTC-COLLECT-FILE-LOG":
case "NTC-COLLECT-HTTP-DOC-LOG":
case "NTC-COLLECT-HTTP-AV-LOG"://schema等同于NTC-COLLECT-HTTP-DOC-LOG
case "NTC-COLLECT-HTTP-AV-LOG":
case "NTC-COLLECT-FTP-DOC-LOG":
case "NTC-COLLECT-MAIL-LOG":
case "NTC-COLLECT-TELNET-LOG":
/**
* 状态消息
*/
case "INFLUX-SAPP-BPS-STAT-LOG"://读取回写的influxDB合计数据用作状态上传
sendMsgLog(transaction, contents);//20191209移除文件发送,仅处理消息
//sendMsgController(transaction, contents);
case "INFLUX-SAPP-BPS-STAT-LOG":
sendMsgLog(transaction, contents);
break;
default:
logger.error("YbHttpAvroSinkFile can't find this topic:" + topicName + ".Please confirm this topicName is correct!!!");
@@ -241,7 +239,6 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
} finally {
if (transaction != null) {
transaction.close();
logger.debug("close Transaction");
}
}
return result;
@@ -260,7 +257,7 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
}
if (statusCode == 200) {
if (StringUtils.isNotBlank(acCheckMsgResBody.getSessionId())) {
logger.warn("AC msg successfully,msg sessionId is ===>" + acCheckMsgResBody.getSessionId());
logger.info("AC msg successfully,msg sessionId is ===>" + acCheckMsgResBody.getSessionId());
msgSessionCookie = acCheckMsgResBody.getSessionId();
}
} else if (statusCode == 0) {
@@ -269,8 +266,7 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
logger.error("AC msg from ZX is error,statusCode is " + statusCode + "(case)=" + acCheckMsgResBody.getCode() + "(getMethod)<===");
logger.error("This " + statusCode + " ResponseBody(contain sessionId) is ===>" + acCheckMsgResBody.toString() + "<===");
}
updateConfigInfo();//getMsgSessionCookie()更新
// return producerResBody;
updateConfigInfo();
}
/**
@@ -301,7 +297,9 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
private static ConfigInfo updateConfigInfo() {
configInfo.setCheckMsgUrl(checkMsgUrl);
configInfo.setPostMsgUrl(postMsgUrl);
configInfo.setSingleProduceUrl(singleProduceUrl);
configInfo.setBatchProduceUrl(batchProduceUrl);
configInfo.setMsgSessionCookie(msgSessionCookie);
configInfo.setMonitorSessionCookie(monitorSessionCookie);//缓存monitorSessionCookie
@@ -314,147 +312,9 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
return configInfo;
}
/**
* 发送消息控制器
* @param transaction
* @param contents
*/
private void sendMsgController(Transaction transaction, List<String> contents) {
try {
AvroMonitorTimerTask.addTotalNum(contents.size());
AvroMonitorTimerTask.addReadyPostNum(contents.size());
/**
* 获取状态回传sessionID,检查认证是否存在
*/
if (StringUtils.isBlank(monitorSessionCookie)
|| StringUtils.isBlank(msgSessionCookie)) {
getMonitorSessionCookie();//sendMsgLog-首次获取monitorSessionCookie
getMsgSessionCookie();//sendMsgLog-msgSessionCookie为空,首次接入验证
if (!checkTimerStart) {
checkCookieEveryWeek();//sendMsgLog-第一次启动检测到monitorSessionCookie为空时启动任务但不进行验证,后续间隔一段时间后开始验证,每次申请monitorSessionCookie和msgSessionCookie两个Cookie
checkTimerStart = true;
logger.warn("CheckMsgAndFileCookie Timer is started......");
}
if (!monitorStart) {//消息定时上报
AvroMonitorTimerTask.monitorMsg(monitorSessionCookie, postMsgUrl,
"monitor-msg", 1, userAgent, topicName);//sendMsgLog-日志消息
monitorStart = true;
logger.warn("MonitorMsg Timer is started......");
}
if (!redirectContentsPostStart) {
postRedirectDataEveryMin();
redirectContentsPostStart = true;
logger.warn("RedirectContents Timer Post is started......");
}
}
if (isSingle(topicName)) {
for (String content : contents) {
pool.execute(new Runnable() {
@Override
public void run() {
try {
HttpClientUtil.asyncProducerAvroToZX(postMsgUrl, topicName, content,
userAgent, msgSessionCookie, 1, System.currentTimeMillis());//初始发送count计数为0
} catch (Exception e) {
logger.error("sendMsgLog multi-thread is error==>" + e + "<==,Thread is==>" + Thread.currentThread().getName() + "<==.");
}
}
});
}
} else {
xTag = HttpClientUtil.getXTAG(contents.get(0), topicName);
dataJson = contents.get(0);
int size = contents.size() / 100;
if (size > 0) {
for (int i = 0; i < size; i++) {
// byte[] msgResults = HttpClientUtil.encapsulationData(topicName, contents.subList(i*1, (i+1)*1));
//concurrentLinkedQueue.add(msgResults);
pool.execute(new Producer(contents.subList(i * 100, (i + 1) * 100)));
}
if (contents.size() % 100 > 0) {
pool.execute(new Producer(contents.subList(size, contents.size())));
}
} else {
pool.execute(new Producer(contents));
}
}
} catch (Exception e) {
logger.error("YbHttpAvroSinkFile send Msg is error===>" + e + "<===");
transaction.commit();
} finally {
if (transaction != null) {
transaction.commit();
}
AvroMonitorTimerTask.subReadyPostNum(contents.size());
}
}
class Producer implements Runnable {
private List<String> contents;
public Producer(List<String> contents) { ;
this.contents = contents;
}
@Override
public void run() {
try {
logger.debug("Current Producer Thread number ID ：" + Thread.currentThread().getId());
byte[] msgResults = HttpClientUtil.encapsulationData(topicName, contents);
concurrentLinkedQueue.add(msgResults);
} catch (Exception e) {
logger.error("YbHttpAvroSinkFile send Msg is error===>" + e + "<===");
}
}
}
class Consumer implements Runnable {
@Override
public void run() {
logger.debug("Current Consumer Thread number ID ：" + Thread.currentThread().getId());
while(true) {
if (concurrentLinkedQueue.isEmpty()) {
logger.info("当前队列无数据，等待数据接入!");
} else {
byte[] result = concurrentLinkedQueue.poll();
HttpClientUtil.asyncProducerAvroMessageToZX_toBatch(postMsgUrl, topicName, result, dataJson,
userAgent, msgSessionCookie, 100, System.currentTimeMillis());//初始发送count计数为0
logger.info("生产数据，等待数据接入!");
}
}
}
}
/**
* 往zx发送文件数据的消息,即发送文件的message数据(结构化数据)
* 本来是作为文件消息发送,现该方法主要用于单条发送数据-20191224
*
* @param transaction
* @param contents
*/
@@ -462,7 +322,7 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
try {
logger.info("Current sendMsgLog Thread number ID ：" + Thread.currentThread().getId());
logger.debug("Current sendMsgLog Thread number ID ：" + Thread.currentThread().getId());
/**
* 获取状态回传sessionID,检查认证是否存在
@@ -478,33 +338,25 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
}
if (!monitorStart) {//消息定时上报
AvroMonitorTimerTask.monitorMsg(monitorSessionCookie, postMsgUrl,
AvroMonitorTimerTask.monitorMsg(monitorSessionCookie, singleProduceUrl,
"monitor-msg", 1, userAgent, topicName);//sendMsgLog-日志消息
monitorStart = true;
logger.debug("MonitorMsg Timer is started......");
}
if (!redirectContentsPostStart) {
postRedirectDataEveryMin();
redirectContentsPostStart = true;
logger.debug("RedirectContents Timer Post is started......");
}
logger.info("AC msg sessionId already exists,msg sessionId is ===>" + msgSessionCookie);
}
AvroMonitorTimerTask.addTotalNum(contents.size());
logger.debug("AC msg sessionId already exists,msg sessionId is ===>" + msgSessionCookie);
AvroMonitorTimerTask.addReadyPostNum(contents.size());
if (isSingle(topicName)) {
for (String content : contents) {
pool.execute(new Runnable() {
@Override
public void run() {
try {
HttpClientUtil.asyncProducerAvroToZX(postMsgUrl, topicName, content,
HttpClientUtil.singleAsyncProduceMessage(singleProduceUrl, topicName, content,
userAgent, msgSessionCookie, 1, System.currentTimeMillis());//初始发送count计数为0
} catch (Exception e) {
logger.error("sendMsgLog multi-thread is error==>" + e + "<==,Thread is==>" + Thread.currentThread().getName() + "<==.");
@@ -514,61 +366,30 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
}
} else {
long beginAvroTime = System.currentTimeMillis();
xTag = HttpClientUtil.getXTAG(contents.get(0), topicName);
byte[] msgResults = HttpClientUtil.encapsulationData(topicName, contents);
logger.debug("Batch Avro and Encapsulation Data Time(ms)："+ (System.currentTimeMillis() - beginAvroTime) );
HttpClientUtil.asyncProducerAvroMessageToZX_bk(postMsgUrl, topicName, msgResults, contents.get(0),
logger.info("Batch Avro and Encapsulation Data Time(ms)："+ (System.currentTimeMillis() - beginAvroTime) );
HttpClientUtil.batchAsyncProduceMessage(batchProduceUrl, topicName, msgResults, xTag,
userAgent, msgSessionCookie, contents.size(), System.currentTimeMillis());
/* for (String content : contents) {
HttpClientUtil.asyncProducerAvroMessageToZX_bk(postMsgUrl, topicName, msgResults, contents.get(0),
userAgent, msgSessionCookie, contents.size(), System.currentTimeMillis());//初始发送count计数为0
}*/
/*for (String content : contents) {
pool.execute(new Runnable() {
@Override
public void run() {
try {
HttpClientUtil.asyncProducerAvroMessageToZX_bk(postMsgUrl, topicName, msgResults, contents.get(0),
userAgent, msgSessionCookie, contents.size(), System.currentTimeMillis());
} catch (Exception e) {
logger.error("sendMsgLog multi-thread is error==>" + e + "<==,Thread is==>" + Thread.currentThread().getName() + "<==.");
}
}
});
}*/
/* for (String content : contents) {
pool.execute(new Runnable() {
@Override
public void run() {
try {
SinkService.getInstance().producerAvroMessageToBus(postMsgUrl, topicName, msgResults, contents.get(0),
userAgent, msgSessionCookie, contents.size());
} catch (Exception e) {
logger.error("sendMsgLog multi-thread is error==>" + e + "<==,Thread is==>" + Thread.currentThread().getName() + "<==.");
}
}
});
}*/
}
} catch (Exception e) {
logger.error("YbHttpAvroSinkFile send Msg is error===>" + e + "<===");
logger.error("YbHttpAvroSinkFile Send Msg is error===>" + e + "<===");
transaction.commit();
} finally {
if (transaction != null) {
transaction.commit();
}
AvroMonitorTimerTask.subReadyPostNum(contents.size());
}
} finally {
if (transaction != null) {
transaction.commit();
}
AvroMonitorTimerTask.subReadyPostNum(contents.size());
}
}
/**
* 单条消息发送处理逻辑
* @param topicName
* @return
*/
private static boolean isSingle(String topicName) {
if (topicName.equals("NTC-COLLECT-FILE-LOG")
@@ -619,31 +440,25 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
}, 1000 * 60 * 60 * 24 * 7, 1000 * 60 * 60 * 24 * 7);//每隔7天执行一次
}
/**
* 重定向数据集上传定时器,每隔一段时间扫描并上传一次-用于上传因为服务器资源紧张暂未上传的数据
*/
private void postRedirectDataEveryMin() {
Timer timer = new Timer();
timer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
try {
if (redirectContents.size() > 0) {
List<String> tmpListFreq = new ArrayList<>(redirectContents);
redirectContents.clear();
AvroMonitorTimerTask.msgReadyPostSum += tmpListFreq.size();
for (String content : tmpListFreq) {
HttpClientUtil.asyncProducerAvroToZX(postMsgUrl, topicName, content, userAgent, msgSessionCookie, 0, System.currentTimeMillis());//postRedirectDataEveryMin定时器-初始发送count计数为0
}
logger.info("PostRedirectDataEveryMin post to zx RedirectData size==>" + tmpListFreq.size() + "<==.");
}
} catch (Exception e) {
logger.error("PostRedirectDataEveryMin to zx everyMin is error===>>>" + e + "<===");
class RetrySendMessage implements Runnable {
@Override
public void run() {
logger.debug("Current Consumer Thread number ID ：" + Thread.currentThread().getId());
while(true) {
if (!HttpClientUtil.getRetryQueue().isEmpty()) {
logger.info("当前队列大小：" + HttpClientUtil.getRetryQueue().size());
HttpClientUtil.singleAsyncProduceMessage(singleProduceUrl, topicName,
HttpClientUtil.getRetryQueue().poll(), userAgent, msgSessionCookie, 1, System.currentTimeMillis());
}
}
}, 1000 * 60, 1000 * 60);//每隔1分钟执行一次
}
}
/**
* 动态负载均衡变更cookie-20200818
*
@@ -654,7 +469,8 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
/**
* 变更postMsgUrl与checkMsgUrl
*/
postMsgUrl = redirectUrlPort + urlToolHm.get("post_suf_path");
singleProduceUrl = redirectUrlPort + urlToolHm.get("post_single_suf_path");
batchProduceUrl = redirectUrlPort + urlToolHm.get("post_batch_suf_path");
checkMsgUrl = redirectUrlPort + urlToolHm.get("check_suf_path");
/**
@@ -662,9 +478,14 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
*/
updateCookie();
logger.info("YbHttpAvroSinkFile->changeUrl->change postMsgUrl:" + postMsgUrl + ",change checkMsgUrl:" + checkMsgUrl);
logger.warn("YbHttpAvroSinkFile->changeUrl->"
+ "change singleProduceUrl:" + singleProduceUrl
+ "change batchProduceUrl:" + batchProduceUrl
+ ",change checkMsgUrl:" + checkMsgUrl);
}
/**
* 动态负载均衡更新cookie
*/
@@ -672,7 +493,8 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
getMonitorSessionCookie();//动态负载均衡修改url,重新获取cookie
getMsgSessionCookie();//动态负载均衡修改url,重新获取cookie
logger.info("YbHttpAvroSinkFile->updateCookie update cookie,postMsgUrl:" + postMsgUrl
logger.warn("YbHttpAvroSinkFile->updateCookie update cookie,singleProduceUrl:" + singleProduceUrl
+ "batchProduceUrl:" + batchProduceUrl
+ ",checkMsgUrl:" + checkMsgUrl
+ ",获取monitorSessionCookie:" + monitorSessionCookie
+ ",获取msgSessionCookie:" + msgSessionCookie);
@@ -685,7 +507,8 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
* @param urlType
*/
private static void makeUrlSplitMap(String oldUrlPath, String urlType) {
String[] split = oldUrlPath.replace("http://", "").replace("https://", "").split("/", 2);
String[] split = oldUrlPath.replace("http://", "")
.replace("https://", "").split("/", 2);
if (split.length == 2) {
/*

2
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/avroUtils/AvroMonitorTimerTask.java
View File

@@ -88,7 +88,7 @@ public class AvroMonitorTimerTask {
if ("yb".equals(RealtimeCountConfig.MONITOR_TYPE)) {//只有当类型为一部(yb)时才进行状态上报
String sendMsg = getJson(RealtimeCountConfig.MONITOR_SYSTEM_COMPONENT_CODE, RealtimeCountConfig.MONITOR_SYSTEM_COMPONENT_CODE_FLUME, topicType);//新版-20200428
logger.info("Send monitor message is===>>>" + sendMsg + "<<<===");
HttpClientUtil.asyncProducerAvroToZX(postMonitorUrl, monitorCategory, sendMsg, userAgent, monitorSessionCookie, 0, System.currentTimeMillis());//静态方法无返回值用于多线程,初始发送count计数为0
HttpClientUtil.singleAsyncProduceMessage(postMonitorUrl, monitorCategory, sendMsg, userAgent, monitorSessionCookie, 0, System.currentTimeMillis());//静态方法无返回值用于多线程,初始发送count计数为0
}
} catch (Exception e) {
logger.error("Send monitorMsg to zx is error===>>>" + e + "<===");

27
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/avroUtils/HttpManager.java
View File

@@ -32,6 +32,7 @@ import org.apache.http.client.protocol.HttpClientContext;
import org.apache.http.config.Registry;
import org.apache.http.config.RegistryBuilder;
import org.apache.http.conn.ConnectTimeoutException;
import org.apache.http.conn.ConnectionKeepAliveStrategy;
import org.apache.http.conn.socket.ConnectionSocketFactory;
import org.apache.http.conn.socket.PlainConnectionSocketFactory;
import org.apache.http.conn.ssl.*;
@@ -41,6 +42,8 @@ import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.client.LaxRedirectStrategy;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.message.BasicHeaderElementIterator;
import org.apache.http.protocol.HTTP;
import org.apache.http.protocol.HttpContext;
import org.apache.http.util.EntityUtils;
import org.apache.log4j.Logger;
@@ -171,6 +174,21 @@ public class HttpManager {
// 声明重定向策略对象
LaxRedirectStrategy redirectStrategy = new LaxRedirectStrategy();
ConnectionKeepAliveStrategy myStrategy = (response, context) -> {
HeaderElementIterator it = new BasicHeaderElementIterator
(response.headerIterator(HTTP.CONN_KEEP_ALIVE));
while (it.hasNext()) {
HeaderElement he = it.nextElement();
String param = he.getName();
String value = he.getValue();
if (value != null && param.equalsIgnoreCase("timeout")) {
return Long.parseLong(value) * 1000;
}
}
return 10 * 1000;//如果没有约定，则默认定义时长为60s
};
/**
* 原版
*/
@@ -179,6 +197,7 @@ public class HttpManager {
.setDefaultRequestConfig(requestConfig)
.setRedirectStrategy(redirectStrategy)
.setRetryHandler(myRetryHandler)
.setKeepAliveStrategy(myStrategy)
.build();
return httpClient;
@@ -629,6 +648,7 @@ public class HttpManager {
httpPost.addHeader("User-Agent", userAgent);
httpPost.addHeader("X-Tag", xTag);//根据最新文档,目前已经不需要此头-20191217
httpPost.addHeader("Content-Type", "application/json");
httpPost.setHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE);
StringEntity payload = new StringEntity(data, Charset.forName("utf-8"));
//payload.setContentType("text/xml; charset=UTF-8");
payload.setContentEncoding("utf-8");
@@ -671,14 +691,15 @@ public class HttpManager {
// 执行httpClient.execute(httpGet)的异常
e.printStackTrace();
} finally {
if (response != null) {
if (null != response) {
try {
EntityUtils.consumeQuietly(response.getEntity());
response.close();
} catch (IOException e) {
e.printStackTrace();
logger.error("释放链接错误: " + e.getMessage());
}
}
httpPost.abort();
//httpPost.abort();
}
return acResBody;
}

22
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/bean/configBean/ConfigInfo.java
View File

@@ -9,7 +9,8 @@ package cn.ac.iie.cusflume.sink.bean.configBean;
*/
public class ConfigInfo {
private String postMsgUrl;
private String singleProduceUrl;
private String batchProduceUrl;
private String postFileUrl;
private String checkMsgUrl;
@@ -26,12 +27,20 @@ public class ConfigInfo {
private String xTag;
private int batchSize;
public String getPostMsgUrl() {
return postMsgUrl;
public String getSingleProduceUrl() {
return singleProduceUrl;
}
public void setPostMsgUrl(String postMsgUrl) {
this.postMsgUrl = postMsgUrl;
public void setSingleProduceUrl(String singleProduceUrl) {
this.singleProduceUrl = singleProduceUrl;
}
public String getBatchProduceUrl() {
return batchProduceUrl;
}
public void setBatchProduceUrl(String batchProduceUrl) {
this.batchProduceUrl = batchProduceUrl;
}
public String getPostFileUrl() {
@@ -126,7 +135,8 @@ public class ConfigInfo {
@Override
public String toString() {
return "ConfigInfo{" +
"postMsgUrl='" + postMsgUrl + '\'' +
", singleProduceUrl='" + singleProduceUrl + '\'' +
", batchProduceUrl='" + batchProduceUrl + '\'' +
", postFileUrl='" + postFileUrl + '\'' +
", checkMsgUrl='" + checkMsgUrl + '\'' +
", checkFileUrl='" + checkFileUrl + '\'' +

10
yb_http_avro_sink_file/src/main/resources/flume_config.properties
View File

@@ -2,19 +2,19 @@
http.async.parallelism=10
#异步Http客户端-等待数据超时时间,根据业务调整
http.async.socketTimeout=10000
http.async.socketTimeout=60000
#异步Http客户端-连接超时时间
http.async.connectTimeout=10000
http.async.connectTimeout=30000
#异步Http客户端-连接池最大连接数
http.async.poolSize=80
http.async.poolSize=4
#异步Http客户端-每个主机的并发最多只有1500
http.async.maxPerRoute=80
http.async.maxPerRoute=4
#异步Http客户端-从连接池中后去连接的timeout时间
http.async.connectionRequestTimeout=15000
http.async.connectionRequestTimeout=30000
#Schema配置信息