springboot集成MQTT

spring的MQTT 支持

spring-integration的版本为：5.2.1

Spring Integration提供了入站适配器和出站适配器以支持MQTT协议。

你需要在你的项目中加入spring-integration-mqtt依赖：

Maven:

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mqtt</artifactId>
    <version>5.2.1.RELEASE</version>
</dependency>

Gradle:

compile "org.springframework.integration:spring-integration-mqtt:5.2.1.RELEASE"

当前的实现采用的是Eclipse Paho MQTT Client库。

入站适配器和出站的适配器的配置都是通过DefaultMqttPahoClientFactory实现的。关于更多的配置选项可以查看Paho的文档。

我们推荐你配置一个MqttConnectOptions对象，并把它注入到连接工厂中，而不是在连接工厂中自己设置。

入站(消息驱动)通道适配器

入站适配器是通过MqttPahoMessageDrivenChannelAdapter来实现的。为方便起见，你可以使用命名空间来配置，最小的配置如下：


<bean id="clientFactory"
        class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory">
    <property name="connectionOptions">
        <bean class="org.eclipse.paho.client.mqttv3.MqttConnectOptions">
            <property name="userName" value="${mqtt.username}"/>
            <property name="password" value="${mqtt.password}"/>
        </bean>
    </property>
</bean>

<int-mqtt:message-driven-channel-adapter id="mqttInbound"
    client-id="${mqtt.default.client.id}.src"
    url="${mqtt.url}"
    topics="sometopic"
    client-factory="clientFactory"
    channel="output"/>

下面这段配置展示了可以用到的属性：


<int-mqtt:message-driven-channel-adapter id="oneTopicAdapter"
    client-id="foo"
    url="tcp://localhost:1883"
    topics="bar,baz"
    qos="1,2"
    converter="myConverter"
    client-factory="clientFactory"
    send-timeout="123"
    error-channel="errors"
    recovery-interval="10000"
    channel="out" />

client-id 客户端id
url broker的地址
topics 此适配器接收消息的主题，以逗号分割
qos 逗号分割的Qos值
converter 此处指的是MqttMessageConverter,可选。默认会使用DefaultPahoMessageConver,此消息转换器会产生一个消息体为字符串的消息，这个消息带有这些头信息：
. mqtt_topic : 消息来自于哪一个topic
. mqtt_duplicate: true，如果此消息为重复消息的话
. mqtt_qos : 消息的Qos
client-factory 客户端工厂
send-timeout 发送的超时时间。它只在通道阻塞时起作用
error-channel 错误通道：下载流的异常会以ErrorMessage的形式被发送到此通道内。消息体为MessagingException
，MessagingException 中包含有失败信息和原因。
recovery-interval 恢复的时间间隔，此选项控制着适配器在失败后重连的间隔时间，默认是10000ms即10s。

注意：

从4.1开始，你可以在上面的配置中省略URL，而只需要在DefaultMqttPahoClientFactory中配置serverURIs 即可。开启这项配置，你可以连接到一个高可用的集群。

从4.2.2开始，如果此适配器成功订阅这些topic，就会生成一个MqttSubscribedEvent 事件。当连接或订阅失败时，会生成一个MqttConnectionFailedEvent事件。只要我们定义一个ApplicationListener的类，就可以接收这些事件。同样，recoveryInterval 此选项控制着适配器在失败后重连的间隔时间，默认是10000ms即10s。

注意：

xxxx

注意：

xxxx

在运行时添加和移除topic

从4.1版本开始，你可以通过程序来修改此适配器订阅的topic。Spring Integration提供了addTopic()和removeTopic()方法。
在添加topic时，你可以指定Qos值（如果不指定的话，默认值是1)。你也可以发送一个带有特定消息体的消息给<control-bus/>来修改topic。例如： "myMqttAdapter.addTopic('foo', 1)"。

停止和启动适配器对主体列表不会产生影响(它不会恢复原来在配置中所做的设置)。应用上下文的生命周期之外，这些修改不会被保留下来。一个新的应用上下文会恢复到所做的配置。

在适配器停止时(或与broker断开连接时)，修改主题，会在下一次新连接建立时，生效。

使用java进行配置

下面这个示例展示了如何使用java配置入站适配器：

@SpringBootApplication
public class MqttJavaApplication {

    public static void main(String[] args) {
        new SpringApplicationBuilder(MqttJavaApplication.class)
                .web(false)
                .run(args);
    }

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "testClient",
                                                 "topic1", "topic2");
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {

            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                System.out.println(message.getPayload());
            }

        };
    }

}

使用java DSL进行配置

下面这个示例展示了如何使用Java DSL配置入站适配器：

@SpringBootApplication
public class MqttJavaApplication {

    public static void main(String[] args) {
        new SpringApplicationBuilder(MqttJavaApplication.class)
            .web(false)
            .run(args);
    }

    @Bean
    public IntegrationFlow mqttInbound() {
        return IntegrationFlows.from(
                         new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883",
                                        "testClient", "topic1", "topic2");)
                .handle(m -> System.out.println(m.getPayload()))
                .get();
    }

}

出站通道适配器

出站通道适配器是通过MqttPahoMessageHandler来实现的，它被封装在ConsumerEndpoint中。为方便起见，你可以使用命名空间来配置它。

从版本4.1开始，适配器支持了异步发送操作，避免了在递交确认之前的阻塞。如果需要的话，你可以通过应用事件来开启递交确认。

下面这个列表展示了出站通道适配器的可用属性配置：

<int-mqtt:outbound-channel-adapter id="withConverter"
    client-id="foo"
    url="tcp://localhost:1883"
    converter="myConverter"
    client-factory="clientFactory"
    default-qos="1"
    qos-expression=""
    default-retained="true"
    retained-expression=""
    default-topic="bar"
    topic-expression=""
    async="false"
    async-events="false"
    channel="target" />

client-id 客户端id
url broker的地址

converter MqttMessageConverter (这项配置是可选的)，默认的DefaultPahoMessageConverter承认下面这些头部：
. mqtt_topic: 消息被发送到哪个主题上

                                               . mqtt_retained             如果消息需要保留的话，就设置为true。

                                               . mqtt_qos                  服务质量

client-factory 客户端工厂
default-qos 默认的服务质量。在没有mqtt_qos头部或qos_expression为null的情况下，将使用这个服务质量。如果你提供的有自定义的转换器的话，此配置不会应用。
qos-expression 计算Qos的表达式，默认值是： headers[mqtt_qos]
default-retained 保留标志的默认值。在没有发现mqtt_retained头部的情况下，将使用此值。但是如果你提供的有自定义的转换器的话，这项配置不会应用。
retained-expression 计算保留标志的表达式。默认值是headers[mqtt_retained]。
default-topic 默认主题，如果消息没有mqtt_topic头部的话，将默认把消息发送给该主题。
topic-expression 计算目标主题的表达式，默认值是：headers['mqtt_topic']。
async 当为true的时候，调用者不会被阻塞住。不然的话，当消息发送之后，它会一直等待确认消息。默认值是false。(发送操作会一直阻塞住知道收到确认消息)

async-events 默认值为false。如果async和async-events都是true的话，就会生成一个MqttMessageSentEvent事件。事件中包含此消息，主题、messageId，clientId以及clientInstance 。
当客户端库完成递交确认之后，就会产生一个MqttMessageDeliveredEvent 事件，事件中包含： messageId、clientId、clientInstance，使得消息递交和发送操作相关联。
任意一个ApplicationListener 或 event inbound channel adapter 都能接收这些事件。注意：MqttMessageDeliveredEvent事件有可能先于MqttMessageSentEvent事件被接收。

注意：

从4.1开始，你可以在上面的配置中省略URL，而只需要在DefaultMqttPahoClientFactory中配置serverURIs 即可。开启这项配置，你可以连接到一个高可用的集群。

使用java进行配置

下面这个示例展示了如何使用java配置出站适配器：

@SpringBootApplication
@IntegrationComponentScan
public class MqttJavaApplication {

    public static void main(String[] args) {
        ConfigurableApplicationContext context =
                new SpringApplicationBuilder(MqttJavaApplication.class)
                        .web(false)
                        .run(args);
        MyGateway gateway = context.getBean(MyGateway.class);
        gateway.sendToMqtt("foo");
    }

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[] { "tcp://host1:1883", "tcp://host2:1883" });
        options.setUserName("username");
        options.setPassword("password".toCharArray());
        factory.setConnectionOptions(options);
        return factory;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler =
                       new MqttPahoMessageHandler("testClient", mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic("testTopic");
        return messageHandler;
    }

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    @MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
    public interface MyGateway {

        void sendToMqtt(String data);

    }

}

使用Java DSL进行配置

下面这个示例展示了如何使用java DSL配置出站适配器：

@SpringBootApplication
public class MqttJavaApplication {

    public static void main(String[] args) {
        new SpringApplicationBuilder(MqttJavaApplication.class)
            .web(false)
            .run(args);
    }

    @Bean
    public IntegrationFlow mqttOutboundFlow() {
        return f -> f.handle(new MqttPahoMessageHandler("tcp://host1:1883", "someMqttClient"));
    }

}

应用实践案例

springboot版本：2.1.3.RELEASE。

引入依赖：


<!-- MQTT -->
       <dependency>
           <groupId>org.springframework.boot</groupId>
           <artifactId>spring-boot-starter-integration</artifactId>
           <scope>compile</scope>
           <version>2.1.3.RELEASE</version>
           <exclusions>
               <exclusion>
                   <artifactId>jackson-module-kotlin</artifactId>
                   <groupId>com.fasterxml.jackson.module</groupId>
               </exclusion>
           </exclusions>
       </dependency>
       <dependency>
           <groupId>org.springframework.integration</groupId>
           <artifactId>spring-integration-stream</artifactId>
           <scope>compile</scope>
           <version>5.2.1.RELEASE</version>
           <exclusions>
               <exclusion>
                   <artifactId>jackson-module-kotlin</artifactId>
                   <groupId>com.fasterxml.jackson.module</groupId>
               </exclusion>
           </exclusions>
       </dependency>
       <dependency>
           <groupId>org.springframework.integration</groupId>
           <artifactId>spring-integration-mqtt</artifactId>
           <scope>compile</scope>
           <version>5.2.1.RELEASE</version>
           <exclusions>
               <exclusion>
                   <artifactId>jackson-module-kotlin</artifactId>
                   <groupId>com.fasterxml.jackson.module</groupId>
               </exclusion>
           </exclusions>
       </dependency>

入站通道配置

代码如下：

/**
 * @Author: chihaojie
 * @Date: 2020/4/8 18:55
 * @Version 1.0
 * @Note
 */
@Configuration
// spring Integration组件扫描，MessageChannel使用的就是这个组件
@IntegrationComponentScan
@Slf4j
public class MqttInboundConfiguration {

    @Autowired
    MqttMessageSender sender;

    @Autowired
    DeviceLogRepository deviceLogRepository;

    private MqttPahoMessageDrivenChannelAdapter adapter;

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        String[] urls = MqttInboundProperties.url.split(",");

        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setConnectionTimeout(60000);
        options.setKeepAliveInterval(10);//30s的保活时间
        options.setAutomaticReconnect(true);
        options.setMaxInflight(1000);
        options.setServerURIs(urls);
        options.setUserName(MqttInboundProperties.username);
        options.setPassword(MqttInboundProperties.password.toCharArray());
        factory.setConnectionOptions(options);
        return factory;
    }


    @Bean
    public MessageProducerSupport mqttInbound() {

        String[] topics = MqttInboundProperties.topics.split(",");
        if (adapter == null ){
            adapter = new MqttPahoMessageDrivenChannelAdapter(MqttInboundProperties.clientId,
                    mqttClientFactory(), topics);
        }
        adapter.addTopic(MqttTopics.DEVICE_LOG_FILENAME_TOPIC);
        adapter.setCompletionTimeout(60000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setRecoveryInterval(10000);
        adapter.setQos(0);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {

            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                try {
                    log.info(message.toString());
                    MessageHeaders headers = message.getHeaders();
                    String mqtt_receivedTopic = (String) headers.get("mqtt_receivedTopic");
                    log.info("【获取到的消息的topic 是：】{} ",mqtt_receivedTopic);
                    if(message instanceof MqttPublishMessage){
                        MqttPublishMessage pub=(MqttPublishMessage) message;
                        log.info("检测入站消息有无topic,topic是【】",pub.variableHeader().topicName());
                        log.info(pub.toString());
                    }
                    String payload = (String) message.getPayload();
                    JSONObject jsonObject = JSONObject.parseObject(payload);
                    String type = String.valueOf(jsonObject.get("Type"));
                    switch (type){
                        case MqttMessageTypeConstant.TO_SERVER_FILE_NAME:
                            deviceLogFileName(jsonObject.get("deviceId").toString(),payload);
                            break;
                        default:
                            break;
                    }
                    log.info("=========入站消息============");
                    log.info(payload);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
    }


    /**
     * 添加主题
     * @param topic
     */
    public void addTopic(String topic){
        if (adapter == null){
            adapter = new MqttPahoMessageDrivenChannelAdapter(MqttInboundProperties.clientId + System.currentTimeMillis(),
                    mqttClientFactory(), "");
        }
        adapter.addTopic(topic,0);
        log.info("添加 :"+ topic);
    }

    /**
     * 移除主题
     * @param topic
     */
    public void removeTopic(String topic){
        if (adapter == null){
            adapter = new MqttPahoMessageDrivenChannelAdapter(MqttInboundProperties.clientId + System.currentTimeMillis(),
                    mqttClientFactory(), "");
        }
        log.info("移除 :"+topic);
        adapter.removeTopic(topic);
    }

}

出站通道配置

代码如下：

@Configuration
@Slf4j
public class MqttOutboundConfiguration {




//    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        String[] urls = MqttOutboundProperties.url.split(",");
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setAutomaticReconnect(true);
        options.setMaxInflight(1000);
        options.setConnectionTimeout(60000);
        options.setKeepAliveInterval(10);//30s的保活时间
        options.setServerURIs(urls);
        options.setUserName(MqttOutboundProperties.username);
        options.setPassword(MqttOutboundProperties.password.toCharArray());
        factory.setConnectionOptions(options);
        return factory;
    }


// publisher

    @Bean
    public IntegrationFlow mqttOutFlow() {
        return IntegrationFlows.from(CharacterStreamReadingMessageSource.stdin(),
                e -> e.poller(Pollers.fixedDelay(2000)))
                .transform(p -> p + "")
                .handle(mqttOutbound())
                .get();
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(MqttOutboundProperties.clientId+System.currentTimeMillis(), mqttClientFactory());
        //开启异步
        messageHandler.setDefaultQos(0);
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(MqttOutboundProperties.out_topics);
        return messageHandler;
    }

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }


}

消息网关

消息网关用于完成消息的发送：

/**
 * @Author: chihaojie
 * @Date: 2020/4/8 19:07
 * @Version 1.0
 * @Note
 */
@Component
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {

    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
}

封装一个消息发送器

代码如下：

/**
 * @Author: chihaojie
 * @Date: 2019/5/30 17:57
 * @Version 1.0
 * @Note
 */

@Component
public class MqttMessageSender {



    @Autowired
    private MqttGateway mqttGateway;




    public void send(String deviceId, String msgBody){
        //mqttOutFlow.getInputChannel().send(new GenericMessage<>(JSONObject.toJSONString(mqttMessage)));
        mqttGateway.sendToMqtt("/to/device/"+deviceId,0,msgBody);
    }

    public void   send(MqttMessageSendCommon msg){
        if(!ObjectUtils.isEmpty(msg)){
            String data=JSONObject.toJSONString(msg);
            String topic=msg.getMsgReceiver();
            mqttGateway.sendToMqtt("/to/device/"+topic,0,data);
        }
    }

    public void   sendMsgBatch(OTAUpdateMessageBody msgBody, List<String> clientIdList){
        if(!ObjectUtils.isEmpty(clientIdList) && !ObjectUtils.isEmpty(msgBody)){
            for (String clientId: clientIdList) {
                String topic=clientId;
                MqttMessageSendCommon msg=new MqttMessageSendCommon();
                msg.setMsgIdentifier(String.valueOf(System.currentTimeMillis()));
                msg.setMsgType("OTA_UPDATE");
                msg.setMsgSender("SERVER");
                msg.setMsgReceiver(topic);
                msg.setMsgBody(JSONObject.toJSONString(msgBody));
                msg.setSendTime(new Date());
                msg.setNeedAck(true);
                String data=JSONObject.toJSONString(msg);
                mqttGateway.sendToMqtt("/to/device/"+topic,0,data);
            }
        }
    }

}

消息体对象

通用格式的消息对象

/**
 * @Author: chihaojie
 * @Date: 2019/5/30 17:22
 * @Version 1.0
 * @Note  通用消息格式
 */
@Data
public class MqttMessageSendCommon {

    public MqttMessageSendCommon() {
        sendTime = new Date();
    }

    //消息标识
    private String   msgIdentifier;
    //消息类型
    private String   msgType;
    //是否需要应答
    private Boolean  needAck;
    //消息体
    private String   msgBody;


    //消息发送方
    private String   msgSender;
    //消息接收方
    private String   msgReceiver;
    //发送时间
    private Date     sendTime;


    public static void main(String[] args) {
      /*  MqttMessageSendCommon sendCommon=new MqttMessageSendCommon();
        //
        sendCommon.setSendTime(new Date());
        //xxx
        OTAUpdateMessageBody messageBody=new OTAUpdateMessageBody();
        messageBody.setConfFileName("xxx.conf");
        sendCommon.setMsgBody(JSONObject.toJSONString(messageBody));*/

    }

}

和业务相关的消息体对象

以OTA升级为例，代码如下：

/**
 * @Author: chihaojie
 * @Date: 2020/4/15 14:23
 * @Version 1.0
 * @Note
 */
@Data
public class OTAUpdateMessageBody {


    private String pubRecordId;

    private String msgReceiver;


    private String deviceId;

    private String deviceType;

    private String otaName;


    private Integer otaVersion;

    private String fileMd5;

    private String  uploadPath;

    private String fileUrl;

    private String confFileName;

    private String confFileUploadPath;

    private String  confFileUrl;
}

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 205,236评论 6赞 478
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 87,867评论 2赞 381
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 151,715评论 0赞 340
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 54,899评论 1赞 278
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 63,895评论 5赞 368
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 48,733评论 1赞 283
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 38,085评论 3赞 399
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 36,722评论 0赞 258
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 43,025评论 1赞 300
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 35,696评论 2赞 323
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 37,816评论 1赞 333
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,447评论 4赞 322
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 39,057评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 30,009评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,254评论 1赞 260
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 45,204评论 2赞 352
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,561评论 2赞 343

springboot集成MQTT

spring的MQTT 支持

入站(消息驱动)通道适配器

在运行时添加和移除topic

使用java进行配置

使用java DSL进行配置

出站通道适配器

使用java进行配置

使用Java DSL进行配置

应用实践案例

引入依赖：

入站通道配置

出站通道配置

消息网关

封装一个消息发送器

消息体对象

通用格式的消息对象

和业务相关的消息体对象

推荐阅读更多精彩内容