简介

开放平台消息通道，用于第三方开发者获取云眸实时性消息，用户基于消息通道可以收到各类消息，具体消息类型参见各平台消息类型定义

消息通道具有高可用分布式集群技术,具备高吞吐、消息堆积、可靠重试等特性

消息通道区别于查询接口，需要开发者轮询拉取，消息通道中的消息最多保留7天，过期没有拉取则会被清理

权限开通

在使用消息通道功能之前，需要先开通对应消息类型的权限。开通方式如图：

核心概念

**消息：**设备或平台产生的各类事件、业务数据的统称

**生产者：**也称为消息发布者，负责生产并发送消息至消息通道，一般理解为设备本身或平台作为消息的生产者

**消费者：**也称为消息订阅者，负责从消息通道消费消息，一般理解为第三方开发者作为消息的消费者

**消费组：**从消息通道从消费消息的消费者分组，目前固定为3个，即：group1,group2,group3，各个消费组之间数据的消费、偏移量的管理互相独立

**偏移量：**消息通道中某个消费组的消费者当前消费消息的位置标记

核心流程

消息通道数据模型

说明：
1.每个消费组的消息数据完全一致，各自独立消费
2.每个消费组最多支持5个消费者，只有1个消费者可以获取到消息
3.禁止多线程消费，否则会导致消息错乱

消费消息流程

说明：

【1】每个消费组最多存在5个消费者，每个消费者有效期5分钟，5分钟内未使用会被销毁；当前消费者数量达到5个时，创建消费者接口会报错，此时需要等待其他消费者失效后才可以重新创建

【2】消费者首次拉取消息时需要与消息通道建立连接（可能会耗时2-3秒），此次调用实际拉到消息列表为空，再次调用即可拉取到消息，30秒内必须再次调用拉取消息接口，否则将断开与消息通道的连接。因此拉取间隔需要在30秒以内，否则每次拉到的消息是空的

消费消息只能在单线程中进行，每个消费组只能有1个消费者可以消费到消息，禁止使用多线程消费

【3】自动提交偏移量的机制是在下一次消费时自动提交上一次消费的偏移量，这是为了避免消费者没有收到返回消息偏移量却提交了的情况

【4】消费消息接口报错的情况，除了无效的consumerId以外，其他情况均可以进行重试

常见问题

消息通道拉取不到消息

问题原因：

1.消息拉取方式不正确，消费者创建之后第一次消费消息时需要建立连接（可能会耗时2-3秒），此时获取到的消息列表是空的

2.拉取消息的间隔超过30秒，消费者建立连接的有效期是30秒，拉取间隔超过30秒会导致每次拉取时都需要建立连接而实际拉到消息列表是空的

3.存在多个消费者同时消费的情况，此时只有1个消费者（最早获取连接的）可以获取到消息

4.消息通道权限未开通或开通不足1小时

解决办法：

1.消费消息的接口多尝试调用几次

2.消费消息的间隔要在30秒以内

3.检查账号下是否存在多个消费者消费的情况

4.登录开发者服务进入秘钥管理页面开启对应的消息权限

消息通道消费到的消息重复

问题原因：偏移量提交不正确

解决办法：

1.如果是采用自动提交偏移量的方式，需要保证消费间隔在30秒以内（自动提交偏移量是在下一次请求是提交上一次消费的偏移量），消费到消息之后的消息处理一定要是异步处理的，否则可能会造成间隔超过30秒，产生消息重复的现象

部分第三方开发者可能在初期数据量不多时感知不到这个问题，后期数据量大了之后如果是同步处理的，很有可能会出现处理时间超过30秒的情况

2.如果是采用手动提交偏移量的方式，需要保证消费到消息之后正确提交偏移量（未消费到消息不可以提交偏移量，可能会造成偏移量被重置）

3.可根据消息的msgID去重

demo

demo下载地址：点击下载

demo使用说明：

1.自定义处理消息的类

         public class MyPushMessageService implements IPushMessageService {
             @Override
             public void pushMessage(List<Message> messageList) {
                 // TODO Auto-generated method stub
                 // 异步处理消费到的消息
            }
         }

消息异步处理非常重要，否则可能在消息量大了之后处理时间过长导致获取到重复消息

2.初始化消费者类

         // 开发者账号信息，使用云眸管理员账号登录open2.hik-cloud.com，进入秘钥管理页面获取
         String clientId = "";
         String clientSecret = "";
         MyPushMessageService myPushMessageService = new MyPushMessageService();
         Consumer consumer = new Consumer(clientId, clientSecret, myPushMessageService);
         // 启动消费线程
         consumer.run();

对接指南

简介#

权限开通#

核心概念#

核心流程#

消息通道数据模型#

消费消息流程#

常见问题#

消息通道拉取不到消息#

消息通道消费到的消息重复#

demo#

简介