Browse Source

消息队列代码提交

fenghanchao 3 years ago
parent
commit
4e173db4f1

BIN
config.xls


+ 10 - 4
pom.xml

@@ -9,8 +9,8 @@
 		<maven.compiler.target>1.8</maven.compiler.target>
 	</properties>
 
-	<repositories>
-		<repository>
+	<!--<repositories>
+	&lt;!&ndash;	<repository>
 			<id>SagaCloud</id>
 			<name>SagaCloud</name>
 			<url>http://172.16.0.86:8081/repository/SagaCloud/</url>
@@ -20,7 +20,7 @@
 			<releases>
 				<enabled>true</enabled>
 			</releases>
-		</repository>
+		</repository>&ndash;&gt;
 		<repository>
 			<id>alimaven</id>
 			<name>aliyun maven</name>
@@ -35,7 +35,7 @@
 			<name>Sonatype Nexus Snapshots</name>
 			<url>http://oss.sonatype.org/content/repositories/snapshots</url>
 		</repository>
-	</repositories>
+	</repositories>-->
 
 	<dependencies>
 		<dependency>
@@ -86,6 +86,12 @@
 			<version>2.6.12</version>
 		</dependency>
 
+		<!--activemq配置-->
+		<dependency>
+			<groupId>org.apache.activemq</groupId>
+			<artifactId>activemq-core</artifactId>
+			<version>5.7.0</version>
+		</dependency>
 	</dependencies>
 
 	<build>

+ 90 - 0
src/main/java/com/saga/thread/common/LoadConfigThread.java

@@ -138,6 +138,12 @@ public class LoadConfigThread extends Thread {
 				excuteRulesBlackList(file);
 				excuteUploadMysql(file);
 				excuteUploadHbase(file);
+
+				/**»ñÈ¡ÏûÏ¢¶ÓÁÐbegin**/
+				executeActiveMQ(file);
+				executeActiveMqPointList(file);
+				/**»ñÈ¡ÏûÏ¢¶ÓÁÐend**/
+
 			}
 		} catch (Exception e) {
 			// TODO Auto-generated catch block
@@ -624,6 +630,90 @@ public class LoadConfigThread extends Thread {
 	}
 
 	/**
+	 * ¼ÓÔØactiveMQÅäÖÃ
+	 */
+	private static void executeActiveMQ(File file) {
+		// TODO Auto-generated method stub
+		int begin = 65535;
+		int end = 65535;
+		Map<String, Integer> pty = new HashMap<String, Integer>();
+		// Map<Integer, String> ptyValue = new HashMap<Integer, String>();
+
+		Map<Integer, Map<Integer, Map<Integer, String>>> excelData = ExcelUtil.readExcel(file);
+		for (int k = 0; k < excelData.size(); k++) {
+			Map<Integer, Map<Integer, String>> sheetData = excelData.get(k);
+			for (int i = 0; i < sheetData.size(); i++) {
+				Map<Integer, String> rowData = sheetData.get(i);
+				if ("ActiveMQ".equalsIgnoreCase(rowData.get(0))) {
+					begin = i;
+					// Constant.info("Db-Hbase begin: " + i);
+					for (int j = 1; j < rowData.size(); j++) {
+						pty.put(rowData.get(j), j);
+						// ptyValue.put(j, rowData.get(j));
+					}
+				} else if ((rowData.get(0).length() > 0) && !("ActiveMQ".equalsIgnoreCase(rowData.get(0)))) {
+					begin = 65535;
+					end = 65535;
+					pty = new HashMap<>();
+					// ptyValue = new HashMap<Integer, String>();
+					// Constant.info("Db-Hbase end: " + i);
+				} else if (i > begin && i < end) {
+					// Constant.info("Db-Hbase add: " + i);
+					if (rowData.get(1) == null || rowData.get(1).length() == 0) {
+						continue;
+					}
+					pty = Constant.toLower(pty);
+					Constant.activeMqBrokerUrl = rowData.get(pty.get("broker-url"));
+				}
+			}
+		}
+	}
+
+	/**
+	 * ¼ÓÔØactiveMQÅäÖÃ
+	 */
+	private static void executeActiveMqPointList(File file) {
+		int begin = 65535;
+		int end = 65535;
+		Map<String, Integer> pty = new HashMap<String, Integer>();
+
+		Map<Integer, Map<Integer, Map<Integer, String>>> excelData = ExcelUtil.readExcel(file);
+		for (int k = 0; k < excelData.size(); k++) {
+			Map<Integer, Map<Integer, String>> sheetData = excelData.get(k);
+			for (int i = 0; i < sheetData.size(); i++) {
+				Map<Integer, String> rowData = sheetData.get(i);
+				if ("ActiveMQ-PointList".equalsIgnoreCase(rowData.get(0))) {
+					begin = i;
+					// Constant.info("Rules-TransferList begin: " + i);
+					for (int j = 1; j < rowData.size(); j++) {
+						pty.put(rowData.get(j), j);
+						// ptyValue.put(j, rowData.get(j));
+					}
+				} else if ((rowData.get(0).length() > 0) && !("ActiveMQ-PointList".equalsIgnoreCase(rowData.get(0)))) {
+					begin = 65535;
+					end = 65535;
+					pty = new HashMap<String, Integer>();
+					// ptyValue = new HashMap<Integer, String>();
+					// Constant.info("Rules-TransferList end: " + i);
+				} else if (i > begin && i < end) {
+					// Constant.info("Rules-TransferList add: " + i);
+					if (rowData.get(1) == null || rowData.get(1).length() == 0) {
+						continue;
+					}
+
+					pty = Constant.toLower(pty);
+
+					String building = rowData.get(pty.get("building"));
+					String meter = rowData.get(pty.get("meter"));
+					String funcid = rowData.get(pty.get("funcid"));
+
+					Constant.activeMqPointSet.add(building + "-" + meter + "-" + funcid);
+				}
+			}
+		}
+	}
+
+	/**
 	 * ¼ÓÔØmysqlÅäÖÃ
 	 */
 	private static void excuteDbMysql(File file) {

+ 7 - 0
src/main/java/com/saga/thread/common/SaveThread.java

@@ -5,6 +5,8 @@ import java.util.Date;
 import java.util.Iterator;
 import java.util.List;
 
+import com.alibaba.fastjson.JSONObject;
+import com.saga.util.MessageUtil;
 import org.zillion.util.log.LogUtil;
 
 import com.saga.entity.Record;
@@ -179,6 +181,11 @@ public class SaveThread extends Thread {
 											record.receivetime = Constant.getOffSetTime(record);
 											record.funcID = Constant.getFunctionConv(record);
 
+											//TODO ÍùactiveMqÏûÏ¢¶ÓÁÐÀï·¢ËÍÏûÏ¢
+											if (Constant.activeMqPointSet.contains(record.buildingSign + "-" + record.meterSign + "-" + record.funcID)) {
+												MessageUtil.sendMessage(JSONObject.toJSONString(record));
+											}
+
 											if (toDB(record.buildingSign, record.meterSign, record.funcID)) {
 												this.strecordList.add(record);
 											}

+ 4 - 0
src/main/java/com/saga/util/Constant.java

@@ -112,6 +112,10 @@ public class Constant {
 	public static Set<String> saveTypeList = new HashSet<String>();
 	public static long delay = Constant.Millisecond.SECOND_1;
 
+
+	public static String activeMqBrokerUrl;
+	public static Set<String> activeMqPointSet = new HashSet<>();
+
 	// static {
 	// String resource = "\\config.json";
 	// String path = null;

+ 72 - 0
src/main/java/com/saga/util/MessageUtil.java

@@ -0,0 +1,72 @@
+package com.saga.util;
+
+import com.alibaba.fastjson.JSONObject;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.commons.lang.StringUtils;
+import org.springframework.util.CollectionUtils;
+
+import javax.jms.*;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @description:
+ * @author: fenghanchao
+ * @since: 2022/1/14 11:44
+ **/
+public class MessageUtil {
+
+    public static final String QUEUE_NAME = "collection.queue.20220114";
+
+    private static MessageProducer producer;
+
+    private static Session session;
+
+    public synchronized static Session getSession() throws Exception {
+        if (session != null) {
+            return session;
+        }
+
+        if (StringUtils.isBlank(Constant.activeMqBrokerUrl)) {
+            throw new Exception("读取不到activeMq的brokerUrl!");
+        }
+
+        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD
+                , "failover:(" + Constant.activeMqBrokerUrl + ")?initialReconnectDelay=1000&maxReconnectDelay=30000");
+        //创建jms连接
+        Connection connection = connectionFactory.createConnection();
+        connection.start();
+        //创建session,参数是否开启事务,签收模式(自动签收Session.AUTO_ACKNOWLEDGE)
+        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        return session;
+    }
+
+    public synchronized static MessageProducer getMessageProducer() throws Exception {
+        if (producer != null) {
+            return producer;
+        }
+
+        Session session = getSession();
+        //创建queue
+        Queue queue = session.createQueue(QUEUE_NAME);//Destination
+        //消息生产者
+        producer = session.createProducer(queue);
+        return producer;
+    }
+
+    public static void sendMessage(String message) throws Exception {
+        MessageProducer messageProducer = getMessageProducer();
+        TextMessage textMessage = getSession().createTextMessage(message);
+        messageProducer.send(textMessage);
+    }
+
+    public static void main(String[] args) throws Exception {
+        while (true) {
+            Map<String, Object> request = new HashMap<>();
+            request.put("key1", "value1");
+            sendMessage(JSONObject.toJSONString(request));
+            Thread.sleep(1000 * 5);
+        }
+
+    }
+}