要想使用MQ的MQTT服务需要先开启MQTT服务,因为RabbitMQ的MQTT默认是关闭的,具体启动方法可以参考:rabbitmq使用mqtt协议_panda_225400的博客-CSDN博客_rabbitmq mqtt
下面具体实现我就直接贴代码吧,一切说明都在代码里面,方便直接
POM依赖
<dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-mqtt</artifactId> </dependency>
配置类
#RabbitMQ缓存配置MQTT
#MQTT默认端口1883,和RabbitMQ不使用同一个端口,具体端口怎么看,可以看我下面的截图
spring.rabbitmq.url=tcp://127.0.0.1:1883
#MQ登录账号
spring.rabbitmq.username=root
#MQ登录密码
spring.rabbitmq.password=123456
spring.rabbitmq.pubTopic=pubTopic
#要订阅的频道,多个用逗号隔开 admin/+/admin表示中间的加号我可以替换任何字符
spring.rabbitmq.subTopic=subTopic,test,admin/+/admin
#默认写法
spring.rabbitmq.clientId=${random.value}
HTTP处理相关类
package com.smart.tcp.mqtt;
import com.smart.tcp.config.MqttConfig;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Objects;
/**
* mqtt服务类
* 一种基于发布/订阅(publish/subscribe)模式的轻量级通讯协议,通过订阅相应的主题来获取消息,
* 是物联网(Internet of Thing)中的一个标准传输协议
* ClientId是MQTT客户端的标识。MQTT服务端用该标识来识别客户端。因此ClientId必须是独立的。
* clientID需为全局唯一。如果不同的设备使用相同的clientID同时连接物联网平台,那么先连接的那个设备会被强制断开。
*
* @author WeiWei
* @date 2022/08/08
*/
@Slf4j
@Configuration
public class MqttServer {
/**
* 出站通道
*/
public static final String OUTBOUND_CHANNEL = "mqttOutboundChannel";
/**
* 输入通道
*/
public static final String INPUT_CHANNEL = "mqttInputChannel";
/**
* mqtt配置
*/
@Resource
private MqttConfig mqttConfig;
/**
* 初始化
*/
@PostConstruct
public void init() {
log.info(mqttConfig.toString());
}
/**
* mqtt客户工厂
*
* @return {@link MqttPahoClientFactory}
*/
@Bean
public MqttPahoClientFactory clientFactory() {
MqttConnectOptions options = new MqttConnectOptions();
//配置MqttConnectOptions
options.setServerURIs(new String[]{mqttConfig.getUrl()});
options.setUserName(mqttConfig.getUsername());
options.setPassword(mqttConfig.getPassword().toCharArray());
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setConnectionOptions(options);
return factory;
}
/**
* mqtt出站通道
*
* @return {@link MessageChannel}
*/
@Bean(value = OUTBOUND_CHANNEL)
public MessageChannel mqttOutboundChannel() {
return new DirectChannel();
}
/**
* mqtt出站handler 1
*
* @return {@link MessageHandler}
*/
@Bean
@ServiceActivator(inputChannel = OUTBOUND_CHANNEL)
public MessageHandler mqttOutboundHandler() {
//MqttPahoMessageHandler初始化
MqttPahoMessageHandler handler = new MqttPahoMessageHandler(mqttConfig.getClientId(), clientFactory());
//设置默认的qos级别
handler.setDefaultQos(1);
//保留标志的默认值。如果没有mqtt_retained找到标题,则使用它。如果提供了自定义,则不使用它converter。这里不启用
handler.setDefaultRetained(false);
//设置发布的主题
handler.setDefaultTopic(mqttConfig.getPubTopic());
//当 时true,调用者不会阻塞。相反,它在发送消息时等待传递确认。默认值为false(在确认交付之前发送阻止)。
handler.setAsync(false);
//当 async 和 async-events 都为 true 时,会发出 MqttMessageSentEvent(请参阅事件)。它包含消息、主题、客户端库生成的messageId、clientId和clientInstance(每次连接客户端时递增)。当客户端库确认交付时,会发出 MqttMessageDeliveredEvent。它包含 messageId、clientId 和 clientInstance,使传递与发送相关联。任何 ApplicationListener 或事件入站通道适配器都可以接收这些事件。请注意,有可能在 MqttMessageSentEvent 之前接收到 MqttMessageDeliveredEvent。默认值为false。
handler.setAsyncEvents(false);
return handler;
}
public MessageHandler mqttOutboundHandler123(String code,String topic) {
//MqttPahoMessageHandler初始化
MqttPahoMessageHandler handler = new MqttPahoMessageHandler(code, clientFactory());
//设置默认的qos级别
handler.setDefaultQos(1);
//保留标志的默认值。如果没有mqtt_retained找到标题,则使用它。如果提供了自定义,则不使用它converter。这里不启用
handler.setDefaultRetained(false);
//设置发布的主题
handler.setDefaultTopic(topic);
//当 时true,调用者不会阻塞。相反,它在发送消息时等待传递确认。默认值为false(在确认交付之前发送阻止)。
handler.setAsync(false);
//当 async 和 async-events 都为 true 时,会发出 MqttMessageSentEvent(请参阅事件)。它包含消息、主题、客户端库生成的messageId、clientId和clientInstance(每次连接客户端时递增)。当客户端库确认交付时,会发出 MqttMessageDeliveredEvent。它包含 messageId、clientId 和 clientInstance,使传递与发送相关联。任何 ApplicationListener 或事件入站通道适配器都可以接收这些事件。请注意,有可能在 MqttMessageSentEvent 之前接收到 MqttMessageDeliveredEvent。默认值为false。
handler.setAsyncEvents(false);
return handler;
}
/**
* mqtt输入通道
*
* @return {@link MessageChannel}
*/
@Bean
public MessageChannel mqttInputChannel() {
return new DirectChannel();
}
/**
* 入站
*
* @return {@link MessageProducer}
*/
@Bean
public MessageProducer inbound() {
//配置订阅端MqttPahoMessageDrivenChannelAdapter
MqttPahoMessageDrivenChannelAdapter adapter =
new MqttPahoMessageDrivenChannelAdapter(
mqttConfig.getClientId() + "_inbound", clientFactory(), mqttConfig.getSubTopic().split(","));
//设置超时时间
adapter.setCompletionTimeout(3000);
//设置默认的消息转换类
adapter.setConverter(new DefaultPahoMessageConverter());
//设置qos级别
adapter.setQos(1);
//设置入站管道
adapter.setOutputChannel(mqttInputChannel());
return adapter;
}
/**
* 消息处理程序
*
* @return {@link MessageHandler}
*/
@Bean
@ServiceActivator(inputChannel = INPUT_CHANNEL)
public MessageHandler messageHandler() {
return message -> {
String topic = Objects.requireNonNull(message.getHeaders().get("mqtt_receivedTopic")).toString();
log.info("主题:"+topic+"+接收到该主题消息为: {}", message.getPayload());
};
}
}
网关类
package com.smart.tcp.mqtt;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
/**
* mqtt网关(发布端需要用到)
*
* @author WeiWei
* @date 2022/08/08
*/
@Component
@MessagingGateway(defaultRequestChannel = MqttServer.OUTBOUND_CHANNEL)
public interface MqttGateway {
/**
* 发送到mqtt
*
* @param payload 有效载荷
*/
void sendToMqtt(String payload);
/**
* 发送到mqtt
*
* @param topic 主题
* @param payload 消息内容
*/
void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);
/**
* 发送到mqtt
*
* @param topic 主题
* @param qos qos
* @param payload 消息内容
*/
void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
}
控制层测试类文章来源:https://www.toymoban.com/news/detail-654603.html
package com.smart.tcp.controller;
import com.ruoyi.common.model.BaseResult;
import com.smart.server.api.mall.AddressApi;
import com.smart.server.domain.Address;
import com.smart.tcp.mqtt.MqttGateway;
import com.smart.tcp.mqtt.MqttServer;
import com.smart.tcp.server.ChargeServer;
import com.smart.tcp.server.EquipmentApiServer;
import org.apache.dubbo.config.annotation.DubboReference;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.util.Assert;
import org.springframework.web.bind.annotation.*;
import java.util.Date;
import java.util.List;
/**
* @author HayDen
* @date 2022-11-15
*/
@RestController
@RequestMapping("/api/top")
public class TopController
{
@Autowired
private MqttGateway mqttGateway;
/**
* 发送MQTT消息
*
* @param message 消息内容
* @return 返回
*/
@PostMapping(value = "/mqtt", produces = MediaType.APPLICATION_JSON_VALUE)
public ResponseEntity<String> sendMqtt(@RequestParam(value = "msg") String message) {
System.out.println("================生产MQTT消息================" + message);
mqttGateway.sendToMqtt(message);
return new ResponseEntity<>("OK", HttpStatus.OK);
}
/**
* 发送MQTT消息
* @param key 表示要发给那个频道,例(admin/123/admin、admin/sss/admin、test、subTopic.............)
* @param message 消息内容
* @return 返回
*/
@PostMapping(value = "/mqtt2", produces = MediaType.APPLICATION_JSON_VALUE)
public ResponseEntity<String> sendMqtt2(@RequestParam(value = "key")String key,@RequestParam(value = "msg") String message) {
System.out.println("================生产MQTT消息================" + message);
mqttGateway.sendToMqtt(key, message);
return new ResponseEntity<>("OK", HttpStatus.OK);
}
}
配置文件 文章来源地址https://www.toymoban.com/news/detail-654603.html
#RabbitMQ缓存配置MQTT #MQTT默认端口1883 spring.rabbitmq.url=tcp://127.0.0.1:1883 spring.rabbitmq.username=root spring.rabbitmq.password=123456 spring.rabbitmq.pubTopic=pubTopic #这里表示需要处理MQTT发送的主题key,+号表示中间任意字符,当然还有#号通配符 # +号admin/+/admin :admin/a/admin 或者 admin/b/admin 或者 admin/c/admin等等 # #号admin/# :admin/a/a 或者 admin/b/a 或者 admin/a/b等等(可以替换后面两位为任意字符) spring.rabbitmq.subTopic=subTopic,test,admin/+/admin #默认写法 spring.rabbitmq.clientId=${random.value}
到了这里,关于Java基于RabbitMQ实现MQTT的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!