如何使用Spring?integration在Springboot中集成Mqtt詳解

更新時(shí)間：2023年02月24日 09:10:52 作者：小艾咪

MQTT是多個(gè)客戶端通過(guò)一個(gè)中央服務(wù)器傳遞信息的多對(duì)多協(xié)議,能高效地將信息分發(fā)給一個(gè)或多個(gè)訂閱者,下面這篇文章主要給大家介紹了關(guān)于如何使用Spring?integration在Springboot中集成Mqtt的相關(guān)資料,需要的朋友可以參考下

前言

接觸Mqtt最開(kāi)始用的是SpingMVC 引入 Mqtt Client Jar包自己實(shí)現(xiàn)收發(fā)重連等操作，而現(xiàn)在則使用Spring Integration 進(jìn)行集成。其復(fù)雜度和代碼簡(jiǎn)潔度天差地別。所以特別寫(xiě)下此文章算是對(duì)自己的一個(gè)記錄。也希望觀看到此文章的人少走一點(diǎn)彎路，哪怕只有一步。

當(dāng)然有精力的話最好還是使用Client自己實(shí)現(xiàn)一遍對(duì)自己來(lái)說(shuō)也是一種提高，生產(chǎn)環(huán)境，我建議使用Spring Integration 集成

關(guān)于Spring Intergration

介紹關(guān)于Spring Integration 的一些概念性的東西，如果只想知道怎么集成可跳過(guò)該部分

Spring Integration provides an extension of the Spring programming model to support the well known Enterprise Integration Patterns. It enables lightweight messaging within Spring-based applications and supports integration with external systems through declarative adapters. Those adapters provide a higher level of abstraction over Spring’s support for remoting, messaging, and scheduling.

大致意思是

Spring Integration 為許多著名的企業(yè)級(jí)集成方案提供了擴(kuò)展接口
它能夠在基于Sping的應(yīng)用中進(jìn)行輕量級(jí)消息傳遞
支持聲明式適配器與發(fā)布系統(tǒng)進(jìn)行集成（見(jiàn)Mqtt send消息的實(shí)現(xiàn)）
這些適配器在 Spring對(duì)遠(yuǎn)程調(diào)用，消息傳遞和系統(tǒng)調(diào)度提供了更高層次的抽象

Spring Intergration核心組件

Message（消息）

Message是對(duì)消息的包裝，在Spring 系統(tǒng)中傳遞的任何消息都會(huì)被包裝為Message?？梢岳斫鉃槭荢pring Integration消息傳遞的基本單位

Message Channel（消息管道）

消息管道：Message 在Message Channel中進(jìn)行傳遞，生產(chǎn)者向管道中投遞消息，消費(fèi)者從管道中取出消息。Spring Integration 支持兩種消息傳遞模型，point-to-point（點(diǎn)對(duì)點(diǎn)模型），Publish-subscribe（發(fā)布訂閱模型）有多種管道類型。

本次使用點(diǎn)對(duì)點(diǎn)模型，消息管道類型DirectChannel

Message Endpoint（消息切入點(diǎn)）

消息切入點(diǎn)：消息在管道中流動(dòng)那必定會(huì)有某些流入或流出的點(diǎn)亦或是在某個(gè)位置（即特定函數(shù)）需要對(duì)消息進(jìn)行處理，過(guò)濾，格式轉(zhuǎn)換等。這些點(diǎn)即為Message Endpoint（實(shí)際為某些處理函數(shù)），例如消息發(fā)送，消息接收都是Message Endpoint。

Message Transformer（消息轉(zhuǎn)化器）

消息轉(zhuǎn)化器：是將消息進(jìn)行特定轉(zhuǎn)換例如將一個(gè) Object 序列化為 Json 字符串

Message Filter

消息過(guò)濾器，過(guò)濾掉特定消息。例如在管道中發(fā)送的含username 和 age 屬性的 User 對(duì)象，如果當(dāng)前消息（一個(gè)User實(shí)例的包裝）的age < 18則將其過(guò)濾掉，那么處在過(guò)濾器之后的消費(fèi)者將無(wú)法接收到 age < 18的User對(duì)象。當(dāng)然過(guò)濾條件不僅是消息負(fù)載的屬性，也可以是消息本身的屬性。

Message Router（消息路由）

消息路由：向管道投遞消息時(shí)可由消息路由根據(jù)路由規(guī)則選擇投遞給那個(gè)管道

Splitter（分割器）

分割器：它從一個(gè)輸入管道接收一條消息并將其分割為多條消息，再將每條消息發(fā)送到不同的輸出管道上

Aggregator（聚合器）

聚合器：與分割器功能剛好相反

Service Activator

Service Activator: 它是一個(gè)用于將系統(tǒng)服務(wù)實(shí)例接入到消息系統(tǒng)的泛型切入點(diǎn)，該切入點(diǎn)必須配置輸入管道。其返回值可是消息類型也可以是一個(gè)消息處理器，當(dāng)返回值為消息類型時(shí)需要指定輸出管道，即在該切入點(diǎn)對(duì)消息加工處理后再發(fā)送到指定的輸出管道，如果返回值為消息處理器。那么消息交由消息處理器進(jìn)行處理。下文中會(huì)為Mqtt消息出站配置Service Activator并且返回值為消息處理器

Channel Adapter（管道適配器）

管道適配器：因?yàn)橥獠繀f(xié)議有無(wú)數(shù)種，消息適配器則用于連接不同協(xié)議的外部系統(tǒng)。從外部系統(tǒng)讀入數(shù)據(jù)并對(duì)數(shù)據(jù)進(jìn)行處理最終與Spring Integration 內(nèi)部的消息系統(tǒng)適配。例如將要進(jìn)行Mqtt集成，那么就需要一個(gè)Mqtt的管道適配器，事實(shí)上也確實(shí)有一個(gè)，下文中將會(huì)看到。

開(kāi)始集成

依賴管理工具使用Gardle

引入spring-integration-mqtt依賴

implementation "org.springframework.integration:spring-integration-mqtt:5.4.6"

創(chuàng)建Mqtt配置類

@Configuration
public class MqttConfig {
    /**
     *  以下屬性將在配置文件中讀取
     **/
    //mqtt Broker 地址
    private String[] uris;
    //連接用戶名
    private String username;
    //連接密碼
    private String password;
    //入站Client ID
    private String inClientId;
    //出站Client ID
    private String outClientId;
    //默認(rèn)訂閱主題
    private String defaultTopic;

    public void setUris(String[] uris) {
        this.uris = uris;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public void setInClientId(String inClientId) {
        this.inClientId = inClientId;
    }

    public void setOutClientId(String outClientId) {
        this.outClientId = outClientId;
    }

    public void setDefaultTopic(String defaultTopic) {
        this.defaultTopic = defaultTopic;
    }
}

這里需要注意為什么創(chuàng)建兩個(gè)client ID,Spring Integration 在集成的時(shí)候入站與出站消息處理并不使用同一個(gè)連接，所以如果clien ID相同，將會(huì)出現(xiàn)Mqtt反復(fù)重連現(xiàn)象，實(shí)為 mqtt 出入站連接交替踢對(duì)方下線。

修改配置文件 application.yml

mqtt:
  uris: tcp://ip:port
  username: user
  password: pwd
  in-client-id: ${random.value} # 隨機(jī)值，使出入站 client ID 不同
  out-client-id: ${random.value}
  default-topic: defaultTopic

在MqttConfig上使用注解 @ConfigurationProperties(prefix = "mqtt")將配置文件中屬性注入到MqttConfig中,但別忘記在啟動(dòng)類上使用@EnableConfigurationProperties啟用屬性注入。

@Configuration
@ConfigurationProperties(prefix = "mqtt")
public class MqttConfig {
    .....
}

創(chuàng)建三個(gè)管道

這三個(gè)管道分別用于處理入站消息，出站消息，錯(cuò)誤消息

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

添加 Mqtt 客戶端工廠

@Bean
public MqttPahoClientFactory mqttPahoClientFactory(){
    DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
    MqttConnectOptions options = new MqttConnectOptions();
    options.setServerURIs(uris);
    options.setUserName(username);
    options.setPassword(password.toCharArray());
    factory.setConnectionOptions(options);
    return factory;
}

添加 Mqtt 管道適配器

@Bean
public MqttPahoMessageDrivenChannelAdapter adapter(MqttPahoClientFactory factory){
    return new MqttPahoMessageDrivenChannelAdapter(inClientId,factory,defaultTopic);
}

添加消息生產(chǎn)者

@Bean
public MessageProducer mqttInbound(MqttPahoMessageDrivenChannelAdapter adapter){
    adapter.setCompletionTimeout(5000);
    adapter.setConverter(new DefaultPahoMessageConverter());
    //入站投遞給入站管道
    adapter.setOutputChannel(mqttInboundChannel());
    adapter.setErrorChannel(errorChannel());
    adapter.setQos(0);
    return adapter;
}

添加出站處理器

出站處理器是一個(gè)Service Activator

@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel")
public MessageHandler mqttOutbound(MqttPahoClientFactory factory){
    MqttPahoMessageHandler handler =
            new MqttPahoMessageHandler(outClientId,factory);
    handler.setAsync(true);
    handler.setConverter(new DefaultPahoMessageConverter());
    handler.setDefaultTopic(defaultTopic);
    return handler;
}

添加消息接收器

通過(guò)前文的配置，當(dāng)Mqtt 訂閱主題產(chǎn)生消息時(shí)會(huì)通過(guò) MessageProducer（本例中是一個(gè)管道適配器）將消息投遞到入站管道中，所以當(dāng)需要接收并處理Mqtt消息時(shí)只需要從入站管道中取出消息即可。取出消息即可使用前文的 Endpoint 本例使用Service Activator

創(chuàng)建一個(gè)接收類，自定義任意類型。重點(diǎn)是使用其內(nèi)部的方法，將其注冊(cè)為Endpoint

@Component
public class Receiver {
    @Bean
    //使用ServiceActivator 指定接收消息的管道為 mqttInboundChannel，投遞到mqttInboundChannel管道中的消息會(huì)被該方法接收并執(zhí)行
    @ServiceActivator(inputChannel = "mqttInboundChannel")
    public MessageHandler handleMessage() {
        return message -> {
            System.out.println(message.getPayload());
        };
    }
}

至此，整個(gè)服務(wù)即可接收Mqtt broker的消息了。默認(rèn)接收的主題為配置文件中指定的 "defaultTopic"。

添加消息發(fā)送器

那么消息發(fā)送器無(wú)疑是向出站管道投遞消息即可。如何實(shí)現(xiàn)，Spring Integration 提供了 @MessagingGateway注解，該注解提供一個(gè)defaultRequestChannel屬性用于指定出站管道。如下

@Component
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttSender {
    void sendToMqtt(String text);

    void sendWithTopic(@Header(MqttHeaders.TOPIC) String topic, String text);

    void sendWithTopicAndQos(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) Integer Qos, String text);
}

此接口內(nèi)函數(shù)參數(shù)將作為消息的負(fù)載被包裝成為消息并投遞到出站管道中。同時(shí)可以看到，方法參數(shù)可以注解的方式自定義消息發(fā)送的 topic qos retain 等屬性。

至此集成完成，用戶可調(diào)用MqttSender發(fā)布消息或是沖Receiver中接收并處理消息

自定義編解碼器

前文中添加消息生產(chǎn)者和添加出站處理器代碼中都可以看到setConverter(new DefaultPahoMessageConverter());方法，該方法用與對(duì)消息負(fù)載進(jìn)行編解碼。多數(shù)情況下我們的消息都是可編碼的。我在消息傳遞過(guò)程中使用的是json編碼。下面我將以json編碼為例，展示如何自定義編解碼器

//協(xié)議對(duì)象
public class User {
    private String username;
    private String password;

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }
}

實(shí)現(xiàn)自己的編解碼器

代碼如下，詳見(jiàn)注釋

@Component
public class AlMingConverter implements MqttMessageConverter {
    private final  static Logger log = LoggerFactory.getLogger(AlMingConverter.class);
    private int defaultQos = 0;
    private boolean defaultRetain = false;
    ObjectMapper om = new ObjectMapper();
    //入站消息解碼
    @Override
    public Message<User> toMessage(String topic, MqttMessage mqttMessage) {
        
        User protocol = null;
        try {
            protocol = om.readValue(mqttMessage.getPayload(), User.class);
        } catch (IOException e) {
            if (e instanceof JsonProcessingException) {
                System.out.println();
                log.error("Converter only support json string");
            }
        }
        assert protocol != null;
        MessageBuilder<User> messageBuilder = MessageBuilder
                .withPayload(protocol);
        //使用withPayload初始化的消息缺少頭信息，將原消息頭信息填充進(jìn)去
        messageBuilder.setHeader(MqttHeaders.ID, mqttMessage.getId())
                .setHeader(MqttHeaders.RECEIVED_QOS, mqttMessage.getQos())
                .setHeader(MqttHeaders.DUPLICATE, mqttMessage.isDuplicate())
                .setHeader(MqttHeaders.RECEIVED_RETAINED, mqttMessage.isRetained());
        if (topic != null) {
            messageBuilder.setHeader(MqttHeaders.TOPIC, topic);
        }
        return messageBuilder.build();
    }
    //出站消息編碼
    @Override
    public Object fromMessage(Message<?> message, Class<?> targetClass) {
        MqttMessage mqttMessage = new MqttMessage();
        String msg = null;
        try {
            msg = om.writeValueAsString(message.getPayload());
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        assert msg != null;
        mqttMessage.setPayload(msg.getBytes(StandardCharsets.UTF_8));
        //這里的 mqtt_qos ,和 mqtt_retained 由 MqttHeaders 此類得出不可以隨便取，如需其他屬性自行查找
        Integer qos = (Integer) message.getHeaders().get("mqtt_qos");
        mqttMessage.setQos(qos == null ? defaultQos : qos);
        Boolean retained = (Boolean) message.getHeaders().get("mqtt_retained");
        mqttMessage.setRetained(retained == null ? defaultRetain : retained);
        return mqttMessage;
    }
    //此方法直接拿默認(rèn)編碼器的來(lái)用的，照抄即可
    @Override
    public Message<?> toMessage(Object payload, MessageHeaders headers) {
        Assert.isInstanceOf(MqttMessage.class, payload,
                () -> "This converter can only convert an 'MqttMessage'; received: " + payload.getClass().getName());
        return this.toMessage(null, (MqttMessage) payload);
    }

    public void setDefaultQos(int defaultQos) {
        this.defaultQos = defaultQos;
    }

    public void setDefaultRetain(boolean defaultRetain) {
        this.defaultRetain = defaultRetain;
    }
}

修改MqttConfig

將自定義編解碼器注入到MqttConfig中。

private final AlMingConverter alMingConverter;
@Autowired
public MqttConfig(AlMingConverter alMingConverter) {
    this.alMingConverter = alMingConverter;
}

將原來(lái)所有DefaultPahoMessageConverter實(shí)例更換為alMingConverter

...
adapter.setConverter(alMingConverter);
...
handler.setConverter(alMingConverter);

修改MqttSender

@Component
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttSender {
    void sendToMqtt(User user);

    void sendWithTopic(@Header(MqttHeaders.TOPIC) String topic, User user);

    void sendWithTopicAndQos(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) Integer Qos, User user);
}

Receiver修改

此時(shí)Reveiver的消息負(fù)載雖然編譯時(shí)為 Object ,但運(yùn)行時(shí)為User可安全的進(jìn)行 Object 到 User 的強(qiáng)轉(zhuǎn)然后進(jìn)行后續(xù)操作

@Component
public class Receiver {
    @Bean
    @ServiceActivator(inputChannel = "mqttInboundChannel")
    public MessageHandler handleMessage() {
        return message -> {
            User user = (User) message.getPayload();
            System.out.println(user.getUsername()+":"+user.getPassword());
        };
    }
}

動(dòng)態(tài)修改訂閱主題

前文中發(fā)布訂閱的主題都是默認(rèn)主題，即在配置文件中指定的主題。生產(chǎn)環(huán)境肯定不止使用一個(gè)主題。那么就需要運(yùn)行時(shí)動(dòng)態(tài)修改主題。Spring Integration也確實(shí)為此提供了支持。上文中向Bean容器中注冊(cè)的管道適配器（MqttPahoMessageDrivenChannelAdapter）提供了addTopic removeTopic等方法可用于運(yùn)行時(shí)修改主題。所以可以創(chuàng)建一個(gè)MqttService專門(mén)用于添加和刪除主題。

但注意只有Spring Integration 4.1以上版本可以這么使用

@Service
public class MqttServiceImpl implements MqttService { //MqttService是自己定義的，僅包含如下方法均已重寫(xiě)
    MqttPahoMessageDrivenChannelAdapter adapter;
    @Autowired
    public MqttServiceImpl(MqttPahoMessageDrivenChannelAdapter adapter) {
        this.adapter = adapter;
    }

    @Override
    public void addTopic(String topic) {
        String[] topics = adapter.getTopic();
        if(!Arrays.asList(topics).contains(topic)){
            adapter.addTopic(topic,0);
        }
    }

    @Override
    public void removeTopic(String topic) {
        adapter.removeTopic(topic);
    }
}

增加或刪除主題時(shí)，注入該服務(wù)調(diào)用addTopic，removeTopic即可

End&附錄

最終MqttConfig完整代碼

@Configuration
@ConfigurationProperties(prefix = "mqtt")
public class MqttConfig {
    /**
     *  以下屬性將在配置文件中讀取
     **/
    //mqtt Broker 地址
    private String[] uris;
    //連接用戶名
    private String username;
    //連接密碼
    private String password;
    //入站Client ID
    private String inClientId;
    //出站Client ID
    private String outClientId;
    //默認(rèn)訂閱主題
    private String defaultTopic;

    private final AlMingConverter alMingConverter;
    @Autowired
    public MqttConfig(AlMingConverter alMingConverter) {
        this.alMingConverter = alMingConverter;
    }

    @Bean
    public MqttPahoClientFactory mqttPahoClientFactory(){
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(uris);
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        factory.setConnectionOptions(options);
        return factory;
    }
    @Bean
    public MqttPahoMessageDrivenChannelAdapter adapter(MqttPahoClientFactory factory){
        return new MqttPahoMessageDrivenChannelAdapter(inClientId,factory,defaultTopic);
    }
    public void setUris(String[] uris) {
        this.uris = uris;
    }
    @Bean
    public MessageProducer mqttInbound(MqttPahoMessageDrivenChannelAdapter adapter){
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(alMingConverter);
        //入站投遞的通道
        adapter.setOutputChannel(mqttInboundChannel());
        adapter.setErrorChannel(errorChannel());
        adapter.setQos(0);
        return adapter;
    }
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound(MqttPahoClientFactory factory){
        MqttPahoMessageHandler handler =
                new MqttPahoMessageHandler(outClientId,factory);
        handler.setAsync(true);
        handler.setConverter(alMingConverter);
        handler.setDefaultTopic(defaultTopic);
        return handler;
    }
    @Bean
    public MessageChannel mqttOutboundChannel(){
        return new DirectChannel();
    }
    @Bean
    public MessageChannel mqttInboundChannel(){
        return new DirectChannel();
    }
    @Bean
    public MessageChannel errorChannel(){
        return new DirectChannel();
    }
    public void setUsername(String username) {
        this.username = username;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public void setInClientId(String inClientId) {
        this.inClientId = inClientId;
    }

    public void setOutClientId(String outClientId) {
        this.outClientId = outClientId;
    }

    public void setDefaultTopic(String defaultTopic) {
        this.defaultTopic = defaultTopic;
    }
}

最后

到此這篇關(guān)于如何使用Spring integration在Springboot中集成Mqtt的文章就介紹到這了,更多相關(guān)Spring integration在Springboot集成Mqtt內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

国产无遮挡裸体免费直播视频,久久精品国产蜜臀av,动漫在线视频一区二区,欧亚日韩一区二区三区,久艹在线 免费视频,国产精品美女网站免费,正在播放 97超级视频在线观看,斗破苍穹年番在线观看免费,51最新乱码中文字幕