物联网亿万级通信一站式解决方案EMQ

/**
 * Created by 传智播客*黑马程序员.
 */
@RestController
@RequestMapping("/mqtt")
public class AuthController {
    
    private Logger log = LoggerFactory.getLogger(AuthController.class);
    
    private Map<String,String> users;
    
    @PostConstruct
    public void init(){
        users = new HashMap<>();
        users.put("user","123456");//实际的密码应该是密文,mqtt的http认证组件传输过来的密码是明文，我们需要自己进行加密验证
        users.put("emq-client2","123456");
        users.put("emq-client3","123456");
    }
    
    @PostMapping("/auth")
    public ResponseEntity<?> auth(@RequestParam("clientid") String clientid, 
                                  @RequestParam("username") String username, 
                                  @RequestParam("password") String password){
        log.info("emqx认证组件调用自定义的认证服务开始认证,clientid={},username={},password={}",clientid,username,password);
        //在此处可以进行复杂也的认证逻辑,但是我们为了演示方便做一个固定操作
        String value = users.get(username);
        if(StringUtils.isEmpty(value)){
            return new ResponseEntity<Object>(HttpStatus.UNAUTHORIZED);
        }
        if(!value.equals(password)){
            return new ResponseEntity<Object>(HttpStatus.UNAUTHORIZED);
        }
        return new ResponseEntity<Object>(HttpStatus.OK);
    }
}

@PostMapping("/superuser")
public ResponseEntity<?> superUser(@RequestParam("clientid") String clientid,
                                   @RequestParam("username") String username){
    log.info("emqx 查询是否是超级用户,clientid={},username={}",clientid,username);
    if(clientid.contains("admin") || username.contains("admin")){
        log.info("用户{}是超级用户",username);
        //是超级用户
        return new ResponseEntity<Object>(HttpStatus.OK);
    }else {
         log.info("用户{}不是超级用户",username);
        //不是超级用户
        return new ResponseEntity<Object>(HttpStatus.UNAUTHORIZED);
    }
}

@PostConstruct
public void init(){
    users = new HashMap<>();
    users.put("user","123456");
    users.put("emq-client2","123456");
    users.put("emq-client3","123456");
    users.put("admin","admin");
}

@PostMapping("/acl")
public ResponseEntity acl(@RequestParam("access")int access,
                          @RequestParam("username")String username,
                          @RequestParam("clientid")String clientid,
                          @RequestParam("ipaddr")String ipaddr,
                          @RequestParam("topic")String topic,
                          @RequestParam("mountpoint")String mountpoint){

    log.info("EMQX发起客户端操作授权查询请求,access={},username={},clientid={},ipaddr={},topic={},mountpoint={}",
             access,username,clientid,ipaddr,topic,mountpoint);

    if(username.equals("emq-client2") && topic.equals("testtopic/#") && access == 1){
        log.info("客户端{}有权限订阅{}",username,topic);
        return new ResponseEntity<>(HttpStatus.OK);
    }

    if(username.equals("emq-client3") && topic.equals("testtopic/123") && access == 2){
        log.info("客户端{}有权限向{}发布消息",username,topic);
        return new ResponseEntity<>(null, HttpStatus.OK);
    }

    log.info("客户端{},username={},没有权限对主题{}进行{}操作",clientid,username,topic,access==1?"订阅":"发布");
    return new ResponseEntity(HttpStatus.UNAUTHORIZED);//无权限
}

package com.lkd.http.controller;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.lkd.entity.VendingMachineEntity;
import com.lkd.feignService.StatusService;
import com.lkd.service.VendingMachineService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.common.Strings;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/acl")
@RequiredArgsConstructor
@Slf4j
public class AclController {
    private final VendingMachineService vmService;
    
    /**
     * 连接控制
     * @param username
     * @param password
     * @param clientid
     * @return
     */
    @PostMapping("/auth")
    public ResponseEntity<?> auth(@RequestParam(value = "username",required = false,defaultValue = "") String username,
                                  @RequestParam(value = "password",required = false,defaultValue = "") String password,
                                  @RequestParam(value = "clientid",required = false,defaultValue = "") String clientid){

        if(Strings.isNullOrEmpty(clientid)){
            return new ResponseEntity<>(null, HttpStatus.BAD_REQUEST);
        }
        if(clientid.startsWith("mqtt")){
            return new ResponseEntity<>(null, HttpStatus.OK);
        }
        log.info("client request connect,clientId:"+clientid);
        //服务器端连接
        if(clientid.startsWith("monitor")){
            return new ResponseEntity<>(null, HttpStatus.OK);
        }
        var qw = new LambdaQueryWrapper<VendingMachineEntity>();
        qw.eq(VendingMachineEntity::getClientId,clientid);
        var vm = vmService.getOne(qw);
        if(vm == null){
            return new ResponseEntity<>(null, HttpStatus.BAD_REQUEST);
        }
        return new ResponseEntity<>(null, HttpStatus.OK);
    }
}

/**
 * 判断是否是超级用户
 * @param clientid
 * @param username
 * @return
 */
@PostMapping("/superuser")
public ResponseEntity<?> superuser(@RequestParam String clientid,@RequestParam String username){
    log.info("超级用户认证");
    if(clientid.startsWith("monitor")){
        log.info(clientid+"是超级用户!!!!");
        return new ResponseEntity<>(null, HttpStatus.OK);
    }
    if(clientid.startsWith("mqtt")){
        log.info(clientid+"是超级用户!!!!");
        return new ResponseEntity<>(null, HttpStatus.OK);
    }
    log.info(clientid+"不是超级用户!!!!");
    return new ResponseEntity<>(null, HttpStatus.BAD_REQUEST);
}

/**
 * 发布订阅控制
 * @param clientid
 * @param topic
 * @param access 动作2：publish、1：subscribe
 * @param username
 * @return
 */
@PostMapping("/pubsub")
public ResponseEntity<?> pubsub(@RequestParam String clientid,
                             @RequestParam String topic,
                             @RequestParam int access,
                             @RequestParam String username,
                             @RequestParam String ipaddr,
                             @RequestParam String mountpoint){
    System.out.println("acl clientid:"+clientid+" username:"+username+" ipaddr:"+ipaddr+" topic:"+topic+" access:"+access+" mountpoint:"+mountpoint);
 log.info("pubsub!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");
    if(Strings.isNullOrEmpty(clientid)){
        return new ResponseEntity<>(null, HttpStatus.BAD_REQUEST);
    }
    var qw = new LambdaQueryWrapper<VendingMachineEntity>();
    qw.eq(VendingMachineEntity::getClientId,clientid);
    var vm = vmService.getOne(qw);
    if(vm == null){
        return new ResponseEntity<>(null, HttpStatus.BAD_REQUEST);
    }
    if(access == 2 && topic.equals("server/"+vm.getInnerCode())){
        return new ResponseEntity<>(null, HttpStatus.OK);
    }
    if(access == 1 && topic.equals("vm/"+vm.getInnerCode())){
        return new ResponseEntity<>(null, HttpStatus.OK);
    }
    log.info(clientid+"不能发布或订阅消息!!!!");
    return new ResponseEntity<>(null, HttpStatus.BAD_REQUEST);
}

package com.lkd.emq;

import lombok.Data;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import java.util.List;

@Configuration
@Component
@Data
@Slf4j
public class MqttConfig {
    @Value("${mqtt.client.username}")
    private String username;
    @Value("${mqtt.client.password}")
    private String password;
    @Value("${mqtt.client.serverURI}")
    private String serverURI;
    @Value("${mqtt.client.clientId}")
    private String clientId;
    @Value("${mqtt.client.keepAliveInterval}")
    private int keepAliveInterval;
    @Value("${mqtt.client.connectionTimeout}")
    private int connectionTimeout;

    @Bean
    public MqttClient mqttClient() {
        try {
            MqttClientPersistence persistence = mqttClientPersistence();
            MqttClient client = new MqttClient(serverURI, clientId, persistence);
            client.setManualAcks(true);
            client.connect(mqttConnectOptions());
            return client;
        } catch (MqttException e) {
            log.error("emq connect error",e);
            return null;
        }
    }

    @Bean
    public MqttConnectOptions mqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setCleanSession(true);
        options.setAutomaticReconnect(true);
        options.setConnectionTimeout(connectionTimeout);
        options.setKeepAliveInterval(keepAliveInterval);
        options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);

        return options;
    }

    public MqttClientPersistence mqttClientPersistence() {
        return new MemoryPersistence();
    }
}

package com.lkd.emq;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.lkd.utils.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class MqttProducer {

    @Value("${mqtt.producer.defaultQos}")
    private int defaultProducerQos;
    @Value("${mqtt.producer.defaultRetained}")
    private boolean defaultRetained;
    @Value("${mqtt.producer.defaultTopic}")
    private String defaultTopic;

    @Autowired
    private MqttClient mqttClient;

    public void send(String payload) {
        this.send(defaultTopic, payload);
    }

    public void send(String topic, String payload) {
        this.send(topic, defaultProducerQos, payload);
    }

    public void send(String topic, int qos, String payload) {
        this.send(topic, qos, defaultRetained, payload);
    }

    public void send(String topic, int qos, boolean retained, String payload) {
        try {
            mqttClient.publish(topic, payload.getBytes(), qos, retained);
        } catch (MqttException e) {
            log.error("publish msg error.",e);
        }
    }

    public <T extends Object> void send(String topic, int qos, T msg) throws JsonProcessingException {
        String payload = JsonUtil.serialize(msg);
        this.send(topic,qos,payload);
    }
}

@RunWith(SpringRunner.class)
@SpringBootTest
public class TestEMQ {
    @Autowired
    private MqttProducer mqttProducer;

    @Test
    public void publish(){
        mqttProducer.send("lkdtest","{'name':'abc'}");
    }
}

package com.lkd.emq;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
@Slf4j
public class MqttCallback implements MqttCallbackExtended {
    //需要订阅的topic配置
    @Value("${mqtt.consumer.consumerTopics}")
    private List<String> consumerTopics;

    @Autowired
    private MqttService mqttService;

    @Autowired
    private MqttClient mqttClient;

    @Override
    public void connectionLost(Throwable throwable) {
        log.error("emq error.",throwable);
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        //处理消息
        mqttService.processMessage(topic, message);
        //处理成功后确认消息
        mqttClient.messageArrivedComplete(message.getId(),message.getQos());
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.info("deliveryComplete---------" + iMqttDeliveryToken.isComplete());
    }


    @Override
    public void connectComplete(boolean b, String s) {
        //和EMQ连接成功后根据配置自动订阅topic
        if(consumerTopics != null && consumerTopics.size() > 0){
            consumerTopics.forEach(t->{
                try {
                        log.info(">>>>>>>>>>>>>>subscribe topic:"+t);
                        mqttClient.subscribe(t, 2);
                    } catch (MqttException e) {
                        log.error("emq connect error", e);
                    }
            });
        }
    }

    public void setMqttClient(MqttClient mqttClient) {
        this.mqttClient = mqttClient;
    }
}

    @Autowired
    private MqttCallback mqttCallback;//添加这句

    @Bean
    public MqttClient mqttClient() {
        try {
            MqttClientPersistence persistence = mqttClientPersistence();
            MqttClient client = new MqttClient(serverURI, clientId, persistence);
            client.setManualAcks(true);
            mqttCallback.setMqttClient(client);//添加这句
            client.setCallback(mqttCallback);//添加这句
            client.connect(mqttConnectOptions());
            return client;
        } catch (MqttException e) {
            log.error("emq connect error",e);
            return null;
        }
    }

/**
 * 处理商品配置请求
 */
@Component
@ProcessType(value = "skuCfgReq")
public class SkuCfgMsgHandler implements MsgHandler{

    @Override
    public void process(String jsonMsg) throws IOException {
        //.......
    }
}

/**
 * 消息处理接口
 */
public interface MsgHandler{
    void process(String jsonMsg) throws IOException;
}

package com.lkd.annotations;
import java.lang.annotation.*;
/**
 * 协议处理类型注解
 */
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE})
public @interface ProcessType{
    String value();
}

private ApplicationContext ctx;
private Map<String, MsgHandler> handlerMap = Maps.newHashMap();

@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
    ctx = applicationContext;
    Map<String,MsgHandler> map = ctx.getBeansOfType(MsgHandler.class);
    map.values().stream().forEach(v->{
        String msgType = v.getClass().getAnnotation(ProcessType.class).value();
        handlerMap.put(msgType,v);
    });
}

public MsgHandler getMsgHandler(String msgType){
    return handlerMap.get(msgType);
}

package com.lkd.emq;

import org.eclipse.paho.client.mqttv3.MqttMessage;

public interface MqttService {
    void processMessage(String topic, MqttMessage message);
}

package com.lkd.emq;

import com.google.common.base.Strings;
import com.lkd.business.MsgHandler;
import com.lkd.business.MsgHandlerContext;
import com.lkd.utils.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
@Slf4j
public class MqttServiceImpl implements MqttService{
    @Autowired
    private MsgHandlerContext msgHandlerContext;

    /**
     * mqtt消息处理
     * @param topic
     * @param message
     */
    @Override
    public void processMessage(String topic, MqttMessage message) {
        String msgContent = new String(message.getPayload());
        log.info("接收到消息:"+msgContent);
        try {
            String msgType = JsonUtil.getValueByNodeName("msgType",msgContent);
            if(Strings.isNullOrEmpty(msgType)) return;
            MsgHandler msgHandler = msgHandlerContext.getMsgHandler(msgType);
            if(msgHandler == null)return;
            msgHandler.process(msgContent);
        } catch (IOException e) {
            log.error("process msg error,msg is: "+msgContent,e);
        }
    }
}

package com.lkd.client.config;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;

@Data
@Configuration
@ConfigurationProperties("emq")
public class EmqConfig {
    private String mqttServerUrl;
    private String mqttPassword;
    private String innerCode;
    private String publisTopicPrefix;
    private String clientId;
    private String subscribeTopic;

}

package com.lkd.client.emq;

import com.lkd.client.config.EmqConfig;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class EmqClient {
    @Autowired
    private EmqConfig emqConfig;

    private MqttClient mqttClient;
    
    
    /**
     * 连接mqtt borker
     */
    public void connect(MqttCallback callback){
        MemoryPersistence persistence = new MemoryPersistence();
        try {
            mqttClient = new MqttClient(emqConfig.getMqttServerUrl(),emqConfig.getClientId(),persistence);
        } catch (MqttException e) {
            log.error("mqtt creat error",e);
        }
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setAutomaticReconnect(false);
        connOpts.setUserName("monitor");
        connOpts.setPassword(emqConfig.getMqttPassword().toCharArray());
        connOpts.setCleanSession(false);
        try {
            mqttClient.setCallback(callback);
            mqttClient.connect(connOpts);
        } catch (MqttException e) {
            log.error("mqtt creat error",e);
        }
    }

  /**
     * 订阅及回调方法
     * @param topicFilter
     * @throws MqttException
     */
    public void subscribe(String topicFilter,MqttCallback callBack) throws MqttException     {
        log.info("subscribe----------- {}",topicFilter);
        mqttClient.subscribe(topicFilter,2);
        mqttClient.setCallback(callBack);
    }


   /**
     * 发布消息
     * @param msgType
     * @param msg
     */
    @Async
    public void publish(String msgType,String msg){
        try {
            MqttMessage mqttMessage = new MqttMessage(msg.getBytes());
            mqttMessage.setQos(0);
            mqttMessage.setRetained(false);
            log.info("publish topic {}",emqConfig.getPublisTopicPrefix()+msgType);
            mqttClient.getTopic(emqConfig.getPublisTopicPrefix()+msgType).publish(mqttMessage);
        } catch (MqttException e) {
            log.error("mqtt publish msg error",e);
        }
    }

}

package com.lkd.client.emq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONPath;
import com.lkd.client.config.EmqConfig;
import com.lkd.client.config.MsgType;
import com.lkd.client.emq.msg.*;
import com.lkd.client.mapper.VersionMapper;
import com.lkd.client.pojo.Version;
import com.lkd.client.service.DataProcessService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * emq回调类
 */
@Component
@Slf4j
public class EmqMsgCallback implements MqttCallbackExtended {
    @Autowired
    private EmqClient emqClient;

    @Autowired
    private  EmqConfig config;


    @Override
    public void connectionLost(Throwable throwable) {
        log.info("emq connect lost",throwable);
        try {
           TimeUnit.SECONDS.sleep(10);
           //连接中断，从新建立连接
           emqClient.connect(this);
           //订阅服务器通知topic
            emqClient.subscribe(config.getSubscribeTopic(),this);
        } catch (Exception e) {
          log.info("订阅失败",e);
        }
    }

    //接受服务器通知
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
       CompletableFuture.runAsync(()->{
            System.out.println(topic);
            String payload = new String(mqttMessage.getPayload());

            String msgType = (String) JSONPath.read(payload, "$.msgType");
            log.info("************msgType {} payload {}*************", msgType,payload);
        });

    }

  

    //发送消息完成后本地回调方法
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
      
    }

    //连接服务器成功后本地回调方法
    @Override
    public void connectComplete(boolean connectComplete, String s) {
        if(!connectComplete){
            return;
        }
        //连接成功后，订阅服务器topic
        try {
            log.info("connectComplete subscribe {} ",config.getSubscribeTopic());
            emqClient.subscribe(config.getSubscribeTopic(),this);
        } catch (MqttException e) {
            log.info("connectComplete",e);
        }
    }
    
    
}

package com.lkd.client.emq;

import com.alibaba.fastjson.JSON;
import com.lkd.client.config.ApplicationContextRegister;
import com.lkd.client.config.EmqConfig;
import com.lkd.client.emq.msg.VersionReq;
import com.lkd.client.mapper.VersionMapper;
import com.lkd.client.pojo.Version;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.core.io.Resource;
import org.springframework.jdbc.datasource.init.ScriptUtils;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;



@Slf4j
@Service
public class EmqInitServcie {
  
    @Autowired
    private EmqMsgCallback emqMsgCallback;

    @Autowired
    private EmqClient emqClient;


    @PostConstruct
    public void init() throws Exception {
        //建立mqtt连接
        emqClient.connect(emqMsgCallback);
    }
}

package com.lkd.client.config;
import lombok.Getter;

@Getter
public enum MsgType {
    versionCfg("versionCfg","同步基础信息"),
    channelCfg("channelCfg","同步货道信息"),
    skuCfg("skuCfgResp","同步商品信息"),
    skuPrice("skuPrice","价格变动"),
    supplyResp("supplyResp","价格变动"),
    vendoutReq("vendoutReq","出货通知"),
    vendoutResp("vendoutResp","出货上报");

    private String type;
    private String desc;

    MsgType(String type, String desc) {
        this.type = type;
        this.desc = desc;
    }
}

package com.lkd.client.emq.msg;

import lombok.Data;

import java.io.Serializable;

@Data
public class BaseData implements Serializable{
    /**
     * 消息类型
     */
    private String msgType;
    /**
     * sn码 唯一标识
     */
    private long sn;
    /**
     * 售货机编码
     */
    private String innerCode;
    /**
     * 是否需要回传确认
     */
    private boolean needResp;

    private long versionId; 
}