Skip to content

P
property-management
Commit a7082f4b by zhangjw
Options

新增ActiveMQ订阅VSS事件并解析

parent 09cb220d
Showing with 254 additions and 7 deletions
jeecg-boot-parent/jeecg-boot-module-demo/pom.xml
...@@ -20,14 +20,60 @@ ...@@ -20,14 +20,60 @@
<groupId>com.chanjx</groupId> <groupId>com.chanjx</groupId>
<artifactId>chanjx-utils</artifactId> <artifactId>chanjx-utils</artifactId>
</dependency> </dependency>
<!--引入微服务启动依赖 starter
<!-- <dependency>-->
<!-- <groupId>org.springframework.boot</groupId>-->
<!-- <artifactId>spring-boot-starter-activemq</artifactId>-->
<!--&lt;!&ndash; <version>1.5.22.RELEASE</version>&ndash;&gt;-->
<!-- </dependency>-->
<!-- <dependency>-->
<!-- <groupId>org.apache.activemq</groupId>-->
<!-- <artifactId>activemq-pool</artifactId>-->
<!-- </dependency>-->
<dependency> <dependency>
<groupId>org.jeecgframework.boot</groupId> <groupId>org.apache.activemq</groupId>
<artifactId>jeecg-boot-starter-cloud</artifactId> <artifactId>activemq-core</artifactId>
<version>5.7.0</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.jeecgframework.boot</groupId> <groupId>com.google.protobuf</groupId>
<artifactId>jeecg-boot-starter-job</artifactId> <artifactId>protobuf-java</artifactId>
</dependency>--> <version>2.3.0</version>
</dependency>
</dependencies> </dependencies>
<build>
<extensions>
<extension>
<groupId>kr.motd.maven</groupId>
<artifactId>os-maven-plugin</artifactId>
<version>1.6.2</version>
</extension>
</extensions>
<plugins>
<plugin>
<groupId>org.xolstice.maven.plugins</groupId>
<artifactId>protobuf-maven-plugin</artifactId>
<version>0.6.1</version>
<configuration>
<protocArtifact>com.google.protobuf:protoc:3.12.0:exe:${os.detected.classifier}</protocArtifact>
<pluginId>grpc-java</pluginId>
<pluginArtifact>io.grpc:protoc-gen-grpc-java:1.32.1:exe:${os.detected.classifier}</pluginArtifact>
<protoSourceRoot>src/main/proto</protoSourceRoot>
<outputDirectory>src/main/java</outputDirectory>
<clearOutputDirectory>false</clearOutputDirectory>
</configuration>
<executions>
<execution>
<goals>
<goal>compile</goal>
<goal>compile-custom</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project> </project>
jeecg-boot-parent/jeecg-boot-module-demo/src/main/java/org/jeecg/modules/pm/handler/ActiveMqConfig.java 0 → 100644
//package org.jeecg.modules.pm.handler;
//
//import org.apache.activemq.ActiveMQConnectionFactory;
//import org.apache.activemq.command.ActiveMQTopic;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.context.annotation.Bean;
//import org.springframework.context.annotation.Configuration;
//import org.springframework.jms.annotation.EnableJms;
//import org.springframework.jms.listener.DefaultMessageListenerContainer;
//
//import javax.jms.ConnectionFactory;
//import javax.jms.Topic;
//
///**
// * @author: JJww
// * @Date:2022/3/25
// */
//@EnableJms
//@Configuration
//public class ActiveMqConfig {
//
//
// private final static String VSS_TOPIC = "openapi.vss.topic";
//
// private final static String EPS_TOPIC = "openapi.eps.topic";
//
// @Bean(name = "topic")
// public Topic topic() {
// return new ActiveMQTopic(VSS_TOPIC);
// }
//
// @Bean
// public ConnectionFactory connectionFactory() {
// return new ActiveMQConnectionFactory("tcp://172.16.16.2:61618");
// }
//
//
// @Bean
// public DefaultMessageListenerContainer defaultMessageListenerContainer(ConnectionFactory connectionFactory) {
// final DefaultMessageListenerContainer defaultMessageListenerContainer = new DefaultMessageListenerContainer();
// defaultMessageListenerContainer.setDestinationName(VSS_TOPIC);
// defaultMessageListenerContainer.setConnectionFactory(connectionFactory);
// defaultMessageListenerContainer.setMessageListener(vssMssageListener);
// defaultMessageListenerContainer.setPubSubDomain(true);
// return defaultMessageListenerContainer;
// }
//
//
// public ActiveMQConnectionFactory activeMQConnectionFactory(){
//
// }
//
//
//}
jeecg-boot-parent/jeecg-boot-module-demo/src/main/java/org/jeecg/modules/pm/handler/EventErrorHandler.java 0 → 100644
package org.jeecg.modules.pm.handler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.ErrorHandler;
/**
* @author: JJww
* @Date:2022/3/25
*/
@Slf4j
@Service
public class EventErrorHandler implements ErrorHandler {
@Override
public void handleError(Throwable t) {
log.error("Error in listener", t);
}
}
jeecg-boot-parent/jeecg-boot-module-demo/src/main/java/org/jeecg/modules/pm/handler/EventMessageListener.java 0 → 100644
//package org.jeecg.modules.pm.handler;
//
//
//import com.google.protobuf.ByteString;
//import lombok.extern.slf4j.Slf4j;
//import org.jeecg.modules.pm.entity.eps.EventDis;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.jms.annotation.JmsListener;
//import org.springframework.jms.core.JmsMessagingTemplate;
//import org.springframework.stereotype.Component;
//
//import javax.jms.BytesMessage;
//import javax.jms.Message;
//
///**
// * @author: JJww
// * @Date:2022/3/23
// */
//@Slf4j
//@Component
//public class EventMessageListener {
//
// @JmsListener(destination = "openapi.vss.topic", containerFactory = "topicListener")
// public void vssMessageListener(Message msg) throws Exception {
//
// try {
// // cms里发送的消息为BytesMessage，此处不做判断亦可
// if (msg instanceof BytesMessage) {
// BytesMessage bytesMessage = (BytesMessage) msg;
// long length = bytesMessage.getBodyLength();
// byte[] bt = new byte[(int) length];
// // 将BytesMessage转换为byte类型
// bytesMessage.readBytes(bt);
// // 壳文件字段，EventDis类为event_dis.proto文件解析而来，CommEventLog类为事件壳文件类
// EventDis.CommEventLog parseFrom = EventDis.CommEventLog.parseFrom(bt);
// // 输出壳文件字段
// // System.out.println(parseFrom.toString());
// // 扩展字段，此字段为设备上报事件内容，部分事件需要使用pb文件再次解析
// ByteString extInfo = parseFrom.getExtInfo();
// // 输出扩展字段
// // System.out.println(extInfo.toStringUtf8());
// }
// } catch (Exception e) {
// e.printStackTrace();
// }
// }
//
//}
jeecg-boot-parent/jeecg-boot-module-demo/src/main/java/org/jeecg/modules/pm/handler/VssMssageListener.java 0 → 100644
package org.jeecg.modules.pm.handler;
import com.google.protobuf.ByteString;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.jeecg.modules.pm.entity.eps.EventDis;
import javax.jms.*;
/**
* @author: JJww
* @Date:2022/3/28
*/
public class VssMssageListener {
/**
* 10.33.47.176:61618和openapi.vss.topic可通过openapi【事件订阅接口】获得
*/
public static final String BROKER_URL = "failover:(tcp://172.16.16.2:61618)?timeout=2000";
/**
* 10.33.47.176:61618和openapi.vss.topic可通过openapi【事件订阅接口】获得
*/
public static final String TARGET = "openapi.vss.topic";
public static void run() {
Connection connection = null;
Session session = null;
try {
// 创建链接工厂
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(BROKER_URL);
// 通过工厂创建一个连接
connection = factory.createConnection();
// factory.createConnection(userName, password)
// 启动连接
connection.start();
// 第一个参数表示是否使用事务，第二个参数指定消息的确认模式
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Topic topic = session.createTopic(TARGET);
MessageConsumer consumer = session.createConsumer(topic);
// 消费者异步接收topic里的消息
consumer.setMessageListener(new MessageListener() {
public void onMessage(Message msg) {
try {
// cms里发送的消息为BytesMessage，此处不做判断亦可
if (msg instanceof BytesMessage) {
BytesMessage bytesMessage = (BytesMessage) msg;
long length = bytesMessage.getBodyLength();
byte[] bt = new byte[(int) length];
// 将BytesMessage转换为byte类型
bytesMessage.readBytes(bt);
// 壳文件字段，EventDis类为event_dis.proto文件解析而来，CommEventLog类为事件壳文件类
EventDis.CommEventLog parseFrom = EventDis.CommEventLog.parseFrom(bt);
// 输出壳文件字段
System.out.println(parseFrom.toString());
// 扩展字段，此字段为设备上报事件内容，部分事件需要使用pb文件再次解析
ByteString extInfo = parseFrom.getExtInfo();
// 输出扩展字段
System.out.println(extInfo.toStringUtf8());
}
} catch (Exception e) {
e.printStackTrace();
}
}
});
System.out.println("ActiveMQ开始监听");
} catch (Exception e) {
e.printStackTrace();
}
}
}
\ No newline at end of file
jeecg-boot-parent/jeecg-boot-module-system/src/main/java/org/jeecg/JeecgSystemApplication.java
...@@ -2,6 +2,8 @@ package org.jeecg; ...@@ -2,6 +2,8 @@ package org.jeecg;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.util.oConvertUtils; import org.jeecg.common.util.oConvertUtils;
import org.jeecg.modules.pm.handler.VssMssageListener;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication; import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration; import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration;
...@@ -18,7 +20,7 @@ import java.net.UnknownHostException; ...@@ -18,7 +20,7 @@ import java.net.UnknownHostException;
*/ */
@Slf4j @Slf4j
@SpringBootApplication(exclude = {MongoAutoConfiguration.class}) @SpringBootApplication(exclude = {MongoAutoConfiguration.class})
public class JeecgSystemApplication extends SpringBootServletInitializer { public class JeecgSystemApplication extends SpringBootServletInitializer implements CommandLineRunner {
@Override @Override
protected SpringApplicationBuilder configure(SpringApplicationBuilder application) { protected SpringApplicationBuilder configure(SpringApplicationBuilder application) {
...@@ -31,6 +33,7 @@ public class JeecgSystemApplication extends SpringBootServletInitializer { ...@@ -31,6 +33,7 @@ public class JeecgSystemApplication extends SpringBootServletInitializer {
String ip = InetAddress.getLocalHost().getHostAddress(); String ip = InetAddress.getLocalHost().getHostAddress();
String port = env.getProperty("server.port"); String port = env.getProperty("server.port");
String path = oConvertUtils.getString(env.getProperty("server.servlet.context-path")); String path = oConvertUtils.getString(env.getProperty("server.servlet.context-path"));
VssMssageListener.run();
log.info("\n----------------------------------------------------------\n\t" + log.info("\n----------------------------------------------------------\n\t" +
"Application Jeecg-Boot is running! Access URLs:\n\t" + "Application Jeecg-Boot is running! Access URLs:\n\t" +
"Local: \t\thttp://localhost:" + port + path + "/\n\t" + "Local: \t\thttp://localhost:" + port + path + "/\n\t" +
...@@ -40,4 +43,8 @@ public class JeecgSystemApplication extends SpringBootServletInitializer { ...@@ -40,4 +43,8 @@ public class JeecgSystemApplication extends SpringBootServletInitializer {
} }
@Override
public void run(String... args) throws Exception {
VssMssageListener.run();
}
} }
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment