云苍穹集成云-kafka（消费者）

1.建立消息队列服务器(集成管理-消息集成-队列服务)

---

2.订阅主题配置(消费订阅)

2.1: 新建集成对象(集成管理-集成元数据-集成对象)

新闻的订阅和消费需要依赖和集成对象
集成对象至少建立一个集成属性(候选键必须在后面的集成方案中指定)

2.2: 新订阅主题(集成管理-消息集成-订阅主题)

新订阅主题完成后，可以在消息接收(集成管理-消息集成-消息接收)中查看消息生产者推送的消息

2.3: 消息的生产者向我们的队列服务发送消息(供应商需要做)

消费者提供kafka服务器IP地址 (队列服务中的链接地址)；

---

3、消费消息

3.1.新数据集成方案(集成管理-数据集成-数据集成方案)

字段映射维护： 至少新增一个属性作为候选人的关键
转换脚本： 脚本内容：tar.srcData = src，

数据处理维护：msvc://zb_pc.zb_pcem.WlwltServiceImpl.getWLWData
msvc:// api 服务类路径 实现方法名

3.二、新启动方案

3.3启动方案

1:方案启用后，可在执行结果和执行日志中查看

---

4.调试及相关代码说明

4.1 新建 web-api类(集成方案3.步骤1-数据处理类对应类)

package zb.pc.mservice.api.pecm;  import java.util.Map;  /**  * Model      :物联网平台-料塔地磅-消息同步接口  * Description:  * Remark     :  *  * @author tanrongtao  * @date 2022/5/5  */ public interface WlwltService {      Map getWLWData(Map map); }

4.2 实现类

package zb.pc.mservice.pecm;  import com.alibaba.fastjson.JSONObject; import kd.bos.logging.Log; import kd.bos.logging.LogFactory; import zb.pc.business.pcem.PcemBussiness; import zb.pc.mservice.api.pecm.WlwltService;  import java.util.Map;  /**  * Model      :物联网平台-料塔地磅-消息同步接口  * Description:  * Remark     :  *  * @author tanrongtao  * @date 2022/5/5  */ public class WlwltServiceImpl implements WlwltService {     private static final Log logger = LogFactory.getLog(WlwltServiceImpl.class);      @Override     public Map getWLWData(Map map) {         logger.info("物联网-料塔地磅数据同步入参： "   JSONObject.toJSONString(map));         try {             Map data = (Map) map.get("data");             Map srcData = (Map) data.get("srcData");             PcemBussiness.synWlwData(srcData);         } catch (Exception e) {             map.put("custFail", e.getMessage());             logger.error("物联网-料塔地磅数据同步异常"   e.getMessage());         }         return map;     } } 

4.3 调试：