设备通过RabbitMQ接入自建云平台

RabbitMQ开启MQTT协议

    设备端使用的是MQTT协议进行通信，当前系统的基础设施已经有了RabbitMQ，思量到维护，所以通过RabbitMQ来将设备接入到系统中。
    当前RabbitMQ的版本为3.9.16，可通过插件开启MQTT协议，但是当前版本仅支持MQTT3版本协议，较新的RabbitMQ支持MQTT5。

• 开启MQTT协议插件下令
  

1. rabbitmq-plugins enable rabbitmq_mqtt

复制代码

• 配置MQTT插件
  在RabbitMQ的配置文件中加入MQTT配置，默认的配置文件位置在/etc/rabbitmq/rabbitmq.conf，没有的话可以创建该配置文件。具体配置可检察RabbitMQ官网文档。
  

1. # MQTT config
2. mqtt.exchange = device_mqtt # 指定MQTT协议的交换机，如果不指定就使用默认的交换机
3. mqtt.tcp.default = 1883 # 设置MQTT监听的端口（默认是1883）
4. mqtt.allow_anonymous = false # 允许匿名访问（生产环境中通常建议使用用户名和密码）。
5. mqtt.vhost = / # 设置MQTT虚拟主机。
6. mqtt.exchange_durable = true # 设置交换机是否持久化。
7. mqtt.exchange_auto_delete = false # 设置交换机是否在最后一个队列删除后自动删除。

复制代码

• 重启RabbitMQ服务
  

1. sudo systemctl restart rabbitmq-server
2. # 或者
3. sudo service rabbitmq-server restart

复制代码

自定义客户端毗连鉴权

    由于设备端的4G模块已经封装死了MQTT毗连，不能对暗码生成规则进行更改，所以更改RabbitMQ端的鉴权规则。
    编写RabbitMQ的鉴权规则有一定难度，同时也不够灵活，将鉴权逻辑移到系齐备一的鉴权中央比较符合。RabbitMQ本身是不支持调用外部http接口进行鉴权，可以通过rabbitmq_auth_backend_http插件实现调用外部http接口鉴权。具体配置可以看插件github地址

• 启用插件
  

1. sudo rabbitmq-plugins enable rabbitmq_auth_backend_http
3. sudo rabbitmq-plugins enable rabbitmq_auth_backend_cache

复制代码

• 编辑RabbitMQ配置
  修改前面提到的/etc/rabbitmq/rabbitmq.conf配置文件。
  

1. # 首先使用cache（缓存）进行认证
2. auth_backends.1 = cache
3. # 其次使用RabbitMQ内置认证
4. auth_backends.2 = internal
5. # 该配置指定了缓存后端所依赖的实际认证后端为 http。也就是说，当使用 cache 认证后端时，它会从 http 后端获取认证信息并进行缓存。如果缓存中存在有效的认证信息，则直接使用缓存数据，否则会通过 http 后端去获取并更新缓存。
6. auth_cache.cached_backend = http
7. # 认证请求类型
8. auth_http.http_method   = post
9. # 用于获取用户认证信息的 http 地址
10. auth_http.user_path = http://localhost:8080/v1/rabbitmqAuth/user
11. # 用于虚拟主机（vhost）相关认证和授权的 http 地址
12. auth_http.vhost_path = http://localhost:8080/v1/rabbitmqAuth/vhost
13. # 资源相关的认证和授权的 http 地址
14. auth_http.resource_path = http://localhost:8080/v1/rabbitmqAuth/resource
15. # 主题（topic）相关的认证和授权的 http 地址
16. auth_http.topic_path = http://localhost:8080/v1/rabbitmqAuth/topic
17. # 缓存时间，单位毫秒
18. auth_cache.cache_ttl = 60000

复制代码

• 实现认证逻辑
  这里以Spring应用为例。也可以检察官方认证模块工程。RabbitMqAuthBackendHttpController类，接口入参检察官方文档。
  

1. @Slf4j
2. @RestController
3. @RequestMapping(path = "/v1/rabbitmqAuth")
4. public class RabbitMqAuthBackendHttpController {
5.     @Autowired
6.     private RabbitmqAuthService rabbitmqAuthService;
8.     @ApiOperation("校验用户信息")
9.     @PostMapping("/user")
10.     public String user(@RequestParam Map<String,String> userCheck) {
11.         return rabbitmqAuthService.checkUser(userCheck);
12.     }
13.     @ApiOperation("校验vhost")
14.     @PostMapping("/vhost")
15.     public String vhost(VirtualHostCheck check) {
16.         return rabbitmqAuthService.checkVhost(check);
17.     }
18.     @ApiOperation("校验资源")
19.     @PostMapping("/resource")
20.     public String resource(ResourceCheck check) {
21.         return rabbitmqAuthService.checkResource(check);
22.     }
23.     @ApiOperation("校验Topic")
24.     @PostMapping("/topic")
25.     public String topic(TopicCheck check) {
26.         return rabbitmqAuthService.checkTopic(check);
27.     }
28. }

复制代码

    鉴权核心逻辑在RabbitmqAuthServiceImpl类

1. @Slf4j
2. @Service
3. public class RabbitmqAuthServiceImpl implements RabbitmqAuthService {
4.     /**
5.      * 校验成功
6.      */
7.     private static final String SUCCESS="allow";
8.     /**
9.      * 校验失败
10.      */
11.     private static final String FAIL="deny";
12.     @Autowired
13.     private IDeviceRabbitmqAuthService deviceRabbitmqAuthService;
14.     /**
15.      * 校验RabbitMQ用户
16.      * @param userCheck
17.      * @return
18.      * <ul>
19.      *     <li>deny:拒绝访问 user / vhost / resource</li>
20.      *     <li>allow:允许访问user / vhost / resource</li>
21.      * </ul>
22.      */
23.     @Override
24.     public String checkUser(Map<String, String> userCheck) {
25.         String username = userCheck.get("username");
26.         if (ObjectUtil.isNotEmpty(username) && !username.contains("&")){
27.             // 业务需要，只有部分账号的校验才需要http远程调用
28.             return FAIL;
29.         }
30.         String password=  userCheck.get("password");
31.         String vhost=  userCheck.get("vhost");
32.         String clientId= userCheck.get("client_id");
33.         // 加载配置的账号信息
34.         DeviceRabbitmqAuth deviceRabbitmqAuth = deviceRabbitmqAuthService.selectDeviceRabbitmqAuthByDevice(username);
35.         // 验证设备连接
36.         checkDeviceConnect(password,clientId,deviceRabbitmqAuth);
37.         // 转换权限
38.         List<String> tags = RabbitMQTagsEnums.permissionCovertTag(deviceRabbitmqAuth.getPermission());
39.         // 校验成功
40.         return SUCCESS+" " + StringUtils.collectionToDelimitedString(tags, " ");
41.     }
43.     /**
44.      * 校验设备端连接
45.      * @param password 设备连接的密码
46.      * @param clientId 设备连接的clientId
47.      * @param deviceRabbitmqAuth 存储的设备信息
48.      * @return
49.      */
50.     public void checkDeviceConnect(String password,String clientId,DeviceRabbitmqAuth deviceRabbitmqAuth) {
51.         if (deviceRabbitmqAuth==null){
52.             throw new RabbitmqAuthException("未找到连接信息");
53.         }
54.         // TDO 不同4G模块使用加密算法、加密方式不同，根据设备自定义账号校验逻辑
55.     }
56.     /**
57.      * 校验vhost
58.      * @param virtualHostCheck
59.      * @return
60.      * <ul>
61.      *     <li>deny:拒绝访问 user / vhost / resource</li>
62.      *     <li>allow:允许访问user / vhost / resource</li>
63.      * </ul>
64.      */
65.     @Override
66.     public String checkVhost(VirtualHostCheck virtualHostCheck) {
67.         log.debug("RabbitMqCheck 校验vhost：{}", JSONObject.toJSONString(virtualHostCheck));
68.         return SUCCESS;
69.     }
70.     /**
71.      * 资源认证
72.      * @param resourceCheck
73.      * @return
74.      * <ul>
75.      *     <li>deny:拒绝访问 user / vhost / resource</li>
76.      *     <li>allow:允许访问user / vhost / resource</li>
77.      * </ul>
78.      */
79.     @Override
80.     public String checkResource(ResourceCheck resourceCheck) {
81.         log.debug("RabbitMqCheck 校验资源：{}", JSONObject.toJSONString(resourceCheck));
82.         return SUCCESS;
83.     }
84.     /**
85.      * 主题认证
86.      * @param topicCheck
87.      * @return
88.      * <ul>
89.      *     <li>deny:拒绝访问 user / vhost / resource</li>
90.      *     <li>allow:允许访问user / vhost / resource</li>
91.      * </ul>
92.      */
93.     @Override
94.     public String checkTopic(TopicCheck topicCheck) {
95.         log.debug("RabbitMqCheck 校验Topic：{}", JSONObject.toJSONString(topicCheck));
96.         return SUCCESS;
97.     }
98. }

复制代码

• 重启RabbitMQ服务
  

1. sudo systemctl restart rabbitmq-server
2. # 或者
3. sudo service rabbitmq-server restart

复制代码

RabbitMQ鉴权流程：

设备接入系统

设备与系统通信流程：

    设备端使用MQTT协议与系统进行交互。互换时机将MQTT协议数据转换为AMQP协议数据转发到绑定的队列。
   由于一些缘故原由，服务器端监听MQTT协议的数据时，服务器端发送MQTT协议的数据，出现了多个监听队列都收到消息了。所以服务器端监听AMQP协议的数据，只有发送时才使用MQTT协议。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。