2.订阅主题配置(消费订阅)
新闻的订阅和消费需要依赖和集成对象集成对象至少建立一个集成属性(候选键必须在后面的集成方案中指定)
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<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 调试：