首先先了解一下底层的协议:

1. MQTT

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅 （publish/subscribe）模式的"轻量级"通讯协议，该协议构建于TCP/IP协议上，由IBM在1999年发布。 MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作 为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应 用。

由于物联网的环境是非常特别的，所以MQTT遵循以下设计原则：

（1）精简，不添加可有可无的功能；
（2）发布/订阅（Pub/Sub）模式，方便消息在传感器之间传递，解耦Client/Server模式，带来的 好处在于不必预先知道对方的存在（ip/port），不必同时运行；
（3）允许用户动态创建主题（不需要预先创建主题），零运维成本；
（4）把传输量降到最低以提高传输效率；
（5）把低带宽、高延迟、不稳定的网络等因素考虑在内；
（6）支持连续的会话保持和控制（心跳）；
（7）理解客户端计算能力可能很低；
（8）提供服务质量（ quality of service level：QoS）管理
（9）不强求传输数据的类型与格式，保持灵活性（指的是应用层业务数据）

2. MQTT协议实现方式:

MQTT传输的消息分为：主题（Topic）和负载（payload）两部分：

（1）Topic， 可以理解为消息的类型，订阅者订阅（Subscribe）后，就会收到该主题的消息内容 （payload）；

（2）payload， 可以理解为消息的内容，是指订阅者具体要使用的内容。

3. Qos: 消息服务质量（Quality of Service）

MQTT 设计了 3 个 QoS 等级。

理解: Qos: 规定自己想要发出,或者接收到的消息的规则

• QoS 0：消息最多传递一次，如果当时客户端不可用，则会丢失该消息。

“至多一次”，消息发布完全依赖底层TCP/IP网络。会发生消息丢失或重复。这一级别可用于如下情况，环境传感器数据，丢失一次读记录无所谓，因为不久后还会有第二次发送。这一种方 式主要普通APP的推送，倘若你的智能设备在消息推送时未联网，推送过去没收到，再次联网也就 收不到了。

• QoS 1：消息传递至少 1 次。

“至少一次”，确保消息到达，但消息重复可能会发生

• QoS 2：消息仅传送一次。

“只有一次”，确保消息到达一次。在一些要求比较严格的计费系统中，可以使用此级别。在计费 系统中，消息重复或丢失会导致不正确的结果。这种最高质量的消息发布服务还可以用于即时通讯类的 APP的推送，确保用户收到且只会收到一次

MQTT 发布与订阅操作中的 QoS 代表了不同的含义:
发布时的 QoS 表示消息发送到服务端时使用的 QoS
订阅时的 QoS 表示服务端向自己转发消息时可以使用的最大 QoS。

基本都是用QoS2

MQTT 发布与订阅操作中的 QoS 代表了不同的含义，发布时的 QoS 表示消息发送到服务端时使用的 QoS，订阅时的 QoS 表示服务端向自己转发消息时可以使用的最大 QoS。

• 当客户端 A 的发布 QoS 大于客户端 B 的订阅 QoS 时，服务端向客户端 B 转发消息时使用的 QoS 为客户端 B 的订阅 QoS。
• 当客户端 A 的发布 QoS 小于客户端 B 的订阅 QoS 时，服务端向客户端 B 转发消息时使用的 QoS 为客户端 A 的发布 QoS。

总结: 也就是说QoS这个东西的设置是对消息的接收来做保证的,即如果是
QoS 1 我只负责发一次,收得到收不到我不管. 当网络状态不稳定的时候就会出现丢失现象
QoS 2 是能够保证至少收到一次,但是存在重复消费的问题. 
QoS 3 保证只有一条信息到达  

* 如果发布和订阅的客户端服务质量等级不相同时,谁的低按谁的为准

4. Topic通配符匹配规则

1. 层级分隔符 /

/ 用来分割主题树的每一层，并给主题空间提供分等级的结构。当两个通配符在一个主题中出现的 时候，主题层次分隔符的使用是很重要的。

示例：
love/you/with/all/my/heart

2. 多层通配符 #

多层通配符有可以表示大于等于0的层次。因此，love/#也可匹配到单独的love，此时#代表0 层。

多层通配符一定要是主题树的最后一个字符。比如说，love/#是有效的，但是love/#/with是无效 的。

3. 单层通配符 +

只匹配主题的一层

1. love/you/+ :匹配love/you/with和love/you/and，但是不匹配
love/you/with/all/my/heart。
2. 单层通配符只匹配1层，love/+不匹配love。
3. 单层通配符可以被用于主题树的任意层级，连带多层通配符。它必须被用在主题层级分隔符/的右边，除非它是指定自己。因此，+和love/+都是有效的，但是love+无效。单层通配符可以用在主题树的末端，也可以用在中间。比如说，love/+和love/+/with都是有效。

通配符注意事项:

1.主题层次分隔符被用来在主题中引入层次。多层的通配符和单层通配符可以被使用，但他们不能被使用来做发布者的消息。
2.Topic命名尽量见名知意，符合规范，主题名字是大小写敏感的。比如说，love和LOVE是两个不同的主题。
3.以/开头会产生一个不同的主题。比如说，/love与love不同。/love匹配"+/+"和/+,但不匹配+
4.不要在任何主题中包含null（Unicode \x0000）字符。
5.在主题树中，长度被限制于64k内但是在这以内没有限制层级的数目 。
6.可以有任意数目的根节点；也就是说，可以有任意数目的主题树。

1.EMQ X

EMQ X 是开源社区中最流行的 MQTT 消息服务器。

Windows安装emqx_windows emqx安装_罗小爬EX的博客-程序员宅基地

优点:

单机能支持百万的 MQTT 连接；集群能支持千 万级别的 MQTT 连接；

支持丰富的物联网协议，包括 MQTT、MQTT-SN、CoAP、 LwM2M、LoRaWAN 和 WebSocket 等；

2. Dashboard(可视化界面)

EMQ X 提供了 Dashboard 以方便用户管理设备与监控相关指标。通过 Dashboard可以查看服务器基本 信息、负载情况和统计数据，可以查看某个客户端的连接状态等信息甚至断开其连接，也可以动态加载 和卸载指定插件。除此之外，EMQ X Dashboard 还提供了规则引擎的可视化操作界面，同时集成了一 个简易的 MQTT 客户端工具供用户测试使用。

3. MQTTX

模拟客户端

MQTTX：跨平台 MQTT 5.0 桌面客户端工具

MQTT X 是 EMQ 开源的一款优雅的跨平台 MQTT 5.0 桌面客户端，它支持 macOS, Linux, Windows。 MQTT X 的 UI 采用了聊天界面形式，简化了页面操作逻辑，用户可以快速创建连接，允许保存多个客 户端，方便用户快速测试 MQTT/MQTTS 连接，及 MQTT 消息的订阅和发布。

发送消息

这里一定要记住是在哪方加前缀

1. 消息延迟发布

此功能由 emqx_mod_delayed 模块提供，需要开启模块后才能使用此功能。

$delayed/{DelayInteval}/{TopicName} 单位: S

当客户端使 用特殊主题前缀 $delayed/{DelayInteval} 发布消息到 EMQ X 时，将触发延迟发布功能

示例:

$delayed/15/x/y : 15 秒后将 MQTT 消息发布到主题 x/y 。

$delayed/60/a/b : 1 分钟后将 MQTT 消息发布到 a/b

示例: 在MQTT X上演示

现在在MQTTX上模拟四个客户端: 洗衣机,空调,电视机,手机

现在让洗衣机,空调,电视机订阅phoneMessage这个主题, 消息服务质量设置为2,以后都设置为2

各个实例订阅的主题和发送的主题如下:

洗衣机客户端订阅主题: phoneMessage
空调客户端订阅主题:   phoneMessage
电视机客户端订阅主题: phoneMessage

手机发送的主题: $delayed/2/phoneMessage       // 2 秒后发送到订阅phoneMessage的客户端

在手机客户端上发送消息： 注意此时特殊主题的前缀在发布者上

两秒后,其他订阅这个主题的客户端都收到了消息.

2. 共享订阅

注意注意注意:

共享订阅的主题格式是针对订阅端来指定的，例如： $share/group1/cookie ；而消息的发布方是向主 题： cookie 发布消息。这样在订阅方才能达到负载均衡的效果。

共享订阅是在多个订阅者之间实现负载均衡的订阅方式：

EMQ X 支持两种格式的共享订阅前缀:

	示例	前缀	真实主题名
方式一(不带群组的共享订阅)	$queue/name/cookie	$queue/	name/cookie
方式二(带群组的共享订阅)	$share/group1/cookie	$share/group1/	cookie

应用场景:

1️⃣ . 对于方式一不带群组的共享订阅:

在我们取钱完成时,我们要求需要向当前用户的手机发送一条短信, 为了提高服务的容错性,我们准备了多台发短信的服务, 但是我们要求只发送一条短信,此时我们就可以使用 方式一, 使只有一台机器收到发送短信的消息如下, 只会有一台机器收到

各个实例订阅的主题和发送的主题如下:

消息服务实例1订阅的主题:  $queue/cookie
消息服务实例2订阅的主题:  $queue/cookie
消息服务实例3订阅的主题:  $queue/cookie

发布者发布的主题:               cookie

2️⃣ 对于方式二带群组的共享订阅:

$share/group1/cookie

带分组的使用的特殊前缀是: $share/{group}/{TopicName}

使用场景如下:

在方式一的上面做一点点更改, 就是在发布者消息发送成功后, 不仅仅需要有一台消息服务实例去发送消息, 也需要有一台邮件服务实例去发送邮件. 所以这里就可以进行分组

message 组 : 消息服务实例1 消息服务实例2 消息服务实例2

email组: 邮件服务实例1 邮件服务实例2

各个实例订阅的主题和发送的主题如下:

(message组)
    消息服务实例1   $share/message/cookie
    消息服务实例2   $share/message/cookie
    消息服务实例2   $share/message/cookie

(email组)
    邮件服务实例1   $share/email/cookie
    邮件服务实例2   $share/email/cookie

    发布者实例       cookie

此时就可以保证两个组中都会有一个实例服务收到消息.

负载均衡策略修改默认 随机

在EMQ X 服务的 etc/emqx.conf中修改

broker.shared_subscription_strategy = random

负载均衡策略	描述
random	在所有订阅中随机选择
round_robin	按照订阅顺序轮询
sticky	一直发往上次选取的订阅者
hash	按照发布者ClientID的哈希值

原生Mqtt代码的使用

1.依赖导入

<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.5</version>
</dependency>

2. 编写Controller类

说明下面的一些配置也可以配置在yml文件中,这里只是演示(消息的发送和接受)所以就直接配置通过Set方法配置了

可以通过@ConfigurationProperties(prefix = "")写一个配置类

这里的server.port= 8888

@RestController
public class TestController {
    
    
    /**
     * 发布消息
     * @throws MqttException
     */
    @GetMapping("/publish")
    public void publish() throws MqttException {
    

        MqttClientPersistence persistence = new MemoryPersistence();  //内存持久化
        // 服务器地址以及本机的clientid
        MqttClient client = new MqttClient("tcp://192.168.200.128:1883", "abc", persistence);
        //连接选项中定义用户名密码和其它配置
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);//参数为true表示清除缓存，也就是非持久化订阅者，这个时候只要参数设为true，一定是非持久化订阅者。而参数设为false时，表示服务器保留客户端的连接记录
        options.setAutomaticReconnect(true);//是否自动重连
        options.setConnectionTimeout(30);//连接超时时间  秒
        options.setKeepAliveInterval(10);//连接保持检查周期  秒
        options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1); //版本
        client.connect(options);//连接
        client.publish("testTopic", "发送内容".getBytes(), 2, false);
    }


    /**
     * 订阅消息
     * @throws MqttException
     */
    @GetMapping("/subscribe")
    public void subscribe() throws MqttException {
    

        MqttClientPersistence persistence = new MemoryPersistence();;//内存持久化
        // 服务器地址以及本机的clientid
        MqttClient client = new MqttClient("tcp://192.168.200.128:1883", "xyz", persistence);

        //连接选项中定义用户名密码和其它配置
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);//参数为true表示清除缓存，也就是非持久化订阅者，这个时候只要参数设为true，一定是非持久化订阅者。而参数设为false时，表示服务器保留客户端的连接记录
        options.setAutomaticReconnect(true);//是否自动重连
        options.setConnectionTimeout(30);//连接超时时间  秒
        options.setKeepAliveInterval(10);//连接保持检查周期  秒
        options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1); //版本
        
		// 设置回调函数    匿名内部类
        client.setCallback(new MqttCallbackExtended() {
    
            @Override
            public void connectionLost(Throwable throwable) {
    
                System.out.println("连接丢失！");
            }

            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
    
                System.out.println( "接收到消息  topic:" +s+"  id:"+mqttMessage.getId() +" message:"+ mqttMessage.toString());
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
    

            }

            @Override
            public void connectComplete(boolean b, String s) {
    
                System.out.println("连接成功！");
            }
        });
        client.connect(options);//连接
        client.subscribe("testTopic");  //订阅主题

    }
}

访问: http://localhost:8888/publish

访问: http://localhost:8888/subscribe

第一次访问两个地址,首先会先EMQ注册自己,和EMQ建立连接(但是一定要快,如果到了10秒任务你就挂了)

等到第二次访问,已经建立过,所以就可以发送消息,和接收消息,此时再看控制台.

OK,原始的方法建立完毕

现在可以设想,我们在一个项目中一般不会只发一次请求, 通过上面的代码我们可以发现每一次的发送其实有绝大部分内容是相同的.对于发送者来说我们要变的只有主题和消息 而对于接收者来说需要根据不同的主题来设计不同的处理方案, 即根据不同的主题调用不同的方案来解决(是不是有一点多态的意思了), 那么有没有一种解决方案来解决这样的问题. bingo~ 必然.

首先先认识一个设计模式: 策略模式 (解决多个 if-else 问题)

策略模式详解 - 知乎 (zhihu.com)

在策略模式（Strategy Pattern）中，一个类的行为或其算法可以在运行时更改。我们要处理的就是上面高亮的部分

在策略模式中，我们创建表示各种策略的对象和一个行为随着策略对象改变而改变的 context 对象。策略对象改变 context 对象的执行算法。

大家可以看上面知乎中的链接的内容; 理解一下在支付时的策略模式的使用,然后在看我们自己的实现.

使用策略模式改造

注：
一般的策略模式大概是这样：

• 定义策略接口
• 定义不同策略实现类
• 提供策略工厂，便于根据策略枚举获取不同策略实现

而在策略比较简单的情况下，我们完全可以用枚举代替策略工厂，简化策略模式。

1. 配置文件

server:
  port: 8082
mqtt:
  client:
    username: admin
    password: public
    serverURI: tcp://192.168.200.128:1883
    clientId: monitor.task.${
    random.int[1000,9999]} # 注意: emq的客户端id 不能重复
    keepAliveInterval: 10  #连接保持检查周期  秒
    connectionTimeout: 30 #连接超时时间  秒
  producer:
    defaultQos: 2
    defaultRetained: false
    defaultTopic: topic/test1
  consumer:
    consumerTopics: $queue/cookie/#, $share/group1/yfs1024  #不带群组的共享订阅    多个主题逗号隔开
    # $queue/cookie/#
    # 以$queue开头，不带群组的共享订阅   多个客户端只能有一个消费者消费

    # $share/group1/yfs1024
    # 以$share开头，群组的共享订阅 多个客户端订阅
    # 如果在一个组 只能有一个消费者消费
    # 如果不在一个组 都可以消费

对应配置类

@Data
@Slf4j
@Configuration
@ConfigurationProperties(prefix = "mqtt.client")
public class MqttProperties {
    
    private int defaultProducerQos;
    private boolean defaultRetained;
    private String defaultTopic;
    private String username;
    private String password;
    private String serverURI;
    private String clientId;
    private int keepAliveInterval;
    private int connectionTimeout;
}

2. 封装消息发送者

下面通过方法重载的方式, 接收不同个数的参数

@Component
@Slf4j
public class MqttProducer {
    
	// @Value() 读取配置 当然也可以批量读取配置,这里就一个一个了
    @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> void send(String topic, int qos, T msg) throws JsonProcessingException {
    
        String payload = JsonUtil.serialize(msg);
        this.send(topic,qos,payload);
    }
}

3. 定义配置类

说明一下,一般对于一台服务来说发送者和接受者使用一个mqtt连接,所以这里配置了一个mqttClient

@Configuration
@Data
@Slf4j
public class MqttConfig {
    

    // 注入配置
    @Autowired
    private MqttProperties mqtt;
    // 注入回调函数
    @Autowired
    private MqttCallback mqttCallback;
    
    @Bean
    public MqttClient mqttClient() {
    
        try {
    
            MqttClient client = new MqttClient(mqtt.getServerURI(), mqtt.getClientId(), new MemoryPersistence());
            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;
        }
    }

    @Bean
    public MqttConnectOptions mqttConnectOptions() {
    
        // 下面有一些配置是写死的, 如果项目需要最好还是写配置文件中, 这样后面可以通过注册中心热更新配置文件
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(mqtt.getUsername());
        options.setPassword(mqtt.getPassword().toCharArray());
        options.setAutomaticReconnect(true);//是否自动重新连接
        options.setCleanSession(true);//是否清除之前的连接信息
        options.setConnectionTimeout(mqtt.getConnectionTimeout());//连接超时时间
        options.setKeepAliveInterval(mqtt.getKeepAliveInterval());//心跳
        options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);//设置mqtt版本

        return options;
    }
}

简单测试一下:

这里发送一个对象

@Data
public class User {
    
    private String id;
    private String name;
    private Integer age;
}

@RestController
public class MyPublish {
    

    @Autowired
    private MqttProducer mqttProducer;

    @GetMapping("/testPublish")
    public void testSend(){
    
        User user = new User();
        user.setId("123");
        user.setName("张三");
        user.setAge(18);
        mqttProducer.send("cookie",2,user);
    }

}

访问: localhost:9999/testPublish

OK,现在消息的发送是已经完成了

我们在Dashboard中订阅这个主题并查看结果:

发送消息已经没有问题了, 注入MqttProducer调用send方法就可以了, 下面就来解决消息接收的问题.

还记得上面说的嘛, 对于接收来说,主要的瓶颈在于根据不同的主题来处理不同的消息, 比如订阅了cookie,和yfs1024这两个主题可以通过如下代码实现:

在接收到的参数中通过判断来确定处理的方法:

@Override
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
    
	if(topic.equals("cookie")){
    
        // cookie 主题的处理逻辑
    }else if(topic.equals("yfs1024")){
    
        // yfs1024 主题的处理逻辑
    }else if(xxx){
     // 其他的主题
	    .........        
    }
}

这种方法可以没有问题, 但是每次都需要通过修改这里的代码来处理业务逻辑, 就像上面知乎文章所说的虽然写起来简单, 但是违反了面向对象的两个设计原则， 单一职责原则，开闭原则。

好那么下面就按照策略模式来对接收消息的代码进行改造。

4. 改造接收代码

1. 定义消息处理接口

大致的逻辑: 因为是不同的主题拥有不同的处理逻辑, 即 一个主题对应一个处理类, 我们要做的就是通过主题拿到这个主题的处理类

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

后面让每个主题的处理类来实现这个接口就可以

@Component
public class CookieHandler implements MsgHandler {
    
    @Override
    public void process(String jsonMsg) throws IOException {
    
        // 解析       这里使用了自己封装JSON 工具类, 我放最下面了
        User byJson = JsonUtil.getByJson(jsonMsg, User.class);
        System.out.println("CookieHandler 接受到消息：" + byJson);
    }
}

我们还需要根据主题使用不同的处理器, 怎么办呢? 可以通过注解中的参数的值区分(通过反射获取)

2. 定义主题注解

用来表示当前类所处理的注解.

@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target({
    ElementType.TYPE})
public @interface Topic {
    
    String value();
}

对上面进行改造如下:

@Component
@Topic("cookie")
public class CookieHandler implements MsgHandler {
    
    @Override
    public void process(String jsonMsg) throws IOException {
    
        // 解析       
        User byJson = JsonUtil.getByJson(jsonMsg, User.class);
        System.out.println("CookieHandler 接受到消息：" + byJson);
    }
}

3. 将主题与对应处理类建立映射

现在处理主题的方法类有了, 对每个处理类也做了对应主题的标记, 那么如果把他们对应起来呢? 建立映射, 代码实现如下:

定义接口

/**
 * 消息处理上下文, 通过主题拿到topic
 */
public interface MsgHandlerContext{
    
    MsgHandler getMsgHandler(String topic);
}

/**
 * 消息处理类加载器
 * 作用:
 * 1. 因为实现了Spring 的 ApplicationContextAware 接口, 项目启动后就会运行实现的方法
 * 2. 获取MsgHandler接口的所有的实现类
 * 3. 将实现类上的Topic注解的值,作为handlerMap的键,实现类(处理器)作为对应的值
 */
@Component
public class MsgHandlerContextImp implements ApplicationContextAware, MsgHandlerContext {
    
    private Map<String, MsgHandler> handlerMap = Maps.newHashMap();

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
    
        // 从spring容器中获取 <所有> 实现了MsgHandler接口的对象
        // key 默认类名首字母小写 value 当前对象
        Map<String, MsgHandler> map = applicationContext.getBeansOfType(MsgHandler.class);
        map.values().forEach(obj > {
    
            // 通过反射拿到注解中的值  即 当前类处理的 topic
            String topic =  obj.getClass().getAnnotation(Topic.class).value();
		   // 将主题和当前主题的处理类建立映射
            handlerMap.put(topic,obj);
        });
    }

    @Override
    public MsgHandler getMsgHandler(String topic) {
    
        return handlerMap.get(topic);
    }
}

OK, 现在主题和对应处理类的问题已经完成了. 那么下一步就是怎么在接收消息的回调函数中根据接收到消息的主题,调用处理类消费消息.

其实我们在上面定义配置的时候已经导入了方法回电函数的实现

// 注入回调函数
@Autowired
private MqttCallback mqttCallback;

在没定义之前考虑一下我们要做什么呢?

1. 在连接成功 之后订阅所有主题
2. 处理回调函数,根据主题获取处理器,处理消息内容

// 回调函数接口的实现类, 重写连接丢失,建立连接,
@Component
@Slf4j
public class MqttCallback implements MqttCallbackExtended {
    
    // 需要订阅的topic配置
    @Value("${mqtt.consumer.consumerTopics}")
    private List<String> consumerTopics;

    @Autowired
    private MsgHandlerContext msgHandlerContext;

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

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
    
        log.info("topic:" + topic + "  message:" + new String(message.getPayload()));
        //处理消息
        String msgContent = new String(message.getPayload());
        log.info("接收到消息:" + msgContent);
        try {
    
            // 根据主题名称 获取 该主题对应的处理器对象
            // 多态 父类的引用指向子类的对象
            MsgHandler msgHandler = msgHandlerContext.getMsgHandler(topic);
            if (msgHandler == null) {
    
                return;
            }
            
            msgHandler.process(msgContent); //执行
        } catch (IOException e) {
    
            log.error("process msg error,msg is: " + msgContent, e);
        }
//        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) {
    
        log.info("连接成功");
        //和EMQ连接成功后根据配置自动订阅topic
        if (consumerTopics != null && consumerTopics.size() > 0) {
    
            // 循环遍历当前项目中配置的所有的主题.
            consumerTopics.forEach(t -> {
    
                try {
    
                    log.info(">>>>>>>>>>>>>>subscribe topic:" + t);
                    // 订阅当前集群中所有的主题 消息服务质量 2 -> 至少收到一个
                    mqttClient.subscribe(t, 2);
                } catch (MqttException e) {
    
                    log.error("emq connect error", e);
                }
            });
        }
    }

    private MqttClient mqttClient;
	// 在配置类中调用传入连接
    public void setMqttClient(MqttClient mqttClient) {
    
        this.mqttClient = mqttClient;
    }
}

5. 测试:

使用步骤:

自定义处理类 通过实现MsgHandler 接口

@Component
@Topic("cookie")
public class CookieHandler implements MsgHandler {
    
    @Override
    public void process(String jsonMsg) throws IOException {
    
        // 解析       这里使用了自己封装JSON 工具类, 我放最下面了
        User byJson = JsonUtil.getByJson(jsonMsg, User.class);
        System.out.println("CookieHandler 接受到消息：" + byJson);
    }
}

此时要注意配置文件中有没有对应的主题, 我的这里有,采用不带群组共享订阅

consumerTopics: $queue/cookie/#, $share/group1/yfs1024  

控制台有

可以接收到数据,并且在启动两个示例的时候也能实现共享订阅, 随机选择

这里注意哈, 如果是测试的话, 订阅者一定要带上前缀$queue或$share/groupName

JsonUtil

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.*;

import java.io.IOException;
import java.util.Map;

public class JsonUtil {
    
    /**
     * 从json字符串中根据nodeName获取值
     * @param nodeName
     * @param json
     * @return
     * @throws IOException
     */
    public static String getValueByNodeName(String nodeName, String json) throws IOException {
    
        ObjectMapper objectMapper = new ObjectMapper();
        JsonNode jsonNode = objectMapper.readTree(json);
        JsonNode node = jsonNode.findPath(nodeName);
        if(node == null) return null;

        return node.asText();
    }

    /**
     * 根据nodeName获取节点内容
     * @param nodeName
     * @param json
     * @return
     * @throws IOException
     */
    public static JsonNode getNodeByName(String nodeName, String json) throws IOException {
    
        ObjectMapper objectMapper = new ObjectMapper();

        return objectMapper.readTree(json).findPath(nodeName);
    }

    /**
     * 反序列化
     * @param json
     * @param clazz
     * @param <T>
     * @return
     * @throws IOException
     */
    public static <T> T getByJson(String json, Class<T> clazz) throws IOException {
    
        ObjectMapper mapper = new ObjectMapper();
        // 在反序列化时忽略在 json 中存在但 Java 对象不存在的属性
        mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        // 在序列化时日期格式默认为 yyyy-MM-dd'T'HH:mm:ss.SSSZ
        mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);

        return mapper.readValue(json, clazz);
    }

    /**
     * 反序列化（驼峰转换）
     * @param json
     * @param clazz
     * @param <T>
     * @return
     * @throws IOException
     */
    public static <T> T getByJsonSNAKE(String json, Class<T> clazz) throws IOException {
    
        ObjectMapper mapper = new ObjectMapper();
        // 在反序列化时忽略在 json 中存在但 Java 对象不存在的属性
        mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        // 在序列化时日期格式默认为 yyyy-MM-dd'T'HH:mm:ss.SSSZ
        mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
        // 设置驼峰和下划线之间的映射
        mapper.setPropertyNamingStrategy(PropertyNamingStrategy.SNAKE_CASE);
        return mapper.readValue(json, clazz);
    }


    /**
     * 序列化
     * @param object
     * @return
     * @throws JsonProcessingException
     */
    public static String serialize(Object object) throws JsonProcessingException {
    
        ObjectMapper mapper = new ObjectMapper();

        return mapper.writeValueAsString(object);
    }


    /**
     * 序列化（驼峰转换）
     * @param object
     * @return
     * @throws JsonProcessingException
     */
    public static String serializeSNAKE(Object object) throws JsonProcessingException {
    
        ObjectMapper mapper = new ObjectMapper();
        // 设置驼峰和下划线之间的映射
        mapper.setPropertyNamingStrategy(PropertyNamingStrategy.SNAKE_CASE);
        return mapper.writeValueAsString(object);
    }


    public static JsonNode getTreeNode(String json) throws JsonProcessingException {
    
        ObjectMapper objectMapper = new ObjectMapper();

        return objectMapper.readTree(json);
    }


    /**
     * 将对象转map
     * @param obj
     * @return
     * @throws IOException
     */
    public static Map<String,Object> convertToMap(Object obj) throws IOException {
    
        ObjectMapper mapper = new ObjectMapper();

        return mapper.readValue(serialize(obj),Map.class);
    }
}

---

如果您发现错误,还望及时提醒,共同进步

后面还有代理订阅,保留消息,认证,ACL,这里暂时还没有用到所以就不在记录, 用到的话再补充