HTTP 上报

新建实时同步任务

在 Dashboard 或者通过命令行工具创建任务,数据源类型使用 Auto Push (自主推送)。

方式一:调用 URL 上报

  1. curl -X POST -d 'groupId=give_your_group_id&streamId=give_your_stream_id&dt=data_time&body=give_your_data_body&cnt=1' http://dataproxy_url:46802/dataproxy/message
  • 参数说明:
参数含义备注
groupId数据流组 id
streamId数据流 ID
body推送的数据内容
dt推送的数据时间毫秒为单位的时间戳
cnt推送条数
  • 返回值:
返回码含义
1成功
非1失败

方式二:封装 HTTP Client(Java)

需要 httpclientcommons-lang3jackson-databind,代码示例:

  1. public class DataPush {
  2. private static CloseableHttpClient httpClient;
  3. private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
  4. private final Random rand = new Random();
  5. private String sendByHttp(List<String> bodies, String groupId, String streamId, long dataTime,
  6. long timeout, TimeUnit timeUnit, List<String> addresses) throws Exception {
  7. if (null == addresses || addresses.isEmpty()) {
  8. throw new RuntimeException("addresses are null");
  9. }
  10. HttpPost httpPost = null;
  11. CloseableHttpResponse response = null;
  12. try {
  13. if (httpClient == null) {
  14. httpClient = constructHttpClient(timeout, timeUnit);
  15. }
  16. int randomNum = rand.nextInt((addresses.size() - 1) + 1);
  17. String url = "http://" + addresses.get(randomNum) + "/dataproxy/message";
  18. httpPost = new HttpPost(url);
  19. httpPost.setHeader(HttpHeaders.CONNECTION, "close");
  20. httpPost.setHeader(HttpHeaders.CONTENT_TYPE, "application/x-www-form-urlencoded");
  21. ArrayList<BasicNameValuePair> contents = getContents(bodies, groupId, streamId, dataTime);
  22. String s = URLEncodedUtils.format(contents, StandardCharsets.UTF_8);
  23. httpPost.setEntity(new StringEntity(s));
  24. response = httpClient.execute(httpPost);
  25. String returnStr = EntityUtils.toString(response.getEntity());
  26. if (StringUtils.isNotBlank(returnStr) && response.getStatusLine().getStatusCode() == 200) {
  27. JsonNode jsonNode = OBJECT_MAPPER.readTree(returnStr);
  28. if (jsonNode.has("code")) {
  29. int code = jsonNode.get("code").asInt();
  30. if (code == 0) {
  31. return "success";
  32. } else {
  33. return "fail";
  34. }
  35. }
  36. } else {
  37. throw new Exception("exception to get response from request " + returnStr + " "
  38. + response.getStatusLine().getStatusCode());
  39. }
  40. } finally {
  41. if (httpPost != null) {
  42. httpPost.releaseConnection();
  43. }
  44. if (response != null) {
  45. response.close();
  46. }
  47. }
  48. return "fail";
  49. }
  50. private static synchronized CloseableHttpClient constructHttpClient(long timeout, TimeUnit timeUnit) {
  51. if (httpClient != null) {
  52. return httpClient;
  53. }
  54. long timeoutInMs = timeUnit.toMillis(timeout);
  55. RequestConfig requestConfig = RequestConfig.custom()
  56. .setConnectTimeout((int) timeoutInMs)
  57. .setSocketTimeout((int) timeoutInMs).build();
  58. HttpClientBuilder httpClientBuilder = HttpClientBuilder.create();
  59. httpClientBuilder.setDefaultRequestConfig(requestConfig);
  60. return httpClientBuilder.build();
  61. }
  62. private static ArrayList<BasicNameValuePair> getContents(List<String> bodies,
  63. String groupId, String streamId, long dt) {
  64. ArrayList<BasicNameValuePair> params = new ArrayList<BasicNameValuePair>();
  65. params.add(new BasicNameValuePair("groupId", groupId));
  66. params.add(new BasicNameValuePair("streamId", streamId));
  67. params.add(new BasicNameValuePair("dt", String.valueOf(dt)));
  68. params.add(new BasicNameValuePair("body", StringUtils.join(bodies, "\n")));
  69. params.add(new BasicNameValuePair("cnt", String.valueOf(bodies.size())));
  70. return params;
  71. }
  72. }