zianxin/pj_service_cloud
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat：添加mqtt服务

Browse Source
This commit is contained in:
dzc committed 2025-01-13 16:54:37 +08:00
1 parent fdf6dc8072
commit 82ed9bd2cc
26 files changed
+906 -6
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pjcloud-common/pjcloud-common-core/src/main/java/com/bsd/pjcloud/common/core/constant/ServiceNameConstants.java
+5
View File
@@ -86,4 +86,9 @@ public interface ServiceNameConstants {
* 硬件模块
*/
String HARDWARE_SERVICE = "pjcloud-hardware-biz";
/**
* mqtt模块
*/
String MQTT_SERVICE = "pjcloud-mqtt-biz";
}
pjcloud-hardware/pjcloud-hardware-biz/pom.xml
+6
View File
@@ -127,6 +127,12 @@
<version>2.1.0</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>com.bsd</groupId>
<artifactId>pjcloud-mqtt-api</artifactId>
<version>5.7.0</version>
<scope>compile</scope>
</dependency>
</dependencies>
<profiles>
<profile>
pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/common/config/MqttConfig.java → pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/mqtt/config/MqttConfig.java
+1 -1
View File
@@ -1,4 +1,4 @@
package com.bsd.pjcloud.hard.common.config;
package com.bsd.pjcloud.hard.mqtt.config;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/service/impl/HardManagerServiceImpl.java
+5 -1
View File
@@ -16,8 +16,9 @@ import com.bsd.pjcloud.hard.api.entity.HardManagerEntity;
import com.bsd.pjcloud.hard.api.vo.HardManagerVo;
import com.bsd.pjcloud.hard.mapper.HardManagerMapper;
import com.bsd.pjcloud.hard.service.HardManagerService;
import com.bsd.pjcloud.hard.common.config.MqttConfig;
import com.bsd.pjcloud.hard.mqtt.config.MqttConfig;
import com.bsd.pjcloud.hard.utils.MqTopicUtil;
import com.bsd.pjcloud.mqtt.api.feign.RemoteMqttEventService;
import com.bsd.pjcloud.wallet.api.feign.RemoteMchService;
import com.bsd.pjcloud.wallet.api.utils.WalletDataUtil;
import com.bsd.pjcloud.wallet.api.vo.MchManageVo;
@@ -46,6 +47,7 @@ public class HardManagerServiceImpl extends ServiceImpl<HardManagerMapper, HardM
private final RemoteMchService mchService;
private final RemoteDeptService deptService;
private final RemoteMqttEventService remoteMqttEventService;
@Override
public Page<HardManagerVo> pageJoin(Page page, HardManagerDto dto) {
@@ -99,6 +101,8 @@ public class HardManagerServiceImpl extends ServiceImpl<HardManagerMapper, HardM
throw new RuntimeException("设备不存在");
}
baseMapper.updateById(dto);
// 更新mq
remoteMqttEventService.sendConfigChangeMsg(detail.getDeptId());
}
@Override
pjcloud-mqtt/.editorconfig
+18
View File
@@ -0,0 +1,18 @@
# 开发组IDE 编辑器标准 V2.1.0
root = true
[*.{adoc, bat, groovy, html, java, js, jsp, kt, kts, md, properties, py, rb, sh, sql, svg, txt, xml, xsd}]
charset = utf-8
[*.{groovy, java, kt, kts, xml, xsd}]
indent_style = tab
indent_size = 4
continuation_indent_size = 8
end_of_line = lf
[*.{js, html}]
indent_style = space
indent_size = 2
end_of_line = lf
insert_final_newline = true
trim_trailing_whitespace = true
pjcloud-mqtt/pjcloud-mqtt-api/pom.xml
+42
View File
@@ -0,0 +1,42 @@
<?xml version="1.0"?>
<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>
<parent>
<groupId>com.bsd</groupId>
<artifactId>pjcloud-mqtt</artifactId>
<version>5.7.0</version>
</parent>
<artifactId>pjcloud-mqtt-api</artifactId>
<dependencies>
<!-- 连表查询注解 -->
<dependency>
<groupId>com.github.yulichang</groupId>
<artifactId>mybatis-plus-join-annotation</artifactId>
</dependency>
<!--core 工具类-->
<dependency>
<groupId>com.bsd</groupId>
<artifactId>pjcloud-common-core</artifactId>
</dependency>
<!--mybatis plus spring,包含了mybatis plus core-->
<dependency>
<groupId>com.baomidou</groupId>
<artifactId>mybatis-plus-spring</artifactId>
</dependency>
<!--feign 工具类-->
<dependency>
<groupId>com.bsd</groupId>
<artifactId>pjcloud-common-feign</artifactId>
</dependency>
<!-- excel 导入导出 -->
<dependency>
<groupId>com.bsd</groupId>
<artifactId>pjcloud-common-excel</artifactId>
</dependency>
</dependencies>
</project>
pjcloud-mqtt/pjcloud-mqtt-api/src/main/java/com/bsd/pjcloud/mqtt/api/feign/RemoteMqttEventService.java
+62
View File
@@ -0,0 +1,62 @@
package com.bsd.pjcloud.mqtt.api.feign;
import com.bsd.pjcloud.common.core.constant.ServiceNameConstants;
import com.bsd.pjcloud.common.core.util.R;
import com.bsd.pjcloud.common.feign.annotation.NoToken;
import com.bsd.pjcloud.mqtt.dto.MqttBaseDto;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;
import java.util.List;
@FeignClient(contextId = "remoteMqttEventService", value = ServiceNameConstants.MQTT_SERVICE)
public interface RemoteMqttEventService {
@NoToken
@GetMapping("/mqtt/send-default-msg")
R<Boolean> sendDefaultMsg(@RequestBody MqttBaseDto dto);
/**
* 设置变化
*/
@NoToken
@GetMapping("/mqtt/send-config-change-msg")
R<Boolean> sendConfigChangeMsg(@RequestParam Long mchDeptId);
/**
* 食堂商品变化
*/
@NoToken
@GetMapping("/mqtt/send-canteen-product-change-msg")
R<Boolean> sendCanteenProductChangeMsg(@RequestParam Long mchDeptId);
/**
* 食堂分类变化
*/
@NoToken
@GetMapping("/mqtt/send-canteen-classify-change-msg")
R<Boolean> sendCanteenClassifyChangeMsg(@RequestParam Long mchDeptId);
/**
* 超市商品变化
*/
@NoToken
@GetMapping("/mqtt/send-supermarket-product-change-msg")
R<Boolean> sendSupermarketProductChangeMsg(@RequestParam Long mchDeptId);
/**
* 超市分类变化
*/
@NoToken
@GetMapping("/mqtt/send-supermarket-classify-change-msg")
R<Boolean> sendSupermarketClassifyChangeMsg(@RequestParam Long mchDeptId);
/**
* 人脸变化
*/
@NoToken
@GetMapping("/mqtt/send-face-change-msg")
R<Boolean> sendFaceChangeMsg(@RequestParam Long mchDeptId);
}
pjcloud-mqtt/pjcloud-mqtt-api/src/main/java/com/bsd/pjcloud/mqtt/constant/MqttActionEnum.java
+21
View File
@@ -0,0 +1,21 @@
package com.bsd.pjcloud.mqtt.constant;
import lombok.AllArgsConstructor;
import lombok.Getter;
/**
* mq action
*/
@AllArgsConstructor
@Getter
public enum MqttActionEnum {
CONFIG("设置变化"),
CANTEEN_PRODUCT("食堂商品变化"),
CANTEEN_CLASSIFY("食堂分类变化"),
SUPERMARKET_PRODUCT("超市商品变化"),
SUPERMARKET_CLASSIFY("超市分类变化"),
PRINT("打印变化"),
FACE("人脸变化"),
;
private final String description;
}
pjcloud-mqtt/pjcloud-mqtt-api/src/main/java/com/bsd/pjcloud/mqtt/dto/MqttBaseDto.java
+35
View File
@@ -0,0 +1,35 @@
package com.bsd.pjcloud.mqtt.dto;
import cn.hutool.core.util.ObjectUtil;
import com.bsd.pjcloud.mqtt.constant.MqttActionEnum;
import lombok.Data;
@Data
public class MqttBaseDto<T> {
// 商户id
private Long merchantDeptId;
// 硬件id
private Long hardwareId;
private MqttActionEnum action;
private Integer qos = 1;
private T data;
public Boolean isExistMchDeptId() {
return ObjectUtil.isNotEmpty(merchantDeptId);
}
public String getTopic() {
if (ObjectUtil.isAllNotEmpty(merchantDeptId, hardwareId)) {
return String.format("topic/android/%s/%s", merchantDeptId, hardwareId);
}
if (ObjectUtil.isAllNotEmpty(merchantDeptId)) {
return String.format("topic/android/%s", merchantDeptId);
}
return "topic/android";
}
}
pjcloud-mqtt/pjcloud-mqtt-api/src/main/java/com/bsd/pjcloud/mqtt/dto/MqttHardBaseMsg.java
+14
View File
@@ -0,0 +1,14 @@
package com.bsd.pjcloud.mqtt.dto;
import com.bsd.pjcloud.mqtt.constant.MqttActionEnum;
import lombok.Data;
/**
* 向硬件发送消息
*/
@Data
public class MqttHardBaseMsg {
private MqttActionEnum action;
}
pjcloud-mqtt/pjcloud-mqtt-api/src/main/resources/META-INF/spring/org.springframework.cloud.openfeign.FeignClient.imports
+1
View File
@@ -0,0 +1 @@
com.bsd.pjcloud.mqtt.api.feign.RemoteMqttEventService
pjcloud-mqtt/pjcloud-mqtt-biz/pom.xml
+119
View File
@@ -0,0 +1,119 @@
<?xml version="1.0"?>
<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>
<parent>
<groupId>com.bsd</groupId>
<artifactId>pjcloud-mqtt</artifactId>
<version>5.7.0</version>
</parent>
<artifactId>pjcloud-mqtt-biz</artifactId>
<dependencies>
<!--mqtt-->
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
</dependency>
<!--必备: 注册中心客户端-->
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>
<!--必备: 配置中心客户端-->
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
</dependency>
<!--必备: 操作数据源相关-->
<dependency>
<groupId>com.bsd</groupId>
<artifactId>pjcloud-common-data</artifactId>
</dependency>
<!--必备：pigx安全模块-->
<dependency>
<groupId>com.bsd</groupId>
<artifactId>pjcloud-common-security</artifactId>
</dependency>
<!--必备：xss 过滤模块-->
<dependency>
<groupId>com.bsd</groupId>
<artifactId>pjcloud-common-xss</artifactId>
</dependency>
<!--必备: sentinel 依赖-->
<dependency>
<groupId>com.bsd</groupId>
<artifactId>pjcloud-common-sentinel</artifactId>
</dependency>
<!--必备: feign 依赖-->
<dependency>
<groupId>com.bsd</groupId>
<artifactId>pjcloud-common-feign</artifactId>
</dependency>
<!--必备: 依赖api模块-->
<dependency>
<groupId>com.bsd</groupId>
<artifactId>pjcloud-mqtt-api</artifactId>
<version>5.7.0</version>
</dependency>
<!--必备: log 依赖-->
<dependency>
<groupId>com.bsd</groupId>
<artifactId>pjcloud-common-log</artifactId>
</dependency>
<!--必备: undertow容器-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-undertow</artifactId>
</dependency>
<!--必备: spring boot web-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!--测试: spring boot test-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
</dependency>
</dependencies>
<profiles>
<profile>
<id>cloud</id>
<activation>
<activeByDefault>true</activeByDefault>
</activation>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<executions>
<execution>
<goals>
<goal>repackage</goal>
</goals>
<configuration>
<loaderImplementation>CLASSIC</loaderImplementation>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>io.fabric8</groupId>
<artifactId>docker-maven-plugin</artifactId>
<configuration>
<skip>false</skip>
</configuration>
</plugin>
</plugins>
</build>
</profile>
<profile>
<id>boot</id>
</profile>
</profiles>
</project>
pjcloud-mqtt/pjcloud-mqtt-biz/src/main/java/com/bsd/pjcloud/mqtt/MqttApp.java
+17
View File
@@ -0,0 +1,17 @@
package com.bsd.pjcloud.mqtt;
import com.bsd.pjcloud.common.feign.annotation.EnablePjcloudFeignClients;
import com.bsd.pjcloud.common.security.annotation.EnablePjcloudResourceServer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
@EnablePjcloudFeignClients
@EnableDiscoveryClient
@EnablePjcloudResourceServer
@SpringBootApplication
public class MqttApp {
public static void main(String[] args) {
SpringApplication.run(MqttApp.class, args);
}
}
pjcloud-mqtt/pjcloud-mqtt-biz/src/main/java/com/bsd/pjcloud/mqtt/config/MqttConfig.java
+43
View File
@@ -0,0 +1,43 @@
package com.bsd.pjcloud.mqtt.config;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
@Data
@ConfigurationProperties(prefix = "mqtt")
@Component
public class MqttConfig {
/**
* 是否使用本地配置文件
*/
private Boolean isLocal = false;
/**
* 证书地址
*/
private String crtPath;
/**
* 用户名
*/
private String username;
/**
* 密码
*/
private String password;
/**
* 连接地址
*/
private String hostUrl;
/**
* clientId
*/
private String clientId;
/**
* 安卓连接地址
*/
private String androidAddress;
/**
* 安卓连接端口
*/
private String androidPort;
}
pjcloud-mqtt/pjcloud-mqtt-biz/src/main/java/com/bsd/pjcloud/mqtt/controller/MqttEventController.java
+101
View File
@@ -0,0 +1,101 @@
package com.bsd.pjcloud.mqtt.controller;
import com.bsd.pjcloud.common.core.util.R;
import com.bsd.pjcloud.common.security.annotation.Inner;
import com.bsd.pjcloud.mqtt.constant.MqttActionEnum;
import com.bsd.pjcloud.mqtt.dto.MqttBaseDto;
import com.bsd.pjcloud.mqtt.service.MqttEventService;
import io.swagger.v3.oas.annotations.security.SecurityRequirement;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.RequiredArgsConstructor;
import org.springframework.http.HttpHeaders;
import org.springframework.web.bind.annotation.*;
@RestController
@RequiredArgsConstructor
@RequestMapping("/mqtt")
@Tag(description = "hard-invoice", name = "发票记录管理")
@SecurityRequirement(name = HttpHeaders.AUTHORIZATION)
public class MqttEventController {
private final MqttEventService mqttEventService;
@Inner
@PostMapping("/send-default-msg")
public R<Boolean> sendDefaultMsg(@RequestBody MqttBaseDto dto) {
return R.ok(mqttEventService.defaultAction(dto));
}
/**
* 设置变化
*/
@Inner
@GetMapping("/send-config-change-msg")
public R<Boolean> sendConfigChangeMsg(@RequestParam Long mchDeptId) {
MqttBaseDto dto = new MqttBaseDto<>();
dto.setAction(MqttActionEnum.CONFIG);
dto.setMerchantDeptId(mchDeptId);
return R.ok(mqttEventService.defaultAction(dto));
}
/**
* 食堂商品变化
*/
@Inner
@GetMapping("/send-canteen-product-change-msg")
public R<Boolean> sendCanteenProductChangeMsg(@RequestParam Long mchDeptId) {
MqttBaseDto dto = new MqttBaseDto<>();
dto.setMerchantDeptId(mchDeptId);
dto.setAction(MqttActionEnum.CANTEEN_PRODUCT);
return R.ok(mqttEventService.defaultAction(dto));
}
/**
* 食堂分类变化
*/
@Inner
@GetMapping("/send-canteen-classify-change-msg")
public R<Boolean> sendCanteenClassifyChangeMsg(@RequestParam Long mchDeptId) {
MqttBaseDto dto = new MqttBaseDto<>();
dto.setMerchantDeptId(mchDeptId);
dto.setAction(MqttActionEnum.CANTEEN_CLASSIFY);
return R.ok(mqttEventService.defaultAction(dto));
}
/**
* 超市商品变化
*/
@Inner
@GetMapping("/send-supermarket-product-change-msg")
public R<Boolean> sendSupermarketProductChangeMsg(@RequestParam Long mchDeptId) {
MqttBaseDto dto = new MqttBaseDto<>();
dto.setMerchantDeptId(mchDeptId);
dto.setAction(MqttActionEnum.SUPERMARKET_PRODUCT);
return R.ok(mqttEventService.defaultAction(dto));
}
/**
* 超市分类变化
*/
@Inner
@GetMapping("/send-supermarket-classify-change-msg")
public R<Boolean> sendSupermarketClassifyChangeMsg(@RequestParam Long mchDeptId) {
MqttBaseDto dto = new MqttBaseDto<>();
dto.setMerchantDeptId(mchDeptId);
dto.setAction(MqttActionEnum.SUPERMARKET_CLASSIFY);
return R.ok(mqttEventService.defaultAction(dto));
}
/**
* 人脸变化
*/
@Inner
@GetMapping("/send-face-change-msg")
public R<Boolean> sendFaceChangeMsg(@RequestParam Long mchDeptId) {
MqttBaseDto dto = new MqttBaseDto<>();
dto.setMerchantDeptId(mchDeptId);
dto.setAction(MqttActionEnum.FACE);
return R.ok(mqttEventService.defaultAction(dto));
}
}
pjcloud-mqtt/pjcloud-mqtt-biz/src/main/java/com/bsd/pjcloud/mqtt/inner/MqttInner.java
+21
View File
@@ -0,0 +1,21 @@
package com.bsd.pjcloud.mqtt.inner;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import com.bsd.pjcloud.mqtt.service.MqttService;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;
@Component
@RequiredArgsConstructor
public class MqttInner {
private final MqttService mqttService;
@PostConstruct
public void initMqtt() {
mqttService.connect();
mqttService.subscribeAll();
}
}
pjcloud-mqtt/pjcloud-mqtt-biz/src/main/java/com/bsd/pjcloud/mqtt/service/MqttCallBackService.java
+54
View File
@@ -0,0 +1,54 @@
package com.bsd.pjcloud.mqtt.service;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import jakarta.annotation.Resource;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets;
@RequiredArgsConstructor
@Slf4j
@Component
public class MqttCallBackService implements MqttCallback {
static final String TAG = "mqtt回调";
@Lazy
private final MqttService mqttService;
// 断开连接的回调
@Override
public void connectionLost(Throwable e) {
log.debug("mqtt断开连接:{}", JSON.toJSON(e));
// 重新连接
mqttService.connect();
}
// 监听消息的回调
@Override
public void messageArrived(String topic, MqttMessage msg) throws Exception {
log.debug("mqtt收到消息topic:{};msg:{};", topic, JSON.toJSON(msg));
if (ObjectUtil.isEmpty(topic)) return;
if (!topic.contains("service")) return;
byte[] payload = msg.getPayload();
String json = new String(payload, StandardCharsets.UTF_8);
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
try {
log.info("|{}|mqtt证书校验成功:|{}|", TAG, token.getTopics());
} catch (Exception e) {
log.error("|{}|mqtt证书校验失败:|{}|", TAG, JSON.toJSON(e));
}
}
}
pjcloud-mqtt/pjcloud-mqtt-biz/src/main/java/com/bsd/pjcloud/mqtt/service/MqttEventService.java
+23
View File
@@ -0,0 +1,23 @@
package com.bsd.pjcloud.mqtt.service;
import com.alibaba.fastjson.JSON;
import com.bsd.pjcloud.mqtt.dto.MqttBaseDto;
import com.bsd.pjcloud.mqtt.dto.MqttHardBaseMsg;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
@RequiredArgsConstructor
@Slf4j
@Component
public class MqttEventService {
private final MqttService mqttService;
public <T extends MqttBaseDto> Boolean defaultAction(T dto) {
MqttHardBaseMsg msg = new MqttHardBaseMsg();
msg.setAction(dto.getAction());
mqttService.publish(dto.getQos(), false, dto.getTopic(), JSON.toJSONString(msg));
return true;
}
}
pjcloud-mqtt/pjcloud-mqtt-biz/src/main/java/com/bsd/pjcloud/mqtt/service/MqttService.java
+123
View File
@@ -0,0 +1,123 @@
package com.bsd.pjcloud.mqtt.service;
import cn.hutool.core.util.ObjectUtil;
import com.bsd.pjcloud.mqtt.config.MqttConfig;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import com.bsd.pjcloud.mqtt.utils.SSLUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Component;
import org.springframework.validation.annotation.Validated;
import java.util.Objects;
@Validated
@Slf4j
@RequiredArgsConstructor
@Component
public class MqttService {
static final String TAG = "mqtt服务";
private final MqttConfig mqttConfig;
private final MqttCallBackService mqttCallBack;
static private MqttClient client;
private static MqttClient getClient() {
return client;
}
private static void setClient(MqttClient client) {
MqttService.client = client;
}
public void connect() {
MqttClient client;
if (ObjectUtil.isNotEmpty(MqttService.getClient()) && MqttService.getClient().isConnected()) return;
try {
client = new MqttClient(mqttConfig.getHostUrl(), mqttConfig.getClientId(), new MemoryPersistence());
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(true);
options.setUserName(mqttConfig.getUsername());
options.setPassword(mqttConfig.getPassword().toCharArray());
crtAuth(options);
client.setCallback(mqttCallBack);
client.connect(options);
if (!client.isConnected()) {
log.error("|{}|mqtt连接失败;", TAG);
} else {
log.info("|{}|mqtt连接成功;", TAG);
}
setClient(client);
} catch (Exception e) {
log.error("mqtt连接错误:{};", e.getMessage());
}
}
public void crtAuth(MqttConnectOptions options) {
try {
String path;
if (!mqttConfig.getIsLocal() && ObjectUtil.isNotEmpty(mqttConfig.getCrtPath())) {
path = mqttConfig.getCrtPath();
} else {
path = getClass().getClassLoader().getResource("crt/emqxsl-ca.crt").getPath();
}
options.setSocketFactory(SSLUtils.getSocketFactory(path));
} catch (Exception e) {
log.error("mqtt 证书错误 {}", e.getMessage());
}
}
private boolean checkMqttConnect() {
if (Objects.isNull(MqttService.getClient()) || !client.isConnected()) {
log.error("|{}|mqtt连接错误", TAG);
return false;
}
return true;
}
// 订阅所有主题
public void subscribeAll() {
if (!checkMqttConnect()) return;
String topic = "topic/#";
subscribe(topic, 1);
}
// 发送消息
public void publish(int qos, boolean retained, String topic, String pushMessage) {
if (!checkMqttConnect()) return;
MqttMessage message = new MqttMessage();
message.setQos(qos);
message.setRetained(retained);
message.setPayload(pushMessage.getBytes());
MqttTopic mTopic = MqttService.getClient().getTopic(topic);
MqttDeliveryToken token;
try {
token = mTopic.publish(message);
token.waitForCompletion();
publishLog(qos, retained, topic, pushMessage);
} catch (MqttException e) {
log.error("|{}|mqtt发送消息错误:|{}|", TAG, e.getMessage());
}
}
public void subscribe(String topic, int qos) {
try {
MqttService.getClient().subscribe(topic, qos);
subscribeLog(topic, qos);
} catch (MqttException e) {
log.error("|{}|mqtt订阅主题错误:|{}|", TAG, e.getMessage());
}
}
private void publishLog(int qos, boolean retained, String topic, String pushMessage) {
log.info("|{}|发布消息成功topic:|{}|qos:|{}|是否保留:|{}|消息:|{}|", TAG, topic, qos, retained, pushMessage);
}
private void subscribeLog(String topic, int qos) {
log.debug("|{}|订阅主题成功topic:{};qos:{};", TAG, topic, qos);
}
}
pjcloud-mqtt/pjcloud-mqtt-biz/src/main/java/com/bsd/pjcloud/mqtt/utils/SSLUtils.java
+33
View File
@@ -0,0 +1,33 @@
package com.bsd.pjcloud.mqtt.utils;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManagerFactory;
import java.io.FileInputStream;
import java.security.KeyStore;
import java.security.cert.Certificate;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
public class SSLUtils {
public static SSLSocketFactory getSocketFactory(String caCertPath) throws Exception {
CertificateFactory certFactory = CertificateFactory.getInstance("X.509");
// load CA certificate into keystore to authenticate server
Certificate caCert = certFactory.generateCertificate(new FileInputStream(caCertPath));
X509Certificate x509CaCert = (X509Certificate) caCert;
KeyStore caKeyStore = KeyStore.getInstance(KeyStore.getDefaultType());
caKeyStore.load(null, null);
caKeyStore.setCertificateEntry("cacert", x509CaCert);
TrustManagerFactory tmFactory = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
tmFactory.init(caKeyStore);
SSLContext sslContext = SSLContext.getInstance("TLSv1.2");
sslContext.init(null, tmFactory.getTrustManagers(), null);
return sslContext.getSocketFactory();
}
}
pjcloud-mqtt/pjcloud-mqtt-biz/src/main/resources/application.yml
+18
View File
@@ -0,0 +1,18 @@
server:
port: 8010
spring:
application:
name: @artifactId@
cloud:
nacos:
username: @nacos.username@
password: @nacos.password@
discovery:
server-addr: ${NACOS_HOST:pjcloud-register}:${NACOS_PORT:8848}
config:
server-addr: ${spring.cloud.nacos.discovery.server-addr}
config:
import:
- optional:nacos:application-@profiles.active@.yml
- optional:nacos:${spring.application.name}-@profiles.active@.yml
pjcloud-mqtt/pjcloud-mqtt-biz/src/main/resources/crt/apiclient_key.pem
+28
View File
@@ -0,0 +1,28 @@
-----BEGIN PRIVATE KEY-----
MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC/dA1s6RUO8ZMy
0exwGp34jgcUY5//dbeXusXmkwqLPO24jtBjfglVoOQF9PpUWFsk8UY2pOP26sq5
pXrCpLYA7nuROVk4ulmU/JDfsi8YK0DpDMT8neAO/uy+kXoIGogW7scVz9ZNkKFI
9g4YBjh8plNb0KaT3piYgxmSccVV9vwWDvFoVFZkHqCCY6PScKm2v2WnINNgXbuO
X3PBjSJKJ/AnaONiwp7F67+LE1meBu+m1vCg+pDry8s2H2vxf3TlzXygG32owsru
bqj/aN5+wr3YcheX1L2TQcTeGOMNoylD/0lzMi+az6IMNM2sV/BOdKbVq2/g238Z
SyFM0xQTAgMBAAECggEAKbgXSNlNrnDz24olFJSoQcZpQocnEVrMg/K+Ny1picfg
IWx1eoi3bgJeTfIBLqoGSnmNDFFIH8EwJd4qbwnHY1KUeKPfjl4FMVMMGsLQQlsC
jQYsUsPwVDQ20JwQ1H07lIu5tEYS89oFW0nK7m9Eld0EoyhHdEd4RxuiUega0U1d
R2TdILJsA2WT/StRuthmbpR/0PJ8GwLF294ARP+ibmk4H1ZuMShf5kHnFrLG1+0Z
g/kfQQfXMNIbLK/i4KxGh3IdaTxFl0ITQq86tX+pZp1d36f1Evnsb3PxdCymBHM8
/TFlZ6llpjN/JLWTr6t93CpYm7DGwcvg99Pada1QKQKBgQDhRGd6+x9pwWrowoC+
wnerxXYe1/MecMXzsVptX7/cj9dieqQnwgsbEGt8Heb3Zbh/IheFEB3LHB4tC80n
1WgVi78OY3fbaEBSnwrkdbYxoiSn8FBoi6NkzitbcAQhzEMu/1GM3XQ9RAD9bNpg
E2Pb/USvO4dd66uGD4UbnQFUzQKBgQDZkq2Htw05QREfcAVVD4EUnoFub3wPXP83
MJMoNU+6v1J/boURiUJ1DVwHZlP9f+K2++hlDleURSQypj37UgupyQGZJS4K+847
x1K/vRW8U6+Ey8syMd+B8F8xVnPwJdYMOOmQCbnoYSXYsY2U4vv5pkcq2gWfuTtS
+k6BRscMXwKBgGhqtrSwtozk1NNzZQ4HQkVi0C/X1Z+kkn01Fj9ghKSOID5Lrs+n
CFLRN6FekvLWOcDkYpaRM94z+GDU+ZGfQBHMpxDaVnJjTn19ryTa00sKZ4dYw0zS
fTh6wGov8BpifjnIq1pqRDkpznOqA0dJdZ5BNZcsNsD4PpdzljWIf1d5AoGAPgBi
my47AnncW00c9+R3MF0LHWF/xREHWb7p8KdglSM4a0Afg6PN4HVyKFmszazH+sVG
sKWpp4MckaJ2P0xeLoAtiGrj+BVR7A8dMa68pvJiLkG1IhS8UNcjCXhNib3+hXUi
ef2LkQE+O8Y9RpR4oegUl82RlLbAhSD3hjeZOZkCgYEAsPHXxe4r3KpXBoWhkZIs
Xd00tYeTzzN1WulIIM6RUx3pNfy+bAne9fEieTZ3frw+b2Hbo+bUgS40hn8gdaxX
HJfwXTsnJ9yw5PnJ8VAk+Te07G2RczbJwl6G4vEifBAZaYqbuK37n7EwQdl6P2Db
5Tg5eEAXoE4XnOWPj5ivIMc=
-----END PRIVATE KEY-----
pjcloud-mqtt/pjcloud-mqtt-biz/src/main/resources/crt/emqxsl-ca.crt
+22
View File
@@ -0,0 +1,22 @@
-----BEGIN CERTIFICATE-----
MIIDrzCCApegAwIBAgIQCDvgVpBCRrGhdWrJWZHHSjANBgkqhkiG9w0BAQUFADBh
MQswCQYDVQQGEwJVUzEVMBMGA1UEChMMRGlnaUNlcnQgSW5jMRkwFwYDVQQLExB3
d3cuZGlnaWNlcnQuY29tMSAwHgYDVQQDExdEaWdpQ2VydCBHbG9iYWwgUm9vdCBD
QTAeFw0wNjExMTAwMDAwMDBaFw0zMTExMTAwMDAwMDBaMGExCzAJBgNVBAYTAlVT
MRUwEwYDVQQKEwxEaWdpQ2VydCBJbmMxGTAXBgNVBAsTEHd3dy5kaWdpY2VydC5j
b20xIDAeBgNVBAMTF0RpZ2lDZXJ0IEdsb2JhbCBSb290IENBMIIBIjANBgkqhkiG
9w0BAQEFAAOCAQ8AMIIBCgKCAQEA4jvhEXLeqKTTo1eqUKKPC3eQyaKl7hLOllsB
CSDMAZOnTjC3U/dDxGkAV53ijSLdhwZAAIEJzs4bg7/fzTtxRuLWZscFs3YnFo97
nh6Vfe63SKMI2tavegw5BmV/Sl0fvBf4q77uKNd0f3p4mVmFaG5cIzJLv07A6Fpt
43C/dxC//AH2hdmoRBBYMql1GNXRor5H4idq9Joz+EkIYIvUX7Q6hL+hqkpMfT7P
T19sdl6gSzeRntwi5m3OFBqOasv+zbMUZBfHWymeMr/y7vrTC0LUq7dBMtoM1O/4
gdW7jVg/tRvoSSiicNoxBN33shbyTApOB6jtSj1etX+jkMOvJwIDAQABo2MwYTAO
BgNVHQ8BAf8EBAMCAYYwDwYDVR0TAQH/BAUwAwEB/zAdBgNVHQ4EFgQUA95QNVbR
TLtm8KPiGxvDl7I90VUwHwYDVR0jBBgwFoAUA95QNVbRTLtm8KPiGxvDl7I90VUw
DQYJKoZIhvcNAQEFBQADggEBAMucN6pIExIK+t1EnE9SsPTfrgT1eXkIoyQY/Esr
hMAtudXH/vTBH1jLuG2cenTnmCmrEbXjcKChzUyImZOMkXDiqw8cvpOp/2PV5Adg
06O/nVsJ8dWO41P0jmP6P6fbtGbfYmbW0W5BjfIttep3Sp+dWOIrWcBAI+0tKIJF
PnlUkiaY4IBIqDfv8NZ5YBberOgOzW6sRBc4L0na4UU+Krk2U886UAb3LujEV0ls
YSEY1QSteDwsOoBrp+uvFRTp2InBuThs4pFsiv9kuXclVzDAGySj4dzp30d8tbQk
CAUw7C29C79Fv1C5qfPrmAESrciIxpg0X40KPMbp1ZWVbd4=
-----END CERTIFICATE-----
pjcloud-mqtt/pjcloud-mqtt-biz/src/main/resources/logback-spring.xml
+70
View File
@@ -0,0 +1,70 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
小技巧: 在根pom里面设置统一存放路径，统一管理方便维护
<properties>
<log-path>/Users/lengleng</log-path>
</properties>
1. 其他模块加日志输出，直接copy本文件放在resources 目录即可
2. 注意修改 <property name="${log-path}/log.path" value=""/> 的value模块
-->
<configuration debug="false" scan="false">
<property name="log.path" value="logs/${project.artifactId}"/>
<!-- 彩色日志格式 -->
<property name="CONSOLE_LOG_PATTERN"
value="${CONSOLE_LOG_PATTERN:-%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}"/>
<!-- 彩色日志依赖的渲染类 -->
<conversionRule conversionWord="clr" converterClass="org.springframework.boot.logging.logback.ColorConverter"/>
<conversionRule conversionWord="wex"
converterClass="org.springframework.boot.logging.logback.WhitespaceThrowableProxyConverter"/>
<conversionRule conversionWord="wEx"
converterClass="org.springframework.boot.logging.logback.ExtendedWhitespaceThrowableProxyConverter"/>
<!-- Console log output -->
<appender name="console" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>${CONSOLE_LOG_PATTERN}</pattern>
</encoder>
</appender>
<!-- Log file debug output -->
<appender name="debug" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${log.path}/debug.log</file>
<rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<fileNamePattern>${log.path}/%d{yyyy-MM, aux}/debug.%d{yyyy-MM-dd}.%i.log.gz</fileNamePattern>
<maxFileSize>50MB</maxFileSize>
<maxHistory>30</maxHistory>
</rollingPolicy>
<encoder>
<pattern>%date [%thread] %-5level [%logger{50}] %file:%line - %msg%n</pattern>
</encoder>
</appender>
<!-- Log file error output -->
<appender name="error" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${log.path}/error.log</file>
<rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<fileNamePattern>${log.path}/%d{yyyy-MM}/error.%d{yyyy-MM-dd}.%i.log.gz</fileNamePattern>
<maxFileSize>50MB</maxFileSize>
<maxHistory>30</maxHistory>
</rollingPolicy>
<encoder>
<pattern>%date [%thread] %-5level [%logger{50}] %file:%line - %msg%n</pattern>
</encoder>
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>ERROR</level>
</filter>
</appender>
<logger name="org.activiti.engine.impl.db" level="DEBUG">
<appender-ref ref="debug"/>
</logger>
<!--nacos 心跳 INFO 屏蔽-->
<logger name="com.alibaba.nacos" level="OFF">
<appender-ref ref="error"/>
</logger>
<!-- Level: FATAL 0 ERROR 3 WARN 4 INFO 6 DEBUG 7 -->
<root level="INFO">
<appender-ref ref="console"/>
<appender-ref ref="debug"/>
</root>
</configuration>
pjcloud-mqtt/pom.xml
+21
View File
@@ -0,0 +1,21 @@
<?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>
<parent>
<groupId>com.bsd</groupId>
<artifactId>pjcloud</artifactId>
<version>5.7.0</version>
</parent>
<artifactId>pjcloud-mqtt</artifactId>
<packaging>pom</packaging>
<!--项目子模块-->
<modules>
<module>pjcloud-mqtt-api</module>
<module>pjcloud-mqtt-biz</module>
</modules>
</project>
pom.xml
+3 -4
View File
@@ -16,9 +16,7 @@
~ this software without specific prior written permission.
~ Author: pjcloud
~
-->
<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">
--><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.bsd</groupId>
@@ -125,7 +123,8 @@
<module>pjcloud-wallet</module>
<module>pjcloud-supermarket</module>
<module>pjcloud-hardware</module>
</modules>
<module>pjcloud-mqtt</module>
</modules>
<dependencyManagement>
<dependencies>