MQTT订阅-数据保存-HTTP发布框架

package com.hyzm.netty; import java.io.IOException; import java.net.Authenticator; import java.net.InetSocketAddress; import java.net.PasswordAuthentication; import java.net.ProxySelector; import java.net.URI; import java.net.URISyntaxException; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; import java.time.Duration; import java.util.concurrent.BlockingQueue; import com.hyzm.netty.entity.OriginalData; public class HttpPublisher extends PublisherBase { /* * 回传数据的消费者，将回传的数据发布到http服务器 */ private BlockingQueue<Message> m_sendMsg; private HttpClient m_client; private String m_hostUrl = "https://postman-echo.com/post"; // 仅用于测试 private String m_user = null; private String m_password = null; public HttpPublisher() { super(); m_sendMsg = Common.sentHttpMsg; connect(); } public HttpPublisher(String host_url) { super(); m_hostUrl = host_url; m_sendMsg = Common.sentHttpMsg; connect(); } @Override protected void connect() { m_client = HttpClient.newBuilder() .version(HttpClient.Version.HTTP_2) .connectTimeout(Duration.ofSeconds(20)) // .authenticator(new Authenticator() { // @Override // protected PasswordAuthentication getPasswordAuthentication() { // return new PasswordAuthentication(m_user, m_password.toCharArray()); // } // }) // .proxy(ProxySelector.of(new InetSocketAddress("proxy.example.com", 80))) .build(); } @Override public void run() { while (true) { try { Message msg = m_sendMsg.take(); // 将得到的数据发布出去 System.out.println("Http Publisher收到回传的数据： " + msg.toJSON()); OriginalData data = msg.getOriginalData(); HttpResponse<String> res = requestPost(m_hostUrl, msg.toJSON()); if (res != null && res.statusCode() == 200) { // 发送成功 data.setConfirmed(1); // 暂时设置此值表示发送成功 } else { data.setConfirmed(-1); // 暂时设置此值表示发送失败 } m_origDataCtrl.save(data); } catch (InterruptedException e) { e.printStackTrace(); } catch (Exception e) { e.printStackTrace(); } } } // 实现了get请求 public HttpResponse<String> requestGet(String url) { HttpResponse<String> response = null; try { HttpRequest request = HttpRequest.newBuilder() .uri(new URI(url)) .timeout(Duration.ofSeconds(10)) .headers("Content-Type", "text/plain;charset=UTF-8") .GET() .build(); response = m_client.send(request, HttpResponse.BodyHandlers.ofString()); } catch (IOException ioe) { ioe.printStackTrace(); } catch (InterruptedException ie) { ie.printStackTrace(); } catch (URISyntaxException use) { use.printStackTrace(); } return response; } // 实现了post请求 public HttpResponse<String> requestPost(String url, String param) { HttpResponse<String> response = null; try { HttpRequest request = HttpRequest.newBuilder() .uri(new URI(url)) .timeout(Duration.ofSeconds(10)) .header("Content-Type", "application/json;charset=UTF-8") .POST(HttpRequest.BodyPublishers.ofString(param)) .build(); response = m_client.send(request, HttpResponse.BodyHandlers.ofString()); } catch (IOException ioe) { ioe.printStackTrace(); } catch (InterruptedException ie) { ie.printStackTrace(); } catch (URISyntaxException use) { use.printStackTrace(); } return response; } }