第五次课：抄表业务分析

前言1：先把之前的netty代码优化一下

netty服务端修改,主要是那个编码器和解码器

 //给workerGroup 的 EventLoop 对应的管道设置处理器
        .childHandler(new ChannelInitializer<SocketChannel>() {
            //给pipeline 设置处理器
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                ChannelPipeline pipeline = socketChannel.pipeline();
                // ChannelOutboundHandler，依照逆序执行
                pipeline.addLast("encoder", new NettyMessageEncoder());
                // 属于ChannelInboundHandler，依照顺序执行
                pipeline.addLast("decoder", new NettyMessageDecoder());
                pipeline.addLast(new NettyServerHandler());
            }
        });

自定义编码器-负责写

public class NettyMessageEncoder extends MessageToByteEncoder<String> {

    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, String s, ByteBuf byteBuf) throws Exception {
        byteBuf.writeBytes(ElectricityMeterUtil.fromHexString(s));
    }
}

自定义解码器-负责读

@Slf4j
public class NettyMessageDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, List<Object> out) throws Exception {
        log.info("开始解码：");
        byte[] b = new byte[byteBuf.readableBytes()];
        //复制内容到字节数组b
        byteBuf.readBytes(b);
        //字节数组转字符串
        String str = new String(b);
        String content = ElectricityMeterUtil.getBufHexStr(b);

        out.add(content);
        log.info("解码结束!");
    }
}

前言2：发送指令时把操作类型存入redis

比如：你是进行的拉闸操作还是合闸操作还是查电表操作。

 //防止一台设备同时下发多条指令，如果缓存中有记录则表示上次的操作还未完成
String type = stringRedisTemplate.opsForValue().get(InstructConstant.handleInstructs + deviceCode);
if(StringUtils.isEmpty(type)){
    stringRedisTemplate.opsForValue().set(InstructConstant.handleInstructs+deviceCode, operationType,30, TimeUnit.SECONDS);
    //后面是发送指令的代码
    

内置软件烧写电表

初始化：为了方便测试，先用下面的软件初始化充点钱，充点电。两个重要的数据（余额和已用电量）

什么时候抄表

每隔两个小时就去检查下，用netty 去给网关发送查询电表数据的指令（相当于抄表员，要把每次抄表的数据保存到起来）。

/**
 * @author:xiaojie
 * @create: 2023-06-06 19:56
 * @Description: 抄表定时任务-每两小时抄一次
 */
@Slf4j
public class ElectricityMeterJob extends QuartzJobBean {
    @Resource
    IotDeviceService iotDeviceService;
    @Resource
    IotGatewayService gatewayService;
    @Resource
    SendInstructsService sendInstructsService;
    @Override
    protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
        log.info("查出所有的电表设备的余额");
        //查出所有的电表设备的余额
        List<IotDevice> list = iotDeviceService.list();
        for (IotDevice device : list) {
            //查出网关
            IotGateway gatewayVO = gatewayService.getById(device.getGatewayId());
            String address = ElectricityMeterUtil.deviceAddressConvert(device.getDeviceCode());
            String ip = "/"+gatewayVO.getGatewayIp()+":"+gatewayVO.getGatewayPort();
            String instruce = ElectricityMeterUtil.getCompleteInstruct(address, InstructConstant.ELECTRICITY_BALANCE);
            sendInstructsService.sendInstructs(instruce,ip,device.getDeviceCode(), InstructTypeConstant.E_BALANCE_SETUPA.getValue());

    }


    }
}

netty回调

在netty的serverHanlder里的channelRead里面。

客户端收到消息后的回调

 //接收到客户端给服务端发的消息
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //可以修改  电表设备的状态
        String returnStr=(String) msg;
        log.info("channelRead:read msg:"+msg);

        sendInstructsService.callbackInstructResultMsg(returnStr);

    }

 @Override
    public void callbackInstructResultMsg(String returnStr) {
        //回调处理
        if(StringUtils.isEmpty(returnStr) || !returnStr.startsWith("FE")){
            return;
        }
        //获取回调设备编号要有好几步骤
        String notFEStr = returnStr.substring(returnStr.indexOf("68")+2);
        String deviceType = notFEStr.startsWith("10")?"w":"e";
        //根据设备号找到数据，并更新数据
        if("e".equals(deviceType)) {//e就是电表  w就是水表

            String addressNo = notFEStr.substring(0,notFEStr.indexOf("68"));
            //获取设备号
            String deviceNo = ElectricityMeterUtil.addressConvertNo(addressNo,true);
            //获取操作类型
            String operationType = stringRedisTemplate.opsForValue().get(InstructConstant.handleInstructs+deviceNo);
            //删除redis中的数据
            stringRedisTemplate.delete(InstructConstant.handleInstructs+deviceNo);
            //电表操作
            handleOperation(deviceNo,operationType,notFEStr);

        }
    }
private void handleOperation(String deviceNo, String operationType, String notFEStr) {


        LambdaQueryWrapper<IotDevice> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(IotDevice::getDeviceCode,deviceNo);
        IotDevice device = deviceMapper.selectOne(wrapper);
        //合总电闸，拉闸或合闸后修改设备表状态
        if(InstructTypeConstant.E_MAIN_GATE_NO.getKey().equals(operationType)){
            device.setElectricityStatus(1);
            deviceMapper.update(device,wrapper);
        }
        //拉总电闸
        if(InstructTypeConstant.E_MAIN_GATE_OFF.getKey().equals(operationType)){
            device.setElectricityStatus(0);
            deviceMapper.update(device,wrapper);
        }

        //获取电表费控数据
        if(InstructTypeConstant.E_BALANCE_SETUPA.getKey().equals(operationType)){
            getElectricityBalanceInfo(notFEStr,deviceNo);
        }

    }

    private void getElectricityBalanceInfo(String notFEStr, String deviceNo) {

        String dataStr = notFEStr.substring(notFEStr.indexOf("81104A1C")+8,notFEStr.length()-4);
        //剩余金额
        String surplusAmount = dataStr.substring(12,20);
        surplusAmount = ElectricityMeterUtil.deviceAddressConvert(surplusAmount);
        surplusAmount = ElectricityMeterUtil.dataAreaSubtract33(surplusAmount);
        log.info("getElectricityBalanceInfo surplusAmount {} {}",surplusAmount);
        int amount = Integer.parseInt(surplusAmount.replaceAll(" ",""),16);
        log.info("getElectricityBalanceInfo amount {} {}",amount);
        String surplusAmountStr = String.valueOf(amount*0.01);
        log.info("getElectricityBalanceInfo surplusAmount {} {}",surplusAmount,surplusAmountStr);

        //查询设备信息
        LambdaQueryWrapper<IotDevice> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(IotDevice::getDeviceCode,deviceNo);
        IotDevice vo = deviceMapper.selectOne(wrapper);
        BigDecimal deviceBalance = vo.getDeviceBalance();//抄表前的剩余金额
        BigDecimal deviceSurplus = vo.getDeviceSurplus();//抄表前的剩余电量
        //计算使用费用
        BigDecimal useBalance = deviceBalance.subtract(new BigDecimal(surplusAmountStr));
        log.info("Electricity balance: {}",useBalance);
        //计算使用量
        BigDecimal useSurplus = useBalance.divide(new BigDecimal(1));
        //计算设备剩余费用
        BigDecimal remnantBalance = deviceBalance.subtract(useBalance);
        BigDecimal remnantQuantity = deviceSurplus.subtract(useSurplus);
        IotElectricityUse billElectricityUse = new IotElectricityUse();
        billElectricityUse.setDeviceId(vo.getId());
        billElectricityUse.setBeforUseBalance(deviceBalance);
        billElectricityUse.setBeforUseElectricity(deviceSurplus);
        //账户余额与电表余额对比
        if(deviceBalance.compareTo(new BigDecimal(surplusAmountStr))>0){
            //使用过电，修改余额和可使用量
            vo.setDeviceBalance(remnantBalance);
            vo.setDeviceSurplus(remnantQuantity);
            //更新设备金额
            deviceMapper.updateById(vo);
            //添加抄表记录
            billElectricityUse.setUseElectricity(useSurplus);
            billElectricityUse.setUseCost(useBalance);
            billElectricityUseDao.insert(billElectricityUse);
        }else{
            //没有使用过电 添加抄表记录
            billElectricityUse.setUseElectricity(BigDecimal.ZERO);
            billElectricityUse.setUseCost(BigDecimal.ZERO);
            billElectricityUseDao.insert(billElectricityUse);
        }

        //结算余额小于0，拉闸停电
        if(deviceBalance.compareTo(BigDecimal.ZERO)<0 || remnantBalance.compareTo(BigDecimal.ZERO)<0){
            log.info("Electricity Balance Not enough");
            IotGateway gatewayVO = gatewayMapper.selectById(vo.getGatewayId());
            String address = ElectricityMeterUtil.deviceAddressConvert(deviceNo);
            String newIp = "/"+gatewayVO.getGatewayIp()+":"+gatewayVO.getGatewayPort();
            String instructs = ElectricityMeterUtil.getCompleteInstruct(address,InstructConstant.ELECTRICITY_METER_PULL_INSTRUCT);
            //获取发送指令
            sendInstructs(instructs,newIp,deviceNo,InstructTypeConstant.E_MAIN_GATE_OFF.getKey());
            log.info("Electricity Balance sendInstructs success");
            //修改总闸状态
            vo.setElectricityStatus(0);
            deviceMapper.updateById(vo);
        }else{
            //余额大于0，合闸送电
            log.info("Electricity Balance enough");
            IotGateway gatewayVO = gatewayMapper.selectById(vo.getGatewayId());
            String address = ElectricityMeterUtil.deviceAddressConvert(deviceNo);
            String newIp = "/"+gatewayVO.getGatewayIp()+":"+gatewayVO.getGatewayPort();
            String instructs = ElectricityMeterUtil.getCompleteInstruct(address,InstructConstant.ELECTRICITY_METER_ON_INSTRUCT);
            //获取发送指令
            sendInstructs(instructs,newIp,deviceNo,InstructTypeConstant.E_MAIN_GATE_NO.getKey());
            log.info("Electricity Balance sendInstructs success");
            //修改总闸状态
            vo.setElectricityStatus(1);
            deviceMapper.updateById(vo);
        }
    }

    /**
     * 数据转换
     * @param aVoltage 元数据
     * @param precisionIndex 几位小数
     * @return
     */
    private static String dataChange(String aVoltage,int precisionIndex){
        String dataBCD = ElectricityMeterUtil.hexAddcharacter(aVoltage," ");
        String subtract33Data =  ElectricityMeterUtil.dataAreaSubtract33(dataBCD);
        log.info("dataChange {}",subtract33Data);
        String changeAfterStr = ElectricityMeterUtil.addressConvertNo(subtract33Data,false);
        Long changeAfterLong = Long.valueOf(changeAfterStr);
        if(precisionIndex==1){
            return String.valueOf(changeAfterLong*0.1);
        }
        if(precisionIndex==2){
            return String.valueOf(changeAfterLong*0.01);
        }
        if(precisionIndex==3){
            return String.valueOf(changeAfterLong*0.001);
        }
        if(precisionIndex==4){
            return String.valueOf(changeAfterLong*0.0001);
        }
        return String.valueOf(changeAfterLong);
    }

回调业务难点

补充 ：非spring管理的类怎么拿bean

 private static final StringRedisTemplate stringRedisTemplate = SpringUtils.getBean("stringRedisTemplate");
 

/**
 * ClassName: SpringUtils
 * Description: spring工具类 方便在非spring管理环境中获取bean
 */
@Component
public class SpringUtils implements BeanFactoryPostProcessor {
    /**
     * Spring应用上下文环境
     */
    private static ConfigurableListableBeanFactory beanFactory;

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
        SpringUtils.beanFactory = beanFactory;
    }

    /**
     * 获取对象
     *
     * @param name
     * @return Object 一个以所给名字注册的bean的实例
     * @throws BeansException
     */
    @SuppressWarnings("unchecked")
    public static <T> T getBean(String name) throws BeansException {
        return (T) beanFactory.getBean(name);
    }

    /**
     * 获取类型为requiredType的对象
     *
     * @param clz
     * @return
     * @throws BeansException
     */
    public static <T> T getBean(Class<T> clz) throws BeansException {
        T result = (T) beanFactory.getBean(clz);
        return result;
    }

    /**
     * 如果BeanFactory包含一个与所给名称匹配的bean定义，则返回true
     *
     * @param name
     * @return boolean
     */
    public static boolean containsBean(String name) {
        return beanFactory.containsBean(name);
    }

    /**
     * 判断以给定名字注册的bean定义是一个singleton还是一个prototype。 如果与给定名字相应的bean定义没有被找到，将会抛出一个异常（NoSuchBeanDefinitionException）
     *
     * @param name
     * @return boolean
     * @throws NoSuchBeanDefinitionException
     */
    public static boolean isSingleton(String name) throws NoSuchBeanDefinitionException {
        return beanFactory.isSingleton(name);
    }

    /**
     * @param name
     * @return Class 注册对象的类型
     * @throws NoSuchBeanDefinitionException
     */
    public static Class<?> getType(String name) throws NoSuchBeanDefinitionException {
        return beanFactory.getType(name);
    }

    /**
     * 如果给定的bean名字在bean定义中有别名，则返回这些别名
     *
     * @param name
     * @return
     * @throws NoSuchBeanDefinitionException
     */
    public static String[] getAliases(String name) throws NoSuchBeanDefinitionException {
        return beanFactory.getAliases(name);
    }

    /**
     * 获取aop代理对象
     *
     * @param invoker
     * @return
     */
    @SuppressWarnings("unchecked")
    public static <T> T getAopProxy(T invoker) {
        return (T) AopContext.currentProxy();
    }
}

老师demo

https://gitee.com/jf3q/school-iot.git