RabbitMQ入门参考博客：http://m./article/details?id=50487028（内含介绍、环境配置及基本实现说明）
RabbitMQ官方文档参考，发送Topic exchange队列：/tutorials/tutorial-five-java.html（内含各种语言的各种实现方法）
项目部分代码：
1.config类
package mon;public class Hxyconfig {//MQ服务器的Host和用户名密码public static String sendMQHostHXY="10.1.4.31";public static String UserNameHXY="admin";public static String PassWordHXY="admin";public static int portHXY=5672;//public static String exchangeName="out_exchange_ict";//多久发送一次MQ数据public static int sendSecond=5;//消息队列名//船舶异常public static String queAnomaly="out_ict_anomaly";//网络监控public static String queAISNetworkState="out_ict_ais_network_state";//船数统计public static String queShipStatistic="out_ict_ship_statistic";//雷达融合public static String queRadarFusion="out_ict_radar_fusion";//Java后端程序部署服务器地址public static String HostHXY="10.1.2.11";}
2.定时请求接口并且发送topic exchange的消息到服务器
package mon;import java.io.BufferedReader;import java.io.IOException;import java.io.InputStreamReader;import java.io.PrintWriter;import .URL;import .URLConnection;import java.util.Date;import java.util.List;import java.util.Map;import java.util.Timer;import java.util.TimerTask;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;public class HttpRequest {/**** @param url* 发送请求的URL* @param param* 请求参数，请求参数应该是 name1=value1&name2=value2 的形式。* @return URL 所代表远程资源的响应结果*/public static String sendGet(String url, String param) {String result = "";BufferedReader in = null;try {String urlNameString = url + "?" + param;URL realUrl = new URL(urlNameString);// 打开和URL之间的连接URLConnection connection = realUrl.openConnection();// 设置通用的请求属性connection.setRequestProperty("accept", "*/*");connection.setRequestProperty("connection", "Keep-Alive");connection.setRequestProperty("user-agent","Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1;SV1)");// 建立实际的连接connection.connect();// 获取所有响应头字段Map<String, List<String>> map = connection.getHeaderFields();// 遍历所有的响应头字段for (String key : map.keySet()) {//System.out.println(key + "--->" + map.get(key));}// 定义 BufferedReader输入流来读取URL的响应in = new BufferedReader(new InputStreamReader(connection.getInputStream()));String line;while ((line = in.readLine()) != null) {result += line;}} catch (Exception e) {System.out.println("发送GET请求出现异常！" + e);e.printStackTrace();}// 使用finally块来关闭输入流finally {try {if (in != null) {in.close();}} catch (Exception e2) {e2.printStackTrace();}}return result;}/*** 向指定 URL 发送POST方法的请求** @param url* 发送请求的 URL* @param param* 请求参数，请求参数应该是 name1=value1&name2=value2 的形式。* @return 所代表远程资源的响应结果*/public static String sendPost(String url, String param) {PrintWriter out = null;BufferedReader in = null;String result = "";try {URL realUrl = new URL(url);// 打开和URL之间的连接URLConnection conn = realUrl.openConnection();// 设置通用的请求属性conn.setRequestProperty("accept", "*/*");conn.setRequestProperty("connection", "Keep-Alive");conn.setRequestProperty("user-agent","Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1;SV1)");// 发送POST请求必须设置如下两行conn.setDoOutput(true);conn.setDoInput(true);// 获取URLConnection对象对应的输出流out = new PrintWriter(conn.getOutputStream());// 发送请求参数out.print(param);// flush输出流的缓冲out.flush();// 定义BufferedReader输入流来读取URL的响应in = new BufferedReader(new InputStreamReader(conn.getInputStream()));String line;while ((line = in.readLine()) != null) {result += line;}} catch (Exception e) {System.out.println("发送 POST 请求出现异常！"+e);e.printStackTrace();}//使用finally块来关闭输出流、输入流finally{try{if(out!=null){out.close();}if(in!=null){in.close();}}catch(IOException ex){ex.printStackTrace();}}return result;}static Channel channel_queAnomaly = null;public static void main(String[] args) {/** * 创建连接连接到MabbitMQ */ConnectionFactory factory = new ConnectionFactory();//设置MabbitMQ所在主机ip或者主机名factory.setHost(Hxyconfig.sendMQHostHXY); factory.setUsername(Hxyconfig.UserNameHXY);factory.setPassword(Hxyconfig.PassWordHXY);factory.setPort(Hxyconfig.portHXY); //创建一个连接Connection connection = null;try {connection = factory.newConnection();//创建一个频道channel_queAnomaly = connection.createChannel();//指定一个exchangeNamechannel_queAnomaly.exchangeDeclare(Hxyconfig.exchangeName, "topic"); } catch (IOException e1) {e1.printStackTrace();}// 定时请求接口数据，并且将返回的数据通过MQ消息队列发送到服务器Timer timer = new Timer();SendRabbit sr=new SendRabbit(); String urlHXYAnomaly="http://"+Hxyconfig.HostHXY+":8080/HxySys/realTime/HXYAnomaly.do";String urlhXYShipStatistic="http://"+Hxyconfig.HostHXY+":8080/HxySys/realTime/HXYShipStatistic.do";String urlhXYAISNetwork="http://"+Hxyconfig.HostHXY+":8080/HxySys/realTime/HXYAISNetwork.do";String urlhXYAISRadarFusion="http://"+Hxyconfig.HostHXY+":8080/HxySys/realTime/HXYAISRadarFusion.do";timer.schedule(new TimerTask() {public void run() {Date da=new Date(); System.out.println("Send RabbitMQ Start : "+da.toString()); //发送 GET 请求String hXYAnomaly=HttpRequest.sendGet(urlHXYAnomaly, "key=123&v=456"); String hXYShipStatistic=HttpRequest.sendGet(urlhXYShipStatistic, "key=123&v=456");String hXYAISNetwork=HttpRequest.sendGet(urlhXYAISNetwork, "key=123&v=456");String hXYAISRadarFusion=HttpRequest.sendGet(urlhXYAISRadarFusion, "key=123&v=456");try {if(hXYAnomaly.equals("")){ }else{// 该方法和本类中的方法一致sr.SendMQ2(channel_queAnomaly,hXYAnomaly,Hxyconfig.queAnomaly);}if(hXYAISRadarFusion.equals("")){}else{sr.SendMQ2(channel_queAnomaly,hXYAISRadarFusion,Hxyconfig.queRadarFusion);}sr.SendMQ2(channel_queAnomaly,hXYShipStatistic,Hxyconfig.queShipStatistic);sr.SendMQ2(channel_queAnomaly,hXYAISNetwork,Hxyconfig.queAISNetworkState);}catch (Exception e){e.printStackTrace();return;}}},0,1000*Hxyconfig.sendSecond);// 设定指定的时间time,此处为2000毫秒//发送 POST 请求/*String sr=HttpRequest.sendPost("/legoguide/server/product/query", "set_number=4-1&lang=zh-CN");System.out.println(sr);*/} public void SendMQ2(Channel channel,String jsStr,String routingKey) throws java.io.IOException{ //往队列中发出一条消息channel.basicPublish(Hxyconfig.exchangeName, routingKey, null,jsStr.getBytes());// System.out.println(jsStr);}}
