MQTT的学习研究（七）基于HTTP POST MQTT 发布消息服务端使用

2022-08-04 11:30:24

参阅官方文档

http://publib.boulder.ibm.com/infocenter/wmqv7/v7r0/topic/com.ibm.mq.csqzau.doc/ts21220_.htm

HTTP POST puts a message to a queue, or a publication to a topic. The HTTPPOST Java sample is an example an HTTP POST request of a message to a queue. Instead of using Java, you could create an HTTPPOST request using a browser form, or an AJAX toolkit instead.

Figure 1 shows an HTTP request to put a message on a queue called myQueue. This request contains the HTTP header x-msg-correlId to set the correlation ID of the WebSphere MQ message.

Figure 1. Example of an HTTP POST request to a queue 
POST /msg/queue/myQueue/ HTTP/1.1

Host: www.example.org

Content-Type: text/plain

x-msg-correlID: 1234567890

Content-Length: 50

Here's my message body that will appear on the queue.

Figure 2 shows the response sent back to the client. There is no response content.

Figure 2. Example of an HTTP POST response 
HTTP/1.1 200 OK

Date: Wed, 2 Jan 2007 22:38:34 GMT

Server: Apache-Coyote/1.1 WMQ-HTTP/1.1 JEE-Bridge/1.1

Content-Length: 0

请求的协议格式和请求的响应格式

The HTTP POST operation puts a message on a WebSphere® MQ queue, or publishes a message to a topic.

Syntax

Request

>>-POST-- --| Path |-- --HTTP version--CRLF--------------------->

   .-CRLF---------------.  .-CRLF---------------.   

   V                    |  V                    |   

>----+----------------+-+----+----------------+-+--------------->

     '-general-header-'      '-request-header-'     

   .-CRLF----------------------------.        .-CRLF----.   

   V                                 |        V         |   

>----+-----------------------------+-+--CRLF----Message-+------><

     '-| entity header (Request) |-'                        

Path

|--/--contextRoot--/-------------------------------------------->

>--msg/--+-queue/--queueName--+-------------+-+--/--------------|

         |                    '-@--qMgrName-' |      

         '-topic/--topicName------------------'      

entity-header (Request)

|--+----------------------------------------------+-------------|

   +-standard entity-header-- --entity-value------+   

   +-x-msg-class-- --message type-----------------+   

   +-x-msg-correlId-- --correlation ID------------+   

   +-x-msg-encoding-- --encoding type-------------+   

   +-x-msg-expiry-- --duration--------------------+   

   +-x-msg-format-- --message format--------------+   

   +-x-msg-msgId-- --message ID-------------------+   

   +-x-msg-persistence-- --persistence------------+   

   +-x-msg-priority-- --priority class------------+   

   +-x-msg-replyTo-- --reply-to queue-------------+   

   +-x-msg-require-headers-- --entity header name-+   

   '-x-msg-usr-- --user properties----------------'   
Note:
  1. If a question mark (?) is used it must be substituted with %3f. For example, orange?topic should be specified as orange%3ftopic.
  2. @qMgrName is only valid on an HTTP POST
Response

>>-HTTP version-- --HTTP Status-Code-- --HTTP Reason-Phrase--CRLF-->

   .-CRLF---------------.  .-CRLF----------------.   

   V                    |  V                     |   

>----+----------------+-+----+-----------------+-+-------------->

     '-general-header-'      '-response-header-'     

   .-CRLF-----------------------------.   

   V                                  |   

>----+------------------------------+-+------------------------><

     '-| entity-header (Response) |-'     

entity-header (Response)

|--+-----------------------------------------+------------------|

   +-standard entity-header-- --entity-value-+   

   +-x-msg-class-- --message type------------+   

   +-x-msg-correlId-- --correlation ID-------+   

   +-x-msg-encoding-- --encoding type--------+   

   +-x-msg-expiry-- --duration---------------+   

   +-x-msg-format-- --message format---------+   

   +-x-msg-msgId-- --message ID--------------+   

   +-x-msg-persistence-- --persistence-------+   

   +-x-msg-priority-- --priority class-------+   

   +-x-msg-replyTo-- --reply-to queue--------+   

   +-x-msg-timestamp-- --HTTP-date-----------+   

   '-x-msg-usr-- --user properties-----------'   
HTTP POST方式实现如下：
  1. package com.etrip.mqttv3.http;
  2. /**
  3. * This sample shows how to post a message. It has the same behaviour as the
  4. * amqsput command in that it will read in lines from the command line and put
  5. * them to the queue. It will put non-persistent String messages on to the queue
  6. * with UNLIMITED expiry and LOW (0) priority. The program is terminated by
  7. * either EOF being put into the entry line (^Z on windows) or a blank line.
  8. * usage: java HTTPPOST <Queue (default=SYSTEM.DEFAULT.LOCAL.QUEUE)> <host:port
  9. * (default localhost:8080> <context-root (the MQ Bridge for HTTP's
  10. * context-root)>
  11. */
  12. import java.io.BufferedReader;
  13. import java.io.BufferedWriter;
  14. import java.io.IOException;
  15. import java.io.InputStreamReader;
  16. import java.io.OutputStream;
  17. import java.io.OutputStreamWriter;
  18. import java.net.HttpURLConnection;
  19. import java.net.MalformedURLException;
  20. import java.net.URL;
  21. /**
  22. *
  23. * 采用HTTP POST发布相关的消息
  24. *     The HTTP POST operation puts a message on a WebSphere® MQ queue, or publishes
  25. *  a message to a topic.
  26. *
  27. *  发布消息到主题或者队列的路径：
  28. *
  29. *
  30. *
  31. *
  32. *
  33. * @author longgangbai
  34. */
  35. public class HTTPPOST
  36. {
  37. private static final String DEFAULT_HOST = "localhost";
  38. private static final String DEFAULT_PORT = "8080";
  39. private static final String DEFAULT_QUEUE = "SYSTEM.DEFAULT.LOCAL.QUEUE";
  40. private static final String DEFAULT_CONTEXT_ROOT = "mq";
  41. private static final String CRLF = "\r\n";
  42. public static int MALFORMED_URL_EXCEPTION_RC = -1;
  43. public static int END_IOEXCEPTION_RC = -2;
  44. /**
  45. * 构建发布主题队列路径
  46. *
  47. * @param host
  48. * @param port
  49. * @param context
  50. * @param queueName
  51. */
  52. private static String getPublishQueueURL(String host, String port,
  53. String context, String queueName) {
  54. StringBuffer urlString =new StringBuffer("http://");
  55. if(StringUtils.isEmtry(host)){
  56. host=DEFAULT_HOST;
  57. }
  58. if(StringUtils.isEmtry(port)){
  59. port=DEFAULT_PORT;
  60. }
  61. urlString.append(host).append(":").append(port);
  62. if(StringUtils.isEmtry(context)){
  63. context=DEFAULT_CONTEXT_ROOT;
  64. }
  65. urlString.append("/");
  66. urlString.append(context);
  67. urlString.append("/msg/queue/");
  68. if(StringUtils.isEmtry(queueName)){
  69. }
  70. queueName=DEFAULT_QUEUE;
  71. urlString.append(queueName);
  72. System.out.println("urlString="+urlString);
  73. return urlString.toString();
  74. }
  75. /**
  76. *
  77. * @param host
  78. * @param port
  79. * @param context
  80. * @param queueName
  81. * @param message
  82. * @return
  83. * @throws MalformedURLException
  84. */
  85. public static boolean publishTopic(String host,String port,String context,String queueName,String message ){
  86. boolean response = true;
  87. HttpURLConnection connection=null;
  88. try {
  89. String publishURL=getPublishQueueURL(host, port, context, queueName);
  90. URL url=new URL(publishURL);
  91. connection = (HttpURLConnection) url.openConnection();
  92. /* Build the headers */
  93. // the verb first
  94. connection.setRequestMethod("POST");
  95. // Content type is a string message
  96. connection.setRequestProperty("content-type", "text/plain");
  97. // set the message priority to low
  98. connection.setRequestProperty("x-msg-priority", "LOW");
  99. // Ensure we can get the output stream from the connection
  100. connection.setDoOutput(true);
  101. OutputStream outputStream = connection.getOutputStream();
  102. // wrapper the outputstream in a writer
  103. BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(
  104. outputStream));
  105. // Now write the actual content.
  106. // Make sure the CRLF is there in case some HTTP servers don't understand
  107. // that it's the end of the message
  108. writer.write(message + CRLF + CRLF);
  109. writer.flush();
  110. // now actually send the message
  111. connection.connect();
  112. // check the response for errors
  113. int responseCode = connection.getResponseCode();
  114. if (responseCode != 200)
  115. {
  116. String responseMessage =connection.getResponseMessage();
  117. System.out.println("responsere sponseCode "+responseCode+" response request ="+responseMessage);
  118. System.out.println("responsere context ");
  119. BufferedReader reader = new BufferedReader(new InputStreamReader(
  120. connection.getErrorStream()));
  121. String line = null;
  122. while ((line = reader.readLine()) != null)
  123. {
  124. System.out.println(line);
  125. }
  126. connection.disconnect();
  127. response = false;
  128. }else{
  129. //获取相应的消息头信息
  130. String responseQueueName=connection.getHeaderField("x-msg-replyTo");
  131. System.out.println("responseQueueName="+responseQueueName);
  132. System.out.println("response successful context :"+connection.getResponseMessage());
  133. }
  134. } catch (MalformedURLException e) {
  135. response = false;
  136. e.printStackTrace();
  137. // TODO: handle exception
  138. } catch (IOException e) {
  139. response = false;
  140. // TODO Auto-generated catch block
  141. e.printStackTrace();
  142. }finally{
  143. connection.disconnect();
  144. }
  145. return response;
  146. }
  147. public static void main(String[] args) {
  148. HTTPPOST.publishTopic("192.168.208.46", "8080", "mq", "java_lover", "this is a message ");
  149. }
  150. }
上一篇：python爬虫，使用urllib2库报错


下一篇：python 爬虫基础知识一