MQTT Java入门 -Spring Boot 整合 artemis

Spring Boot MQTT协议

通过spring boot整合apache artemis实现Java语言MQTT协议通信，搭建MQTT服务器可以参考上一篇 MQTT Java入门-搭建MQTT服务器

Spring Boot项目创建

创建一个spring boot项目使用最新的spring boot版本

 

添加artemis依赖

maven pom.xml添加以下依赖：

        <!--MQTT-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-artemis</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-pool</artifactId>
        </dependency>

编写MQTT消息发布者

import lombok.extern.slf4j.Slf4j;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;

/**
 * 消息生产者
 */
@Slf4j
@Service
public class MessageProducerService {

    @Resource
    JmsTemplate jmsTemplate;


    /**
     * 测试 每5秒发送一个消息
     */
    @Scheduled(cron = "0/5 * * * * ?")
    public void sendMessage(){
        Map<String,String> dataMap=new HashMap<>();
        dataMap.put("Hello Artemis","MQTT");
        jmsTemplate.convertAndSend("destinationName1",dataMap);
        log.info("send message success!");
    }

}

用于发布消息到MQTT服务器

编写MQTT消息消费者

import lombok.extern.slf4j.Slf4j;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Service;

import java.util.Map;

@Slf4j
@Service
public class MessageConsumerService {


    @JmsListener(destination = "destinationName1")
    public void receiveMessage(Map<String,String> dataMap){
        try {
            //等待10秒模拟
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.info("Receive Message : {}",dataMap.toString());
    }

}

用于模拟接受MQTT协议消息内容

编写MQTT配置文件

[user]

import org.apache.activemq.ActiveMQConnectionFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.config.JmsListenerContainerFactory;

@Configuration
public class ActiveMQConfig {

    @Value("${spring.activemq.broker-url}")
    private String brokerUrl;

    @Value("${spring.activemq.user}")
    private String user;

    @Value("${spring.activemq.password}")
    private String password;

    @Bean
    public ActiveMQConnectionFactory connectionFactory() {
        return new ActiveMQConnectionFactory(user, password, brokerUrl);
    }

    @Bean
    public JmsListenerContainerFactory<?> jmsListenerContainerQueue(ActiveMQConnectionFactory connectionFactory){
        DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
        bean.setConnectionFactory(connectionFactory);
        return bean;
    }

    @Bean
    public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ActiveMQConnectionFactory connectionFactory){
        DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
        //设置为发布订阅方式, 默认情况下使用的生产消费者方式
        bean.setPubSubDomain(false);
        bean.setConnectionFactory(connectionFactory);
        return bean;
    }

}

[/user]

在启动类注解启动调度任务

@EnableScheduling

启动项目

查看console面板可以看到消息的发送和接受日志


通过Apache ActiveMQ Artemis web管理也可以看到消息队列


项目源码下载：项目源码点击下载（访问密码：9987）
 

MQTT Java MQTT协议 Spring Boot 整合 artemis

http://blog.xqlee.com/article/994.html