HuangBuZhangTeam
/
mq-active
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat():协议接收解析模块

master
Saisai Liu 2024-06-05 11:24:37 +08:00
commit 492d7c0453
17 changed files with 754 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
.gitignore vendored 100644
View File

@ -0,0 +1,38 @@
target/
!.mvn/wrapper/maven-wrapper.jar
!**/src/main/**/target/
!**/src/test/**/target/
### IntelliJ IDEA ###
.idea/modules.xml
.idea/jarRepositories.xml
.idea/compiler.xml
.idea/libraries/
*.iws
*.iml
*.ipr
### Eclipse ###
.apt_generated
.classpath
.factorypath
.project
.settings
.springBeans
.sts4-cache
### NetBeans ###
/nbproject/private/
/nbbuild/
/dist/
/nbdist/
/.nb-gradle/
build/
!**/src/main/**/build/
!**/src/test/**/build/
### VS Code ###
.vscode/
### Mac OS ###
.DS_Store

8
.idea/.gitignore vendored 100644
View File

@ -0,0 +1,8 @@
# 默认忽略的文件
/shelf/
/workspace.xml
# 基于编辑器的 HTTP 客户端请求
/httpRequests/
# Datasource local storage ignored files
/dataSources/
/dataSources.local.xml

8
.idea/encodings.xml 100644
View File

@ -0,0 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Encoding" defaultCharsetForPropertiesFiles="UTF-8">
<file url="file://$PROJECT_DIR$/src/main/java" charset="UTF-8" />
<file url="file://$PROJECT_DIR$/src/main/resources" charset="UTF-8" />
<file url="PROJECT" charset="UTF-8" />
</component>
</project>

13
.idea/misc.xml 100644
View File

@ -0,0 +1,13 @@
<project version="4">
<component name="ExternalStorageConfigurationManager" enabled="true" />
<component name="MavenProjectsManager">
<option name="originalFiles">
<list>
<option value="$PROJECT_DIR$/pom.xml" />
</list>
</option>
</component>
<component name="ProjectRootManager" version="2" languageLevel="JDK_17" default="true" project-jdk-name="17" project-jdk-type="JavaSDK">
<output url="file://$PROJECT_DIR$/out" />
</component>
</project>

7
.idea/vcs.xml 100644
View File

@ -0,0 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="$PROJECT_DIR$/../../" vcs="Git" />
<mapping directory="$PROJECT_DIR$" vcs="Git" />
</component>
</project>

93
pom.xml 100644
View File

@ -0,0 +1,93 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.mobai</groupId>
<artifactId>mq-active</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.7.15</version>
</parent>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-dependencies</artifactId>
<version>2.7.15</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba.fastjson2</groupId>
<artifactId>fastjson2</artifactId>
<version>2.0.46</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.83</version>
</dependency>
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.5</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- excel工具 -->
<dependency>
<groupId>org.apache.poi</groupId>
<artifactId>poi-ooxml</artifactId>
<version>4.1.2</version>
</dependency>
</dependencies>
</project>

19
src/main/java/com/mobai/MqttApplication.java 100644
View File

@ -0,0 +1,19 @@
package com.mobai;
import lombok.extern.java.Log;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @ClassName MqttApplication
* @Description 描述
* @Author SaiSai.Liu
* @Date 2024/5/31 14:33
*/
@Log
@SpringBootApplication
public class MqttApplication {
public static void main(String[] args) {
SpringApplication.run(MqttApplication.class,args);
}
}

44
src/main/java/com/mobai/api/MqttPublishSample.java 100644
View File

@ -0,0 +1,44 @@
package com.mobai.api;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class MqttPublishSample {
public static void main(String[] args) {
String topic = "MQTT Examples";
String content = "Message from MqttPublishSample";
int qos = 2;
String broker = "tcp://iot.eclipse.org:1883";
String clientId = "JavaSample";
MemoryPersistence persistence = new MemoryPersistence();
try {
MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
MqttConnectOptions connOpts = new MqttConnectOptions();
connOpts.setCleanSession(true);
System.out.println("Connecting to broker: "+broker);
sampleClient.connect(connOpts);
System.out.println("Connected");
System.out.println("Publishing message: "+content);
MqttMessage message = new MqttMessage(content.getBytes());
message.setQos(qos);
sampleClient.publish(topic, message);
System.out.println("Message published");
sampleClient.disconnect();
System.out.println("Disconnected");
System.exit(0);
} catch(MqttException me) {
System.out.println("reason "+me.getReasonCode());
System.out.println("msg "+me.getMessage());
System.out.println("loc "+me.getLocalizedMessage());
System.out.println("cause "+me.getCause());
System.out.println("excep "+me);
me.printStackTrace();
}
}
}

33
src/main/java/com/mobai/api/SubscribeSample.java 100644
View File

@ -0,0 +1,33 @@
package com.mobai.api;
import com.mobai.domian.GetOptions;
import com.mobai.domian.MqttCallBackServiceImpl;
import com.mobai.cofig.MqttFactory;
import com.mobai.cofig.MqttProperties;
import org.eclipse.paho.client.mqttv3.*;
public class SubscribeSample {
public static void main(String[] args) {
MqttProperties mqttProperties = MqttProperties.configBuild("39.98.69.92", "topic0");
// MqttProperties mqttProperties = new MqttProperties();
// mqttProperties.setBroker("tcp://39.98.69.92:1883");
// mqttProperties.setTopic("mqtt/test");
mqttProperties.setUsername("emqx");
mqttProperties.setPassword("public");
// mqttProperties.setClientid("subscribe_client");
int qos = 0;
try {
MqttClient client = new MqttFactory(new MqttCallBackServiceImpl()).buildOptions(mqttProperties);
// 连接参数
MqttConnectOptions options = GetOptions.getMqttOptions(mqttProperties);
// 设置回调
client.setCallback(new MqttCallBackServiceImpl());
// 进行连接
client.connect(options);
client.subscribe(mqttProperties.getTopic(), qos);
} catch (Exception e) {
e.printStackTrace();
}
}
}

44
src/main/java/com/mobai/cofig/MqttFactory.java 100644
View File

@ -0,0 +1,44 @@
package com.mobai.cofig;
import com.mobai.domian.GetOptions;
import com.mobai.domian.MqttCallBackServiceImpl;
import lombok.AllArgsConstructor;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Service;
/**
* @ClassName MqttFactory
* @Description 描述
* @Author SaiSai.Liu
* @Date 2024/5/31 11:35
*/
@Service
@AllArgsConstructor
public class MqttFactory {
private final MqttCallBackServiceImpl mqttCallBackService;
public MqttClient buildOptions( MqttProperties mqttProperties) {
MqttClient client = null;
try {
client = new MqttClient(
mqttProperties.getBroker(),
mqttProperties.getClientid(),
new MemoryPersistence());
MqttConnectOptions optionas = GetOptions.getMqttOptions(mqttProperties);
client.connect(optionas);
client.setCallback(mqttCallBackService);
client.subscribe(mqttProperties.getTopic(), 0);
} catch (MqttException e) {
throw new RuntimeException(e);
}
return client;
}
}

35
src/main/java/com/mobai/cofig/MqttProperties.java 100644
View File

@ -0,0 +1,35 @@
package com.mobai.cofig;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.UUID;
/**
* @ClassName MqttProperties
* @Description 描述
* @Author SaiSai.Liu
* @Date 2024/5/30 20:05
*/
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class MqttProperties {
private String broker ;
private String topic ;
private String username;
private String password;
private String clientid;
public static MqttProperties configBuild(String ip,String topic){
// "tcp://39.98.69.92:1883"
return MqttProperties.builder()
.broker("tcp://"+ip+":1883")
.topic(topic)
.clientid("mobai-mq")
.build();
}
}

43
src/main/java/com/mobai/cofig/RabbitConfig.java 100644
View File

@ -0,0 +1,43 @@
package com.mobai.cofig;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @ClassName RabbitConfig
* @Description 描述
* @Author SaiSai.Liu
* @Date 2024/5/31 21:47
*/
@Configuration
public class RabbitConfig {
@Bean
public Queue initQueue(){
return new Queue("create.topic",true);
}
/**
* 创建主题队列
* @return
*/
@Bean
public DirectExchange direct() {
return new DirectExchange("mobai-mq");
}
/**
* 绑定交换机
* @param direct
* @param initQueue
* @return
*/
@Bean
public Binding binding1a(DirectExchange direct,
Queue initQueue) {
return BindingBuilder.bind(initQueue)
.to(direct)
.with("mobai-mq");
}
}

29
src/main/java/com/mobai/domian/GetOptions.java 100644
View File

@ -0,0 +1,29 @@
package com.mobai.domian;
import com.mobai.cofig.MqttProperties;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
/**
* 获取Mqtt连接参数
* @ClassName GetOptions
* @Description 获取Mqtt连接参数
* @Author SaiSai.Liu
* @Date 2024/5/31 14:26
*/
public class GetOptions {
public static MqttConnectOptions getMqttOptions(MqttProperties mqttProperties){
// 连接参数
MqttConnectOptions options = new MqttConnectOptions();
if (!StringUtils.isAllBlank(mqttProperties.getUsername(),
mqttProperties.getPassword())){
options.setUserName(mqttProperties.getUsername());
options.setPassword(mqttProperties.getPassword().toCharArray());
}
options.setConnectionTimeout(60);
options.setKeepAliveInterval(60);
return options;
}
}

34
src/main/java/com/mobai/domian/MqttCallBackServiceImpl.java 100644
View File

@ -0,0 +1,34 @@
package com.mobai.domian;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.stereotype.Service;
/**
* 回执消息类
* @ClassName MqttCallBackServiceImpl
* @Description 回执消息类
* @Author SaiSai.Liu
* @Date 2024/5/30 20:02
*/
@Service
public class MqttCallBackServiceImpl implements MqttCallback {
@Override
public void connectionLost(Throwable cause) {
System.out.println("connectionLost: " + cause.getMessage());
}
@Override
public void messageArrived(String topic, MqttMessage message) {
System.out.println("topic: " + topic);
System.out.println("Qos: " + message.getQos());
System.out.println("message content: " + new String(message.getPayload()));
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
System.out.println("deliveryComplete---------" + token.isComplete());
}
}

38
src/main/java/com/mobai/rabbitMq/MessageHandler.java 100644
View File

@ -0,0 +1,38 @@
package com.mobai.rabbitMq;
import com.mobai.cofig.MqttFactory;
import com.mobai.cofig.MqttProperties;
import lombok.extern.log4j.Log4j2;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* 消费者：消息处理器
* @ClassName MessageHandler
* @Description 描述
* @Author SaiSai.Liu
* @Date 2024/5/31 14:37
*/
@Log4j2
@Component
public class MessageHandler {
@Autowired
private MqttFactory mqttFactory;
@RabbitListener(queues = {"create.topic" })
private void message(String msg){
log.info("消息内容：{}",msg);
MqttProperties topic0 = MqttProperties.configBuild(
"39.98.69.92",
"topic0");
log.info("接收到消息：{}",topic0);
MqttClient client = mqttFactory.buildOptions(topic0);
log.info("client创建:{}",client);
log.info("clientID创建:{}",client.getClientId());
}
}

261
src/main/java/com/mobai/util/RedisService.java 100644
View File

@ -0,0 +1,261 @@
package com.mobai.util;
import org.apache.poi.ss.formula.functions.T;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.redis.core.BoundSetOperations;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;
import java.util.*;
import java.util.concurrent.TimeUnit;
/**
* spring redis 工具类
*
* @author muyu
**/
@SuppressWarnings(value = {"unchecked", "rawtypes"})
@Component
@Lazy
public class RedisService {
@Autowired
public RedisTemplate redisTemplate;
/**
* 缓存基本的对象，Integer、String、实体类等
*
* @param key 缓存的键值
* @param value 缓存的值
*/
public <T> void setCacheObject(final String key, final T value) {
redisTemplate.opsForValue().set(key, value);
}
/**
* 缓存基本的对象，Integer、String、实体类等
*
* @param key 缓存的键值
* @param value 缓存的值
* @param timeout 时间
* @param timeUnit 时间颗粒度
*/
public <T> void setCacheObject(final String key, final T value, final Long timeout, final TimeUnit timeUnit) {
redisTemplate.opsForValue().set(key, value, timeout, timeUnit);
}
/**
* 设置有效时间
*
* @param key Redis键
* @param timeout 超时时间
* @return true=设置成功；false=设置失败
*/
public boolean expire(final String key, final long timeout) {
return expire(key, timeout, TimeUnit.SECONDS);
}
/**
* 设置有效时间
*
* @param key Redis键
* @param timeout 超时时间
* @param unit 时间单位
* @return true=设置成功；false=设置失败
*/
public boolean expire(final String key, final long timeout, final TimeUnit unit) {
return redisTemplate.expire(key, timeout, unit);
}
/**
* 获取有效时间
*
* @param key Redis键
* @return 有效时间
*/
public long getExpire(final String key) {
return redisTemplate.getExpire(key);
}
/**
* 判断 key是否存在
*
* @param key 键
* @return true 存在 false不存在
*/
public Boolean hasKey(String key) {
return redisTemplate.hasKey(key);
}
/**
* 获得缓存的基本对象。
*
* @param key 缓存键值
* @return 缓存键值对应的数据
*/
public <T> T getCacheObject(final String key) {
ValueOperations<String, T> operation = redisTemplate.opsForValue();
return operation.get(key);
}
/**
* 删除单个对象
*
* @param key
*/
public boolean deleteObject(final String key) {
return redisTemplate.delete(key);
}
/**
* 删除集合对象
*
* @param collection 多个对象
* @return
*/
public boolean deleteObject(final Collection collection) {
return redisTemplate.delete(collection) > 0;
}
/**
* 缓存List数据
*
* @param key 缓存的键值
* @param dataList 待缓存的List数据
* @return 缓存的对象
*/
public <T> long setCacheList(final String key, final List<T> dataList) {
Long count = redisTemplate.opsForList().rightPushAll(key, dataList);
return count == null ? 0 : count;
}
/**
* 获得缓存的list对象
*
* @param key 缓存的键值
* @return 缓存键值对应的数据
*/
public <T> List<T> getCacheList(final String key) {
return redisTemplate.opsForList().range(key, 0, -1);
}
/**
* 缓存Set
*
* @param key 缓存键值
* @param dataSet 缓存的数据
* @return 缓存数据的对象
*/
public <T> BoundSetOperations<String, T> setCacheSet(final String key, final Set<T> dataSet) {
BoundSetOperations<String, T> setOperation = redisTemplate.boundSetOps(key);
Iterator<T> it = dataSet.iterator();
while (it.hasNext()) {
setOperation.add(it.next());
}
return setOperation;
}
/**
* 获得缓存的set
*
* @param key
* @return
*/
public <T> Set<T> getCacheSet(final String key) {
return redisTemplate.opsForSet().members(key);
}
/**
* 缓存Map
*
* @param key
* @param dataMap
*/
public <T> void setCacheMap(final String key, final Map<String, T> dataMap) {
if (dataMap != null) {
redisTemplate.opsForHash().putAll(key, dataMap);
}
}
/**
* 获得缓存的Map
*
* @param key
* @return
*/
public <T> Map<String, T> getCacheMap(final String key) {
return redisTemplate.opsForHash().entries(key);
}
/**
* 往Hash中存入数据
*
* @param key Redis键
* @param hKey Hash键
* @param value 值
*/
public <T> void setCacheMapValue(final String key, final String hKey, final T value) {
redisTemplate.opsForHash().put(key, hKey, value);
}
/**
* 获取Hash中的数据
*
* @param key Redis键
* @param hKey Hash键
* @return Hash中的对象
*/
public <T> T getCacheMapValue(final String key, final String hKey) {
HashOperations<String, String, T> opsForHash = redisTemplate.opsForHash();
return opsForHash.get(key, hKey);
}
/**
* 获取多个Hash中的数据
*
* @param key Redis键
* @param hKeys Hash键集合
* @return Hash对象集合
*/
public <T> List<T> getMultiCacheMapValue(final String key, final Collection<Object> hKeys) {
return redisTemplate.opsForHash().multiGet(key, hKeys);
}
/**
* 删除Hash中的某条数据
*
* @param key Redis键
* @param hKey Hash键
* @return 是否成功
*/
public boolean deleteCacheMapValue(final String key, final String hKey) {
return redisTemplate.opsForHash().delete(key, hKey) > 0;
}
/**
* 获得缓存的基本对象列表
*
* @param pattern 字符串前缀
* @return 对象列表
*/
public Collection<String> keys(final String pattern) {
return redisTemplate.keys(pattern);
}
/**
* 存入一个集合
* @param key
* @param t
* @param <V>
*/
public <V extends List<?>> void setCacheList(String key, T t) {
redisTemplate.opsForList().leftPush(String.valueOf(key),t);
}
public <T> long setCacheList(final String key, final T dataList) {
Long count = redisTemplate.opsForList().rightPushAll(key, dataList);
return count == null ? 0 : count;
}
}

7
src/main/resources/application.yml 100644
View File

@ -0,0 +1,7 @@
server:
port: 8082
spring:
application:
name: mobai-mq
rabbitmq:
host: 43.142.100.73