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.2：新建启动方案
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<String,Object> getWLWData(Map<String,Object> 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<String, Object> getWLWData(Map<String, Object> map) {
logger.info("物联网-料塔地磅料塔地磅数据同步入参： " + JSONObject.toJSONString(map));
try {
Map<String, Object> data = (Map<String, Object>) map.get("data");
Map<String, Object> srcData = (Map<String, Object>) data.get("srcData");
PcemBussiness.synWlwData(srcData);
} catch (Exception e) {
map.put("custFail", e.getMessage());
logger.error("物联网-料塔地磅数据同步异常" + e.getMessage());
}
return map;
}
}
?4.3 调试：
?
