读写Excel基本代码

直接复制不一定能用

实体类

@ExcelIgnore 在导出操作中不会被导出

@ExcelProperty 在导入过程中 可以根据导入模板自动匹配字段， 在导出过程中可用于设置导出的标题名字

@Getter
@Setter
public class Material{
    @ExcelIgnore
    private Long id;

    /** 所属部门 */
    @ExcelProperty(value = {"一级合并标题","所属部门"})
    private String department;

    /** 所属部室/项目部/站区 */
    @ExcelProperty(value = {"一级合并标题","所属部室/项目部/站区"})
    private String area;
}

监听类

监听类控制导入数据的规则限制，监听类继承
AnalysisEventListener<Object>
类实现接口。

监听中无法调用Spring接口，使用反序列化注入Bean工厂中使用

public class MaterialListener extends AnalysisEventListener<Material> {
    // 反序列化注入bean工厂
    ISysDictDataService dictDataService = SpringUtils.getBean(ISysDictDataService.class);
    //可以通过实例获取该值
    private List<Material> datas = new ArrayList<>();
    //错误信息
    private List<String> errorInfo = new ArrayList<>();
    @Override
    public void invoke(Material material, AnalysisContext analysisContext) {
        doSomething(material,analysisContext);//根据自己业务做处理

        datas.add(material);//数据存储到list，供批量处理，或后续自己业务逻辑处理。
    }
    private void doSomething(Material object, AnalysisContext analysisContext) {
        //获取当前行数
        Integer row = analysisContext.readRowHolder().getRowIndex();
        SysDictData sysDictData=new SysDictData();
        if (ObjectUtils.isBlank(object.getType())){
            errorInfo.add(" 第" + (row+1) + "行：物资种类不能为空，请重新填写后再导入!<br>");
        }else {
            sysDictData.setDictType("material_type");
            sysDictData.setDictLabel(object.getType());
            List<SysDictData> list3= dictDataService.selectDictDataList(sysDictData);
            if (ObjectUtils.isBlank(list3)){
                errorInfo.add(" 第" + (row+1) + "行：物资种类:"+object.getType()+"未在数据字典中找到，请先填写后再导入!<br>");
            }
        }
    }
    //返回结果集对象
    public List<Material> getDatas() {
        return datas;
    }

    public void setDatas(List<Material> datas) {
        this.datas = datas;
    }

    //返回错误消息
    public List<String> getErrorInfo() {
        return errorInfo;
    }

    public void setErrorInfo(List<String> errorInfo) {
        this.errorInfo = errorInfo;
    }
    @Override
    public void doAfterAllAnalysed(AnalysisContext analysisContext) {

    }
}

控制器

导入数据方法

在
materialService.saveList(materialList)
中进行处理导入获取过来的数据，用于存取业务数据

    @PostMapping("/importData")
    @ResponseBody
    public AjaxResult  importData(@RequestParam MultipartFile file) throws Exception
    {
        MaterialListener excelListener = new MaterialListener();
        // 这里 需要指定读用哪个class去读，然后读取第一个sheet 文件流会自动关闭
        //EasyExcel.read(哪个文件).sheet(那张sheet表).head(表头什么样子).headRowNumber(表头占几行).registerReadListener(处理数据的监听器类).doRead()
        EasyExcel.read(file.getInputStream(), Material.class, excelListener).sheet().head(Material.class).headRowNumber(2).doRead();
        if(ObjectUtils.isBlank(excelListener.getErrorInfo())){
            if(ObjectUtils.isNotBlank(excelListener.getDatas())){
                List<Material> materialList = excelListener.getDatas();
                materialService.saveList(materialList);
            }
            return AjaxResult.success(1);
        }else{
            return AjaxResult.error("操作失败",excelListener.getErrorInfo());
        }
    }

导出数据方法

单sheet导出在注释的方法出，多sheet导出如下所示

 @GetMapping("/exportData")
    public void exportData(Material material, HttpServletResponse response){
        //换方法连表查询
//        List<Material> list = materialService.selectMaterialListByDict(material);
        List<Material> materialList=new ArrayList<>();
        List<Material> list = materialService.selectMaterialList(material);
        String[] ids=new String[list.size()];
        for (Material p:list) {
		//用于修改导出的数据
            String list7= dictDataService.selectDictLabel("material_type",p.getType());
            if (ObjectUtils.isNotBlank(list7)){
                p.setType(list7);
            }
            materialList.add(p);
        }
        List<MaterialSolid> materialSolidList=new ArrayList<>();
        List<MaterialSolid> materialSolidListS = materialSolidService.selectMaterialSolidListByIds(ids);
        for (MaterialSolid solid:materialSolidListS) {
            String list7= dictDataService.selectDictLabel("material_type",solid.getType());
            if (ObjectUtils.isNotBlank(list7)){
                solid.setType(list7);
            }
            materialSolidList.add(solid);
        }
        try {
            response.setContentType("application/vnd.ms-excel");
            response.setCharacterEncoding("utf-8");
            // 这里URLEncoder.encode可以防止中文乱码 当然和EASYEXCEL没有关系
            String fileName = URLEncoder.encode("物料信息台账", "UTF-8");
            response.setHeader("Content-disposition", "attachment;filename="+ fileName + ".xlsx");
            //调用方法进行写的操作
//            EasyExcel.write(response.getOutputStream(),Material.class).sheet("物料信息台账").doWrite(materialList);
            //多sheet页导出
            ExcelWriter excelWriter = EasyExcel.write(response.getOutputStream()).build();
            WriteSheet writeSheet1 = EasyExcel.writerSheet(0, "物料信息总账").head(Material.class).build();
            WriteSheet writeSheet2 = EasyExcel.writerSheet(1, "物料信息详情").head(MaterialSolid.class).build();
            excelWriter.write(materialList, writeSheet1);
            excelWriter.write(materialSolidList, writeSheet2);
            excelWriter.finish();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

需要完成功能

借助redis Stream 数据结构实现消息队列，异步完成订单创建，其中涉及到了缓存（击穿，穿透，雪崩），锁（Redisson），并发处理，异步处理，Lua脚本

IDE:IDEA 2022

1、读取库存数据 【Lua】

2、判断库存 【Lua】

3、扣减库存 【Lua】

4、创建队列和组 【Java】

5、发送队列消息 【Lua】

6、消息读取并处理 【Java】

在完成功能之前 ，需要了解一下redis 中有关stream 数据结构相关的命令

XACK：确认消息已经处理，redis 会在PEL(pending entries List )中移除一个或多个消息。一般情况下 一个消息被 XREADGROUP 或 XCLAIM之后会被写入PEL。

XADD: 把消息（Entry  ,key-value)追加到队列，默认如果队列不存在会创建，除非使用 NOMKSTREAM ,之后 可以通过XREAD ,XREANGE 等命令读取或通过XDEL,XTRIM移除消息

XCLAIM,XAUTOCLAIM 改变PEL 中的消息的所有者

XDEL：在队列中移除一个或多个消息（entry)

XGROUP CREATE :在指定的队列中创建一个消费者组，队列key有且仅能有一个，否则重复会提示：-BUSYGROUP 不存在会提示：ERR no such key ，可以通过选项 MKSTREAM 在不存在时创建

XGROUP CREATECONSUMER:在一个给定的队列和消费者组中创建一个消费者，不能重复。此外在任何使用到消费者的命令中，如果不存在则自动创建。如：XREADGROUP

XGROUP DELCONSUMER:在一个给定的队列和消费者组中移除一个消费者  在执行此命令之前 相关的PEL 要先执行 XCLAIM 或 XACK进行处理，否则将变得 unclaimable.

XGROUP DESTROY:在给定的队列中删除一个消费者组，相关的consumers 和 PEL 都会被 删除，所以执行之前要慎重

XGROUP SETID :重新设置指定的消费者组的最后一个处理的消息的ID，通常这个ID初始值 是XGROUP CREATE时指定的，每次执行XREDGROUP 也会修改（更新）这个ID，

例如：想让组内的消费者重新处理队列中的所有的entrys时，可以 XGROUP SETID streamkey groupkey 0，在redis 7.0增加了参数 ENTRIESREAD n ，n 为已读数量 ，

此时，xinfo groups streamKey 可以看到组信息如下：

127.0.0.1:6388> xread count 20 streams s1 0
1) 1) "s1"
   2) 1) 1) "1678852071712-0"
         2) 1) "key1"
            2) "vlaue1"
      2) 1) "1678852073882-0"
         2) 1) "key1"
            2) "vlaue1"
      3) 1) "1678852080406-0"
         2) 1) "key2"
            2) "vlaue2"
      4) 1) "1678852588261-0"
         2) 1) "key-1"
            2) "vlaue2"
      5) 1) "1678852591957-0"
         2) 1) "key-2"
            2) "vlaue2"
      6) 1) "1678852595467-0"
         2) 1) "key-3"
            2) "vlaue2"
      7) 1) "1678852599576-0"
         2) 1) "key-4"
            2) "vlaue2"
      8) 1) "1678852616566-0"
         2) 1) "key-4"
            2) "vlaue2"
      9) 1) "1678852946989-0"
         2) 1) "key-5"
            2) "vlaue2"
127.0.0.1:6388> xinfo groupss11)  1) "name"
    2) "g1"
    3) "consumers"
    4) (integer) 1
    5) "pending"
    6) (integer) 5
    7) "last-delivered-id"
    8) "1678852080406-0"
    9) "entries-read"
   10) (integer) 9
   11) "lag"
   12) (integer) 0
127.0.0.1:6388> xreadgroup group g1 lihui  count 1 streams s1 >
1) 1) "s1"
   2) 1) 1) "1678852588261-0"
         2) 1) "key-1"
            2) "vlaue2"
127.0.0.1:6388> xinfo groupss11)  1) "name"
    2) "g1"
    3) "consumers"
    4) (integer) 1
    5) "pending"
    6) (integer) 5
    7) "last-delivered-id"
    8) "1678852588261-0"
    9) "entries-read"
   10) (integer) 10
   11) "lag"
   12) (integer) -1
127.0.0.1:6388>xlen s1

(integer)9
127.0.0.1:6388> xgroup setid s1 g1 1678852073882-0  ENTRIESREAD 2OK127.0.0.1:6388> xinfo groupss11)  1) "name"
    2) "g1"
    3) "consumers"
    4) (integer) 1
    5) "pending"
    6) (integer) 5
    7) "last-delivered-id"
    8) "1678852073882-0"
    9) "entries-read"
   10) (integer) 2
   11) "lag"
   12) (integer) 7

View Code

XINFO [SRTREAM,GROUPS,CONSUMERS] :获取队列，组，消费者的信息

XLEN:返回 stream中entrys的数量

XPENDING :查询通过消费者组读取但未被确认的entrys

XRANGE :返回队列中的某个区间内的entrys

XREVRANGE: 顺序说XRANGE相反的读取队列中的entry

XREAD:从队列 内读取一个或多个entry ,支持block

XREADGROUP:在XREAD的基础上在队列与消费者之间增加了组的概念

XSETID:是一个redis 内部命令，用于记录master 与replicate之间的数据 同步的最后一个ID 的记录

好了，下面是是代码

Lua 脚本 完成 相关功能

--[[判断优惠券是否充足  优惠券  id ARGV[2] ,key KEYS[1]

 判断当前用户是否已经下单 用户id ARGV[1]

 如果库存充足，且用户没有下单 则 1、扣库存 2、保存用户下单信息  订单key KEYS[2] 订单ID ARGV[3]

stream 队列 key  KEYS[3]

consumer group key   KEYS[4]

  利用redis 中set 数据 类型的不重复性，进行重复下单信息的记录

--]]

local sec_kc=redis.call('get',KEYS[1])--if(not sec_kc) then--redis.call('set',KEYS[1],2000)--sec_kc=2000--end
if(not sec_kc or  sec_kc=='{}') then
    --库存 不足
    return 3
end
if(  tonumber(sec_kc)<=0) then
    --库存 不足
    return 1
end
--完成一人一单重复检测
if(redis.call('SISMEMBER',KEYS[2],ARGV[1])==1) then
    --订单重复
    return 2
end
--扣减库存
redis.call('incrby',KEYS[1],-1)--记录订单与用户之间关系 set 数据
redis.call('sadd',KEYS[2],ARGV[1])--获取stream队列中entry数量

--[=[local streamcount=redis.call('xlen',KEYS[3])

--此功能转到java 中完成 CreateStreamAndGroup

if(streamcount==0) then--没有stream 需要创建

    -- XGROUP CREATE stream.order group1 0 mkstream

    redis.call('XGROUP','CREATE',KEYS[3],KEYS[4],'0','mkstream')



end

--]=]
--判断队列是否存在
local streamExists=redis.call('exists',KEYS[3])if(streamExists==0) then
     return 4--队列不存在
end
--创建消息entry--XADD S1 * KEY12 VALUE12
redis.call('XADD',KEYS[3],'*','userId',ARGV[1],'voucherId',ARGV[2],'id',ARGV[3])--key 的取值与对象 VoucherOrder 中变量属性对应，分别是，userId,voucherId,id(订单id)--满足下单条件
return 0

因为在redis 的stream 命令中 没有判断 consumer group 是否存在，所以需要变通的方法，解决这个问题，因为如果组不存的情况下，进行xadd 会报错。

创建队列和消费者组

 /*** 查询队列信息，如果不存在会出现异常，在异常中创建队列

     *@paramstreamKey

     *@paramgroupKey*/
    private voidCreateStreamAndGroup(String streamKey,String groupKey)

    {try{//默认队列和组不存 创建队列
            stringRedisTemplate.opsForStream().createGroup(streamKey, ReadOffset.from("0"), groupKey);

            System.out.println(streamKey+ "队列" + streamKey + "和组" + groupKey + "创建成功。");

        }catch( Exception ex)

        {

            String errx=ex.getMessage().toString();if(errx.indexOf("BUSYGROUP Consumer Group name already exists")>0)//队列与组都已经存在
{//System.out.println(streamKey + "队列" + streamKey + "和组" + groupKey + "都已经存在");
}else{

                log.debug(ex.getMessage());

            }

        }





    }

判断队列是否存在

 /*** 判断consumer group 是否存在,如果队列不存在直接返回 false,

     * 如果队列存在，不论组是否存，都直接创建。*/
    private booleanConsumerGroupExists(String streamKey,String groupKey)

    {boolean isok=false;try{//默认队列和组不存 创建队列
            Collection<String> streamCollect=new ArrayList<>();

            streamCollect.add(streamKey);

            Long c_stream=stringRedisTemplate.countExistingKeys(streamCollect);if(c_stream==0)

            {

                isok= false;//队列不存在
                returnisok;

            }//队列存在 ，那么直接创建消费者组，
            stringRedisTemplate.opsForStream().createGroup(streamKey, ReadOffset.from("0"), groupKey);

            System.out.println(streamKey+ "队列" + streamKey + "存在，组" + groupKey + "创建成功。");

            isok=true;//组创建成功
}catch( Exception ex)

        {

            String errx=ex.getMessage().toString();if(errx.indexOf("BUSYGROUP Consumer Group name already exists")>0)//队列与组都已经存在
{

                isok=true;//组创建成功//System.out.println(streamKey + "队列" + streamKey + "和组" + groupKey + "都已经存在");
}else{

                isok=false;//未知异常
log.debug(ex.getMessage());

            }

        }finally{returnisok;

        }





    }

订单创建入口函数（放在controller 或 seervice impl） 中

publicResult secKillVoucher(Long voucherId) {long userID =UserHolder.getUser().getId();



        String vouchStockKey= RedisConstants.SECKILL_VOUCHERSTOCK_CACHE_KEY +voucherId;

        String userid_s=String.valueOf(userID);

        String vouchOrderKey= RedisConstants.SECKILL_ORDER_CACHE_KEY +voucherId;



        String streamKey=RedisConstants.REDIS_STREAM_QUEUE_ORDER ;

        String groupKey=RedisConstants.REDIS_STREAM_GROUP_ORDER;



        List<String> keys = new ArrayList<>();

        keys.add(vouchStockKey);//库存key KEYS[1]
        keys.add(vouchOrderKey);//订单key KEYS[2]
        keys.add(streamKey);//队列key  KEYS[3]
        keys.add(groupKey);//消费组key  KEYS[4]//check sec_voucher_stock//1 查询优惠券
        int count =seckillVoucherService.getRedisStock(voucherId);if (count <= 0) {return Result.fail("优惠券库存不存在 count：" +count);



        }long orderid =redisIdWorker.nextId(RedisConstants.ID_ORDER);

        CreateStreamAndGroup(streamKey,groupKey);//订单ID ARGV[3]//优惠券  id ARGV[2]//用户id ARGV[1]//执行lua 完成优惠券库存，一人一单，检查 并创建用户与优惠券ID之间的关联//脚本完成 创建消息队列 （执行此脚本之前确保队列 已经创建
        Long sekResult =stringRedisTemplate.execute(SECKILL_SCRIPT,

                keys

                , userid_s,String.valueOf(voucherId),String.valueOf(orderid)

        );int ri =sekResult.intValue();

        System.out.println("lua 脚本执行返回值 ："+ri);if(ri!=0)

        {
            String errtip="优惠券库存不足";switch(ri)

           {case 2:

                   errtip="订单重复一人一单";break;case 3:

                   errtip="库存数据不存在";break;case 4:

                   errtip="消息队列不存在";break;case 1:

                   errtip="优惠券库存不足";break;default:

                   errtip="未知错误";



           }returnResult.fail(errtip);

        }

　　　　/**注意：下面的代理对象要有

 　　　　*/proxy=(IVoucherOrderService)AopContext.currentProxy();returnResult.ok(orderid);



    }

创建线程获取消息队列并处理

 private static final ExecutorService SECKILL_ORDER_EXECUTOR=Executors.newSingleThreadExecutor();

    @PostConstructprivate voidinit(){//本地阻塞队列的方式//SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());//读取redis消息队列并处理
        SECKILL_ORDER_EXECUTOR.submit(newVoucherOrderHandlerRedisQueue());

    }//从redis 消息队列中获取消息
    private class VoucherOrderHandlerRedisQueue implementsRunnable{



        @Overridepublic voidrun() {while(true){//get orderinfo from blockingqueen
                try{

                    String streamKey=RedisConstants.REDIS_STREAM_QUEUE_ORDER ;

                    String groupKey=RedisConstants.REDIS_STREAM_GROUP_ORDER;

                    String cs=RedisConstants.REDIS_STREAM_CONSUMER_ORDER;//判断组是否存在
                    if (!ConsumerGroupExists(streamKey,groupKey)) {//log.error("异步线程读取redis stream 失败：队列 未创建："+streamKey);
                        Thread.sleep(3000);continue;

                    }//从消息队列中获取消息
                    List<MapRecord<String, Object, Object>> queueList =stringRedisTemplate.opsForStream()

                            .read(Consumer.from( groupKey,cs),

                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2))

                            , StreamOffset.create(streamKey,ReadOffset.lastConsumed())

                    );if(queueList==null ||queueList.isEmpty() )

                    {//未获取到消息
                        continue;//continue do next
}//处理消息
                    MapRecord<String, Object, Object> record = queueList.get(0);

                    Map<Object, Object> vq =record.getValue();

                    VoucherOrder voucherOrder= BeanUtil.fillBeanWithMap(vq,new VoucherOrder(),false);//成功获取 创建订单到数据库
handlerVoucherOrder(voucherOrder);//确认消息 在PEL中移除
stringRedisTemplate.opsForStream().acknowledge(streamKey,groupKey,record.getId());



                }catch(Exception e) {



                    HandlePendingList();

                    log.error(e.getMessage().toString());//throw new RuntimeException(e);
}





            }

        }

异常处理 在redis stream pending List 中获取 队列

 private voidHandlePendingList() {

            String streamKey= RedisConstants.REDIS_STREAM_QUEUE_ORDER ;//stream
            String groupKey = RedisConstants.REDIS_STREAM_GROUP_ORDER;//consumer group
            String cs=RedisConstants.REDIS_STREAM_CONSUMER_ORDER;//consumer
            while (true)try{

                    {//read from  pel
                    List<MapRecord<String, Object, Object>> queueList =stringRedisTemplate.opsForStream()

                            .read(Consumer.from ( groupKey,cs),

                            StreamReadOptions.empty().count(1)

                            , StreamOffset.create(streamKey, ReadOffset.from("0"))

                    );if(queueList==null ||queueList.isEmpty())

                    {//未获取到消息
                        break;//continue do next normal
}//处理消息
                    MapRecord<String, Object, Object> record = queueList.get(0);

                    Map<Object, Object> vq =record.getValue();

                    VoucherOrder voucherOrder= BeanUtil.fillBeanWithMap(vq,new VoucherOrder(),false);//成功获取
handlerVoucherOrder(voucherOrder);//确认消息
stringRedisTemplate.opsForStream().acknowledge(streamKey,groupKey,record.getId());



                    }

                }catch(Exception e) {

                    log.debug("消息队列--peding List 处理异常");try{

                        Thread.sleep(50);

                    }catch(InterruptedException ex) {throw newRuntimeException(ex);

                    }



                }

        }

    }

订单处理

 privateIVoucherOrderService proxy;private voidhandlerVoucherOrder(VoucherOrder voucherOrder) {if(voucherOrder==null || voucherOrder.getUserId()==null)

        {

            log.debug("对象为空 或 属性用户ID 为空。");

        }//user ID
        long userID=voucherOrder.getUserId();



        RLock lock=redisson.getLock(RedisConstants.LOCK_VOUCHERORDER_KEY+StrUtil.toString(userID));boolean islock = lock.tryLock();//active watch dog

        if(!islock)

        {

           log.error("锁创建失败");

        }try{//IVoucherOrderService proxy=(IVoucherOrderService)AopContext.currentProxy();//获取spring 对当前对象的代理
proxy.createVoucherOrderByObj(voucherOrder);



        }catch(IllegalStateException e) {throw newRuntimeException(e);

        }finally{//lock.unLock();//my define simple redis lock
            lock.unlock();//redisson release lock
}



    }

实现类中的方法

 @Transactional//因为方法中 订单的操作 和 库存扣减，所以增加事务支持 为防止数据 不同步
    public voidcreateVoucherOrderByObj(VoucherOrder voucher) {//user ID
        if(voucher==null || voucher.getUserId()==null)

        {

            log.debug("对象为空 或 属性用户ID 为空。");return;

        }long userID=voucher.getUserId();int count=query().eq("user_id",userID).eq("voucher_id", voucher.getVoucherId()).count();if(count>0)

        {

           log.error("优惠券仅限每人一个 count："+count);return;

        }//5扣减库存
        boolean success=seckillVoucherService.update()

                .setSql("stock=stock-1")

                .eq("voucher_id", voucher.getVoucherId()).gt("stock",0)

                .update();if(!success)

        {

            log.error("扣减券库存失败Obj: voucher.getVoucherId():"+voucher.getVoucherId());return;

        }

        System.out.println("voucher saved");boolean ds =save(voucher);





    }

The art of being wise is the art of knowing what to overlook .
智慧的艺术是知道该忽视什么。

——威廉·詹姆斯（William James）

1 导引

1.1 跨域推荐简介

推荐系统中常常面临用户
冷启动问题
^[1]
，也即新注册的用户没有足够的交互记录，导致推荐模型不能学习到有效的表征。为了解决用户冷启动问题，近年来
跨域推荐（CDR）
得到了许多关注
^[2]
。一般来讲，跨域推荐旨在利用从其它相关源域收集的用户-物品交互信息以提升目标域的推荐质量。许多跨域推荐的工作会假设大量的用户在两个域都出现过（即
重叠用户, overlapping users
）以搭建起源域和目标域之间的桥梁。只在源域中存在的用户（即
非重叠用户, non overlapping users
）可以被视为目标域的冷启动用户。

1.2 嵌入和映射的思路

为了解决冷启动用户问题，传统的跨域推荐方法常常基于
嵌入和映射（Embedding and Mapping，EMCDR）
的思路，也即学习一个映射函数将预训练的用户表征（embeddings）从源域迁移到目标域。如下图所示：

如上图所示，EMCDR首先用基于协同过滤的模型（CF-based model）来为每个领域生成用户/物品表征，之后训练一个映射函数来将源域和目标域的重叠用户表征。然后，再给定源域的非重叠冷启动用户表征，就能够根据训练好的映射函数来预测目标域的用户表征了，之后再用于目标域的物品推荐。

然而，正如我们上面所说的，这种方法在进行对齐操作之前，各领域需要先通过预训练以独立地得到用户/物品的embeddings。因此，
有偏的（biased）
预训练表征将无可避免地包含
领域特有的（domain-specific）
信息，从而会导致对跨领域迁移信息产生负面影响。

事实上，跨域推荐的关键问题就在于：
究竟需要在不同的域之间共享什么信息？也即如何让表征能够编码到领域间共享(domain-shared)的信息？

1.3 联合训练的思路

这种思路相比于EMCDR方法的优点在于，我们能够联合（jointly）学习跨领域的embeddings，从而能够进一步地关注于领域共享信息并限制领域特有的信息。

在具体的手段层面，这种方法该类方法的大多数工作首先采用两个基础的编码器来对每个领域的交互记录建模，之后再引入不同的迁移层来对称地融合不同编码器学得的表征。比如，CoNet
^[3]
利用MLP做为每个领域的基础编码器，并设计了交叉连接（cross-connections）网络来迁移信息。DDTCDR
^[4]
进一步扩展了ConNet:学习了一个潜在的正交投影函数来迁移跨领域用户的相似度。PPGN
^[5]
使用堆叠的（stacking）GCN来直接聚合来自各领域的表征信息以学得用户/物品表征。BiTGCF
^[6]
利用LightGCN
^[7]
做为编码器来聚合每个领域的交互信息，并进一步引入特征迁移层来增强两个基础的图编码器。CDRIB
^[8]
则采用信息瓶颈的视角来获得领域间共享的信息（不过该方法关注的是为目标域中的不重叠（冷启动）用户做推荐，与前面的方法又有所区别）。

1.4 解耦表征的思路

尽管以上的方法在一定程度上有效，但它们基本上仍然忽略了对领域共享信息和领域特有信息的解耦（CDRIB除外），而这大大限制了模型迁移的效率。

一个显著的例子如上图所示。对于Film和Book这两个领域，领域间共享的信息，比如“Story Topic”和“Category”能够为每个领域都提供有价值的信息。但领域特有的信息，比如Book领域的“Writing Style”可能会提供对于在“Film”领域做推荐无用的信息甚至会导致CDR领域的
负迁移
现象
^[9]
。不幸的是，现有的CDR方法忽视了此问题并直接聚合领域间共享和领域特有的信息。这样的结果就是，学得的用户表征将不同领域的偏好纠缠（entangle）在一起，而这会导致获得次优（sub-optial）的推荐结果。

解决该问题的手段是解耦领域间共享的领域特有的表征，其代表为DisenCDR模型
^[10]
。

如上图所示，DisenCDR模型将领域共享的和领域特有的表征进行解耦，以达到跨领域知识迁移的目的。

2 论文阅读

2.1 ICDE 2022《Cross-Domain Recommendation to Cold-Start Users via Variational Information Bottleneck》
^[8]

本方法属于采用联合训练的跨域推荐方法。本其关注的场景为
当源域和目标域间的用户部分重叠时，为目标域中的不重叠（冷启动）用户做推荐
。该方法所要解决的问题在于，
究竟有哪些信息需要在领域间进行共享
？

为了解决该问题，
本文利用了信息瓶颈（information bottleneck）原理并提出了一个新的方法（CDIRB模型）来使表征编码到领域间共享的信息（domain shared information），从而用于各领域的下游推荐
。为了得到无偏的表征，作者设计了两种正则项，其中
信息瓶颈正则项
来同时建模跨域/域间的用户-物品交互，这样相比EMCDR方法，就能够同时考虑所有域的交互信息从而达到去偏的目的；而
对比信息正则项
则负责捕捉跨域的用户-用户之间的关系（对齐不同域之间的重叠用户表征）。

设有领域
\(X\)
和
\(Y\)
，设
\(D^X=(\mathcal{U}^X, \mathcal{V}^X,\mathcal{E}^X)\)
，
\(D^Y=(\mathcal{U}^X, \mathcal{V}^X,\mathcal{E}^X)\)
表示领域的数据，这里
\(\mathcal{U}\)
、
\(\mathcal{V}\)
、
\(\mathcal{E}\)
分别表示每个领域用户、物品和边的集合。特别地，用户集合
\(\mathcal{U}^X\)
和
\(\mathcal{U}^Y\)
包含重叠的用户子集
\(\mathcal{U}^o = \mathcal{U}^X \cap \mathcal{U}^Y\)
。接着，用户集合可以被形式化为
\(\mathcal{U}^X = \{\mathcal{U}^x, \mathcal{U}^o\}\)
和
\(\mathcal{U}^Y = \{\mathcal{U}^y, \mathcal{U}^o\}\)
，这里
\(\mathcal{U}^x\)
和
\(\mathcal{U}^y\)
为在每个领域中不重叠的用户集合。设
\(\boldsymbol{A}^X=\{0,1\}^{\left|\mathcal{U}^X\right| \times\left|\mathcal{V}^X\right|}\)
和
\(A^Y=\{0,1\}^{\left|\mathcal{U}^Y\right| \times\left|\mathcal{V}^Y\right|}\)
为存储用户-物品交互信息的两个二值矩阵。这样，本文的任务可形式化地描述为：给定来自源域
\(X\)
的非重叠的（冷启动）用户
\(u_i\in \mathcal{U}^x\)
，我们想要为其推荐来自目标域
\(Y\)
的物品
\(v_j \in \mathcal{V}^Y\)
（或为来自
\(\mathcal{U}^y\)
的用户推荐来自
\(\mathcal{V}^X\)
的物品）。

接下来作者借鉴了论文
^[11][12]
提出的信息瓶颈理论，该理论旨在学习有效表征，这种有效表征能够在简洁性和广泛的预测能力之间做权衡（trade-off）
^[13]
。形式化地，标准信息瓶颈有如下所示的目标函数：

该目标函数可以被解释为两部分：（1）最小化
\(I(Z; X)\)
旨在惩罚
\(Z\)
和
\(X\)
之间的互信息，也即使得
\(Z\)
尽量“忘掉”
\(X\)
的信息。(2) 最大化
\(I(Z; Y)\)
则鼓励
\(Z\)
去预测
\(Y\)
。综合来看，信息瓶颈原理的目标为压缩
\(X\)
以得到表征
\(Z\)
，该表征能够去除掉对预测
\(Y\)
无用的因素而保留相关因素
^[14]
。这也就是说IB使得
\(Z\)
做为一个最小充分统计量
^[15]
（在我们这个CDR应用中即领域间应该共享的信息）。在实践中，直接优化互信息是难解（intractable）的，因此变分近似
^[16]
常常用于构建用于优化互信息目标函数的下界
^[13][17]
。

本文提出的CDIRB模型包含变分子图编码器（variational bipartite graph encoder，VBGE）和两种的跨领域信息正则项，整体框架图如下图所示：

其中绿色部分的网格代表物品表征，黄色和蓝色颜色的网格分别代表重叠和不重叠的用户表征。信息瓶颈正则项（图中的Information Bottleneck）捕捉了领域间用户和物品的相关性，而对比信息正则项（图中的Contrastive Information）则捕捉了领域间重叠用户之间的相关性。

接下来我们叙述每个部分的细节。

嵌入层

嵌入层得到的领域
\(X\)
的用户/物品表征分别记作
\(\boldsymbol{U}^X \in \mathbb{R}^{|\mathcal{U}^X |\times F}\)
和
\(\boldsymbol{V}^X \in \mathbb{R}^{\left|\mathcal{V}^X\right| \times F}\)
；领域
\(Y\)
的用户/物品表征分别记作
\(\boldsymbol{U}^Y \in \mathbb{R}^{\left|\mathcal{U}^{Y}\right| \times F}\)
和
\(\boldsymbol{V}^Y \in \mathbb{R}^{\left|\mathcal{V}^Y\right| \times F}\)
。

变分二分图编码器
为了在原始用户/物品表征的基础上，进一步提炼出用户/物品的隐向量表征，论文提出了变分二分图编码器（VBGE）。比如，生成
\(X\)
领域的用户隐向量表征
\(Z_v^X\)
的过程如下：

信息瓶颈正则项

接下来，论文引入了信息瓶颈正则项和对比信息正则项这两种正则项来捕捉领域间的相关性，以学得包含领域间共享信息的无偏表征。其中信息瓶颈正则化项旨在捕捉领域间用户和物品间的相关性，而对比信息正则化项旨在捕捉领域间的重叠用户和用户之间的相关性。

设
\(\mathbf{X}\)
，
\(\mathbf{X}^u\)
，
\(\mathbf{X}^v\)
分别为领域
\(X\)
中所观测到的交互信息、用户信息和物品信息。领域
\(X\)
的用户集合包括重叠用户
\(\mathcal{U}^o\)
和非重叠用户
\(\mathcal{U}^x\)
这两个群体，领域
\(Y\)
亦然。以领域
\(X\)
为例，将用户表征
\(\boldsymbol{Z}_u^X \in \mathbb{R}^{\left|\mathcal{U}^{X}\right| \times F}\)
也划分为两个群体：
\(\boldsymbol{Z}_u^{x o} \in \mathbb{R}^{\left|\mathcal{U}^o\right| \times F}\)
和
\(\boldsymbol{Z}_u^x \in \mathbb{R}^{\left|\mathcal{U}^x\right| \times F}\)
。

信息瓶颈正则项又可继续分为跨域（cross-domain）信息瓶颈正则项和领域内（in-domain）信息瓶颈正则项。首先我们来看跨域（cross-domain）信息瓶颈正则项，它包括包括信息压缩（即互信息最小化）和重构两部分，其结构化示意图如下：

正如上图（a）所示。
\(\boldsymbol{Z}_u^{x o}\)
，
\(\boldsymbol{Z}_u^{y o}\)
是编码了各领域用户信息的重叠用户表征，而图（b）中的
\(\boldsymbol{Z}_u^x\)
，
\(Z_u^y\)
是非重叠的（冷启动）用户表征。这里
\(\boldsymbol{Z}_v^{X}\)
，
\(\boldsymbol{Z}_v^Y\)
是物品表征，默认是不重叠的。

以
\(X\)
领域迁移到
\(Y\)
领域为例（图中标红部分），我们需要使重叠用户隐向量
\(\boldsymbol{Z}^{xo}_u\)
和同领域的用户表征
\(\mathbf{X}^u\)
互斥（信息压缩），而去接近于
\(Y\)
领域的交互信息
\(\mathbf{Y}\)
（跨域重构）；此外，对于
\(Y\)
领域的物品隐向量
\(\boldsymbol{Z}^Y_v\)
也需要使其与物品表征
\(\mathbf{Y}^v\)
互斥，并去接近于
\(\mathbf{Y}\)
（因为不同领域物品不会重叠，这里采取域内重构）。

其中的跨域重构部分可以进一步通过互信息链式法则化简得到：

（这里假设
\(\boldsymbol{Z}_u^{x o}\)
和
\(\boldsymbol{Z}_v^Y\)
独立）

最后，
\(X\)
领域导出的损失函数包括最小化（minimality）和跨域重构（reconstruction）两部分：

接下来我们来看领域内（in-domain）信息瓶颈正则项，其结构化示意图如下：

我们还是以
\(X\)
领域为例子（图中红色箭头部分），可以看到其损失函数同样也包括最小化和领域内重构两部分：

对比信息正则项

在对比信息正则化项中，作者通过最大化
\(X\)
的重叠用户表征
\(\boldsymbol{Z}^{xo}_u\)
和来自领域
\(Y\)
的重叠用户表征
\(\boldsymbol{Z}^{yo}_u\)
间的互信息，以进一步提炼重叠用户的表征。对比信息正则化项的定义如下所示：

可求解的目标函数

将上述的两种信息瓶颈正则项和对比信息正则项累加起来（同时包括
\(X\)
和
\(Y\)
领域的），就得到了目标函数：

要想求解该目标函数，接下来还需要将互信息其转换为KL散度，比如对于
\(I\left(\boldsymbol{Z}_u^{x o} ; \mathbf{X}^u\right)\)
就有

该互信息项是难以求解的，这里需要转而去优化其上界：

对于重构项，我们以
\(I\left(\boldsymbol{Z}_u^{x o}, \boldsymbol{Z}_v^Y ; \mathbf{Y}\right)\)
为例，我们有

该优化函数同样是难解的，这里需要转而去优化其下界：

对于对比互信息项，论文借鉴了infomax
^[14][20]
小想法，利用神经网络来度量对比互信息。具体来说，论文定义了判别器
\(\mathcal{D}\)
来度量来自不同领域的重叠用户隐向量（来自领域
\(X\)
的
\(z^{xo}_{u_i}\)
和来自领域
\(Y\)
的
\(z^{yo}_{u_i}\)
）之间的相似度。因此，对比项的下界可表示如下：

这里

这样，我们就将原始目标函数转化为了最终完全可求解的目标函数。

2.2 SIGIR 2022 《DisenCDR: Learning Disentangled Representations for Cross-Domain Recommendation》
^[4]

本方法属于采用解耦表征的跨域推荐方法。与2.1所讲的基于信息瓶颈视角的方法不同的是，本方法旨在
为两个领域中的重叠用户做推荐，因此在模型中只考虑两个领域中的重叠用户
。在本方法中，所要解决的关键问题在于
对于两个领域重叠用户的表征，如何分别出共享和不共享的部分？

为了解决该问题，
本文基于信息论提出了DisenCDR模型，该模型能够解耦领域间共享和领域特有的信息，从而只迁移领域间共享的信息以增强推荐表现
。该方法包含了两个互信息正则项（包括
用于解耦的正则项
和
用于信息增强的正则项
，详情参见后文），并据此导出了一个可以求解的解耦目标函数。

本文采用和上面的文章几乎一样的符号，就是需要注意此处领域
\(X\)
和领域
\(Y\)
的用户空间相同。设领域
\(X\)
和领域
\(Y\)
的数据分别表示为
\(\mathcal{D}^X=(\mathcal{U}, \mathcal{V}^X,\mathcal{E}^X)\)
，
\(\mathcal{D}^Y=(\mathcal{U}, \mathcal{V}^X,\mathcal{E}^X)\)
，这里
\(\mathcal{U}\)
、
\(\mathcal{V}\)
、
\(\mathcal{E}\)
分别表示每个领域用户、物品和边的集合。设
\(\boldsymbol{A}^X=\{0,1\}^{\left|\mathcal{U}\right| \times\left|\mathcal{V}^X\right|}\)
和
\(A^Y=\{0,1\}^{\left|\mathcal{U}\right| \times\left|\mathcal{V}^Y\right|}\)
为存储用户-物品交互信息的两个二值矩阵。

这里
\(Z^X_v\)
，
\(Z^X_u\)
，
\(Z^Y_u\)
和
\(Z^Y_v\)
是领域特有的用户/物品表征，且
\(Z^S_u\)
是用户的领域共享表征，则DisenCDR的框架图可表示如下：

注意，这里蓝色的KL意为使用先验分布
\(\mathcal{N}(0, \mathbf{I})\)
计算KL散度，绿色的KL意为计算输入之间的KL散度。隐变量
\(\widehat{Z}_u^S\)
、
\(\widetilde{Z}_u^S\)
用于计算我们的解耦目标函数。

下面我们来详细介绍该方法各个组成部分的细节：

嵌入层

嵌入层的作用同2.1中所述的方法相同，也即将用户和物品嵌入到低维空间中。不过还是正如我们前面所说的，这里
\(X\)
领域和
\(Y\)
领域的用户空间相同。设
\(\boldsymbol{U}^S\in \mathbb{R}^{|\mathcal{U}|\times F}\)
为领域
\(X\)
和领域
\(Y\)
的共享初始嵌入矩阵，
\(\boldsymbol{U}^X \in \mathbb{R}^{|\mathcal{U}|\times F}\)
和
\(\boldsymbol{V}^X \in \mathbb{R}^{\left|\mathcal{U}\right| \times F}\)
分别为领域
\(X\)
和
\(Y\)
特有的初始化嵌入矩阵。此外，
\(\boldsymbol{V}^X \in \mathbb{R}^{\left|\mathcal{V}^X\right| \times F}\)
和
\(\boldsymbol{V}^Y \in \mathbb{R}^{\left|\mathcal{V}^Y\right| \times F}\)
分别为领域
\(X\)
和领域
\(Y\)
的物品表征。

变分二分图编码器

DisenCDR和变分二分图编码器和我们 2.1 中讲的第一个基于信息瓶颈思想的模型一样，唯一的区别就是这里的共享隐向量同时利用了
\(X\)
领域的
\(\boldsymbol{\overline{\mu}}_{u}^X\)
和
\(Y\)
领域的
\(\overline{\boldsymbol{\mu}}_u^Y\)
：

生成和推断

论文遵循VAE
^[18]
的框架，这里假定所观测的交互信息
\(\mathcal{D}^X\)
和
\(\mathcal{ D}^Y\)
采自一个联合概率分布
\(p_{\mathcal{D}}(u, v^X, v^Y)\)
，每个元组
\(\left(u_i, v_j, v_k\right) \sim p_{\mathcal{D}}\left(u, v^X, v^Y\right)\)
描述了用户
\(u_i\)
和物品
\(v_j \in \mathcal{V}^X\)
和物品
\(v_k \in \mathcal{V}^Y\)
的交互信息。而交互数据正是经由领域共享表征(比如
\(Z_u^S\)
）和领域特有（比如
\(Z^X_u\)
，
\(Z^X_v\)
，
\(Z^Y_u\)
和
\(Z^Y_v\)
）表征生成，也即：

下图（a）正是描述了交互数据的生成过程，而图（b）则描述了反向推断步骤：

在推断过程中，直接最大化联合概率分布
\(p_\theta\left(u, v^X, v^Y\right)\)
的似然是难解的，因为后验分布
\(p_\theta\left(Z_u^X, Z_u^Y, Z_u^S, Z_v^X, Z_v^Y \mid \mathbf{X}, \mathrm{Y}\right)\)
未知。因此采用近似推断
^[19]
来近似真实的后验分布。根据上图（b）中的结构化假设，论文将近似后验分布分解为：

解耦目标函数

接下来作者从信息论的角度来探究领域间表征纠缠的问题，并推导了一个解耦目标函数。

为了使领域间共享和领域特有的隐向量能够编码互斥的信息，作者引入了互斥正则项来最小化二者的互信息。为了分析最小化互信息的影响，作者又将互信息进行了进一步改写。我们以领域
\(X\)
为例，其对应的领域共享和领域特有隐向量的互信息
\(I(Z^X_u; Z^S_u)\)
可做如下改写：

接下来我们看另外一个用于信息增强的正则项（对应我们在2.2信息瓶颈方法中所介绍的重构正则项）。该正则项旨在使每个领域共享的表征
\(Z^S_u\)
信息更丰富，这里作者最大化互信息
\(I\left(Z_u^S ; \mathrm{X} ; \mathrm{Y}\right)\)
来使得
\(Z^S_u\)
编码领域共享的信息。我们以领域
\(X\)
为例，有：

总目标函数
将上面所说的两个解耦目标函数（包括
\(X\)
领域和
\(Y\)
领域的）加起来，就得到了总的目标函数：

进一步将物品隐向量
\(Z_v^X\)
和
\(Z_v^Y\)
引入，可以将损失函数放缩为：

这样，解耦目标函数中的一部分可以视为变分推断中标准的证据下界（Evidence Lower Bound， ELBO）。最后，论文按照VAE的思路，继续将其化为了可以求解的目标函数：

参考

• [1] Lin X, Wu J, Zhou C, et al. Task-adaptive neural process for user cold-start recommendation[C]//Proceedings of the Web Conference 2021. 2021: 1306-1316.

• [2] Zhu F, Wang Y, Chen C, et al. Cross-domain recommendation: challenges, progress, and prospects[J]. arXiv preprint arXiv:2103.01696, 2021.

• [3] Hu G, Zhang Y, Yang Q. Conet: Collaborative cross networks for cross-domain recommendation[C]//Proceedings of the 27th ACM international conference on information and knowledge management. 2018: 667-676

• [4] Li P, Tuzhilin A. Ddtcdr: Deep dual transfer cross domain recommendation[C]//Proceedings of the 13th International Conference on Web Search and Data Mining. 2020: 331-339.

• [5] Kipf T N, Welling M. Semi-supervised classification with graph convolutional networks[J]. arXiv preprint arXiv:1609.02907, 2016.

• [6] Meng Liu, Jianjun Li, Guohui Li, and Peng Pan. 2020. Cross Domain Recom- mendation via Bi-directional Transfer Graph Collaborative Filtering Networks. In ACM International Conference on Information and Knowledge Management (CIKM).

• [7] Xiangnan He, Kuan Deng, Xiang Wang, Yan Li, Yongdong Zhang, and Meng Wang. 2020. Lightgcn: Simplifying and Powering Graph Convolution Network for Recommendation. In ACM International Conference on Research on Development in Information Retrieval (SIGIR).

• [8] Cao J, Sheng J, Cong X, et al. Cross-domain recommendation to cold-start users via variational information bottleneck[C]//2022 IEEE 38th International Conference on Data Engineering (ICDE). IEEE, 2022: 2209-2223.

• [9] Zang T, Zhu Y, Liu H, et al. A survey on cross-domain recommendation: taxonomies, methods, and future directions[J]. ACM Transactions on Information Systems, 2022, 41(2): 1-39.

• [10] Cao J, Lin X, Cong X, et al. DisenCDR: Learning Disentangled Representations for Cross-Domain Recommendation[C]//Proceedings of the 45th International ACM SIGIR Conference on Research and Development in Information Retrieval. 2022: 267-277.

• [11] Tishby N, Pereira F C, Bialek W. The information bottleneck method[J]. arXiv preprint physics/0004057, 2000.

• [12] Tishby N, Zaslavsky N. Deep learning and the information bottleneck principle[C]//2015 ieee information theory workshop (itw). IEEE, 2015: 1-5.

• [13] Alemi A A, Fischer I, Dillon J V, et al. Deep variational information bottleneck[J]. arXiv preprint arXiv:1612.00410, 2016.

• [14] M. I. Belghazi, A. Baratin, S. Rajeshwar, S. Ozair, Y. Bengio, A. Courville, and D. Hjelm, “Mutual infor- mation neural estimation,” in International Conference on Machine Learning (ICML), 2018.

• [15] Wu, H. Ren, P. Li, and J. Leskovec, “Graph infor- mation bottleneck,” in Annual Conference on Neural Information Processing Systems (NeurIPS), 2020.

• [16] S. Gershman and N. Goodman, “Amortized inference in probabilistic reasoning,” in Proceedings of the Annual Meeting of The Cognitive Science Society, 2014.

• [17] Wang Z, Chen X, Wen R, et al. Information theoretic counterfactual learning from missing-not-at-random feedback[J]. Advances in Neural Information Processing Systems, 2020, 33: 1854-1864.

• [18] Kingma D P, Welling M. Auto-encoding variational bayes[J]. arXiv preprint arXiv:1312.6114, 2013.

• [19] Gershman S, Goodman N. Amortized inference in probabilistic reasoning[C]//Proceedings of the annual meeting of the cognitive science society. 2014, 36(36).

• [20] Hjelm R D, Fedorov A, Lavoie-Marchildon S, et al. Learning deep representations by mutual information estimation and maximization[J]. arXiv preprint arXiv:1808.06670, 2018.

一图胜千言：

flow和miniflow

在介绍之前先说一些概念：里面有两个结构很重要，一个是flow一个是miniflow这里介绍一下他们的数据结构和构造函数。

flow：

flow的特点是8字节对齐的，存储报文相关字段和其他原数据，用于匹配流表，数据包含四个层次：

1. metadata: 入端口号，寄存器等信息
2. l2: 源目的mac，vlan和mpls等信息
3. l3: ipv4/ipv6源目的ip，ttl等信息
4. l4: 源目的端口号，icmp code和type等信息。

flow的坏处就是占用了很大的字节，并且有很多字段都是0，在2.8版本中flow的大小是672字节。

miniflow

miniflow是flow的压缩版，因为flow占用字节很大，比如可以支持ARP，IP等报文，填充了arp字段，icmp报文就是空的了，浪费了很多信息。过程中用到hash作为key，也是根据miniflow计算hash值，不是用的flow。

structminiflow {structflowmap map;

};structflowmap {

    map_t bits[FLOWMAP_UNITS];

};

miniflow其包含两部分内容：

1. struct flowmap map;是bit数组，使用其中的bit表示flow中哪个8字节存在有效数据，flow中占多少个8字节，那么就需要map中多个个bit，并且按照64bit向上取整。
2. 第二部分是有效数据，有效数据动态分配，根据struct flowmap map;中1bit数个数进行分配，大小为bit数*8字节，该部分直接跟在map后面。该部分存储在netdev_flow_key结构中的buf数组。

miniflow数据结构：

//flow是8字节对齐的，除8得到flow中包含8字节的个数
#define FLOW_U64S (sizeof(struct flow) / sizeof(uint64_t))

//map大小为8字节，MAP_T_BITS 为64位
typedef unsigned long longmap_t;#define MAP_T_BITS (sizeof(map_t) * CHAR_BIT)

//每位表示一个u64，FLOWMAP_UNITS 表示最少需要几个64位
#define FLOWMAP_UNITS DIV_ROUND_UP(FLOW_U64S, MAP_T_BITS)

structflowmap {

    map_t bits[FLOWMAP_UNITS];

};structminiflow {structflowmap map;/*Followed by:

     *     uint64_t values[n];

     * where 'n' is miniflow_n_values(miniflow).*/};structnetdev_flow_key {

    uint32_t hash;     

    uint32_t len;struct miniflow mf;  //bits
    uint64_t buf[FLOW_MAX_PACKET_U64S];  //就是上边所说的value
};//有些字段是互斥的
#define FLOW_MAX_PACKET_U64S (FLOW_U64S                                   \
    /*Unused in datapath*/  -FLOW_U64_SIZE(regs)                       \-FLOW_U64_SIZE(metadata)                   \/*L2.5/3*/              - FLOW_U64_SIZE(nw_src)  /*incl. nw_dst*/\-FLOW_U64_SIZE(mpls_lse)                   \/*L4*/                  -FLOW_U64_SIZE(tp_src)                     \

                             )

miniflow优点：

1. 使用miniflow可以节省内存
2. 如果只想遍历flow中的非0字段时，使用miniflow找到对应的非0字段，可以节省时间

flow->miniflow函数：
miniflow_extract()

voidminiflow_extract(struct dp_packet *packet, struct miniflow *dst)

{

    ...//初始化赋值有两个关键，一个是这个values： return (uint64_t *)(mf + 1);//就是上边说的
    uint64_t *values =miniflow_values(dst);struct mf_ctx mf ={ FLOWMAP_EMPTY_INITIALIZER, values,

                         values+FLOW_U64S };

    ...if (md->skb_priority || md->pkt_mark) {

        miniflow_push_uint32(mf, skb_priority, md->skb_priority);

        miniflow_push_uint32(mf, pkt_mark, md->pkt_mark);

    }

    miniflow_push_be16(mf, dl_type, dl_type);

    miniflow_pad_to_64(mf, dl_type);

    ...//去取网络层信息,从这里可以看出，ovs暂时只支持IP,IPV6,ARP,RARP报文
    if (OVS_LIKELY(dl_type ==htons(ETH_TYPE_IP))){...}else if...//提取传输层，从这里可以看出，ovs暂时支持传输层协议有TCP,UDP,SCTP,ICMP,ICMPV6
    if (OVS_LIKELY(nw_proto ==IPPROTO_TCP)){...}else if...

miniflow_push_uint32()

在上面将value保存到miniflow时，用到了几个辅助函数，比如下面的miniflow_push_uint32用来将一个32位的值保存到miniflow中FIELD对应的位置。其首先调用offsetof获取field在flow中的偏移字节数，因为flow是8字节对齐的，所以一个四字节的成员变量要么位于8字节的起始位置，要么位于8字节的中间位置，即对8取模值肯定为0或者4，再调用miniflow_push_uint32_保存到对应的位置，并设置map中对应的bit为1。

#define miniflow_push_uint32(MF, FIELD, VALUE)                      \miniflow_push_uint32_(MF, offsetof(structflow, FIELD), VALUE)#define miniflow_push_uint32_(MF, OFS, VALUE)   \{                                           \

    MINIFLOW_ASSERT(MF.data<MF.end);          \

                                                \//成员变量位于起始位置，需要调用miniflow_set_map设置对应的bit为1
    if ((OFS) % 8 == 0) {                       \

        miniflow_set_map(MF, OFS/ 8);          \*(uint32_t *)MF.data =VALUE;           \

    }else if ((OFS) % 8 == 4) {                \//成员变量不在起始位置，要判断此变量所在的bit为1
        miniflow_assert_in_map(MF, OFS / 8);    \*((uint32_t *)MF.data + 1) =VALUE;     \

        MF.data++;                              \

    }                                           \

}

miniflow->flow函数：
miniflow_expand()

/*Initializes 'dst' as a copy of 'src'.*/
voidminiflow_expand(const struct miniflow *src, struct flow *dst)

{

    memset(dst,0, sizeof *dst);

    flow_union_with_miniflow(dst, src);

}/*Perform a bitwise OR of miniflow 'src' flow data with the equivalent

 * fields in 'dst', storing the result in 'dst'.*/
static inline voidflow_union_with_miniflow(struct flow *dst, const struct miniflow *src)

{

    flow_union_with_miniflow_subset(dst, src, src->map);

}static inline voidflow_union_with_miniflow_subset(struct flow *dst, const struct miniflow *src,structflowmap subset)

{

    uint64_t*dst_u64 = (uint64_t *) dst;const uint64_t *p =miniflow_get_values(src);

    map_t map;//遍历所有的map
FLOWMAP_FOR_EACH_MAP (map, subset) {

        size_t idx;//遍历map中所有的非0bit
MAP_FOR_EACH_INDEX(idx, map) {

            dst_u64[idx]|= *p++;

        }

        dst_u64+=MAP_T_BITS;

    }

}

流表查询过程

概要

该部分入口在lib/dpif-netdev.c，就是最开始的那个图。

查询的缓存分为两层：一个是DFC，一个是dpcls，相当于microflow和megaflow，DFC由两部分组成，DFC（datapath flow cache）：EMC（Exact match cache）+SMC（Signature match cache），另一部分就是dpcls(datapath classifer)。

SMC默认关闭：bool smc_enable = smap_get_bool(other_config, "smc-enable", false);

函数执行流程(不包含
SMC
的)：

入口在dp_netdev_input__()

static voiddp_netdev_input__(struct dp_netdev_pmd_thread *pmd,struct dp_packet_batch *packets,boolmd_is_valid, odp_port_t port_no)

{#if !defined(__CHECKER__) && !defined(_WIN32)
    const size_t PKT_ARRAY_SIZE =dp_packet_batch_size(packets);#else
    /*Sparse or MSVC doesn't like variable length array.*/
    enum { PKT_ARRAY_SIZE =NETDEV_MAX_BURST };#endifOVS_ALIGNED_VAR(CACHE_LINE_SIZE)structnetdev_flow_key keys[PKT_ARRAY_SIZE];struct netdev_flow_key *missed_keys[PKT_ARRAY_SIZE];structpacket_batch_per_flow batches[PKT_ARRAY_SIZE];

    size_t n_batches;structdp_packet_flow_map flow_map[PKT_ARRAY_SIZE];

    uint8_t index_map[PKT_ARRAY_SIZE];

    size_t n_flows, i;



    odp_port_t in_port;



    n_batches= 0;//1. dfc_processing之后会把miss的放到packets里//找到的可能已经batched了，或者放到flow_map里了//flow_map里是未bathed的，可能直接是*flow或者是NULL，是NULL再去下一层cache查
    dfc_processing(pmd, packets, keys, missed_keys, batches, &n_batches,

                   flow_map,&n_flows, index_map, md_is_valid, port_no);//2. 如果有miss的，再去找fast-path，也就是查dpcls
    if (!dp_packet_batch_is_empty(packets)) {  

        in_port= packets->packets[0]->md.in_port.odp_port;

        fast_path_processing(pmd, packets, missed_keys,

                             flow_map, index_map, in_port);

    }/*Batch rest of packets which are in flow map.*/
    for (i = 0; i < n_flows; i++) {struct dp_packet_flow_map *map = &flow_map[i];if (OVS_UNLIKELY(!map->flow)) {continue;

        }

        dp_netdev_queue_batches(map->packet, map->flow, map->tcp_flags,

                                batches,&n_batches);

    }for (i = 0; i < n_batches; i++) {

        batches[i].flow->batch =NULL;

    }//执行每个packet的action
    for (i = 0; i < n_batches; i++) {

        packet_batch_per_flow_execute(&batches[i], pmd);

    }

}

1. DFC查询：dfc_processing()

staticinline size_t

dfc_processing(struct dp_netdev_pmd_thread *pmd,struct dp_packet_batch *packets_,struct netdev_flow_key *keys,struct netdev_flow_key **missed_keys,struct packet_batch_per_flow batches[], size_t *n_batches,struct dp_packet_flow_map *flow_map,

               size_t*n_flows, uint8_t *index_map,boolmd_is_valid, odp_port_t port_no)

{struct netdev_flow_key *key = &keys[0];

    size_t n_missed= 0, n_emc_hit = 0;struct dfc_cache *cache = &pmd->flow_cache;struct dp_packet *packet;

    size_t cnt=dp_packet_batch_size(packets_);//emc的插入概率，如果为0，表示不开启emc
    uint32_t cur_min = pmd->ctx.emc_insert_min;inti;

    uint16_t tcp_flags;boolsmc_enable_db;//记录未batched的个数
    size_t map_cnt = 0;//这个变量用于保序
    bool batch_enable = true;//获取smc是否开启参数
    atomic_read_relaxed(&pmd->dp->smc_enable_db, &smc_enable_db);

    pmd_perf_update_counter(&pmd->perf_stats,

                            md_is_valid?PMD_STAT_RECIRC : PMD_STAT_RECV,

                            cnt);



    do_dfc_hook(pmd, packets_, batches, n_batches); 

    cnt=dp_packet_batch_size(packets_);//逐个对dp_packet_batch中的每一个packet进行处理
DP_PACKET_BATCH_REFILL_FOR_EACH (i, cnt, packet, packets_) {struct dp_netdev_flow *flow;//若packet包长小于以太头的长度直接丢包
        if (OVS_UNLIKELY(dp_packet_size(packet) <ETH_HEADER_LEN)) {

            dp_packet_delete(packet);

            COVERAGE_INC(datapath_drop_rx_invalid_packet);continue;

        }//对数据手工预取可减少读取延迟，从而提高性能
        if (i != cnt - 1) {struct dp_packet **packets = packets_->packets;/*Prefetch next packet data and metadata.*/OVS_PREFETCH(dp_packet_data(packets[i+1]));

            pkt_metadata_prefetch_init(&packets[i+1]->md);

        }//初始化metadata首先将pkt_metadata中flow_in_port前的字节全部设为0//将in_port.odp_port设为port_no， tunnel.ipv6_dst设为in6addr_any
        if (!md_is_valid) {

            pkt_metadata_init(&packet->md, port_no);

        }//报文转化为miniflow, 上文有讲
        miniflow_extract(packet, &key->mf);  

        key->len = 0; /*Not computed yet.*/
        //计算当前报文miniflow的hash值
        key->hash =(md_is_valid== false)? dpif_netdev_packet_get_rss_hash_orig_pkt(packet, &key->mf)

                : dpif_netdev_packet_get_rss_hash(packet,&key->mf);//根据key->hash,emc_entry alive,miniflow 3个条件得到dp_netdev_flow//cur_min = 0,表示不可能插入，后面有讲什么时候才会插入EMC
        flow = (cur_min != 0) ? emc_lookup(&cache->emc_cache, key) : NULL;if(OVS_LIKELY(flow)) {

            tcp_flags= miniflow_get_tcp_flags(&key->mf);

            n_emc_hit++; //命中次数+1//为了保证报文的顺序，所有的packet对应的flow都用flow_map存储//flow_map里面就是packet数量对应的(packet,flow,tcp_flag)//最后会把这些在dp_netdev_input__里重新把顺序合并一下
            if(OVS_LIKELY(batch_enable)) {//把查到的flow加到batches里第n_batches个batch里
dp_netdev_queue_batches(packet, flow, tcp_flags, batches,

                                        n_batches);

            }else{

 

                packet_enqueue_to_flow_map(packet, flow, tcp_flags,

                                           flow_map, map_cnt++);

            }

        }else{//这些数据结构用于smc查询时的记录//没查到把packet放到packets_里，从下标0再开始放//最后packets_都是未查到的
dp_packet_batch_refill(packets_, packet, i);

            index_map[n_missed]=map_cnt;

            flow_map[map_cnt++].flow =NULL;

            missed_keys[n_missed]=key;

            key= &keys[++n_missed];

            batch_enable= false; //之后的都是未batched的
}

    }*n_flows =map_cnt;



    pmd_perf_update_counter(&pmd->perf_stats, PMD_STAT_EXACT_HIT, n_emc_hit);//如果没有开启smc，直接返回了
    if (!smc_enable_db) {returndp_packet_batch_size(packets_);

    }



    smc_lookup_batch(pmd, keys, missed_keys, packets_,

                     n_missed, flow_map, index_map);returndp_packet_batch_size(packets_);

}

1.1 emc查询：emc_lookup()

static inline struct dp_netdev_flow *emc_lookup(struct emc_cache *cache, const struct netdev_flow_key *key)

{struct emc_entry *current_entry;//这里说一下，一个hash分配两个桶，长度为13位，cache桶的大小为1<<13//struct emc_cache {//struct emc_entry entries[EM_FLOW_HASH_ENTRIES];//int sweep_idx;                /* For emc_cache_slow_sweep(). *///};
    EMC_FOR_EACH_POS_WITH_HASH (cache, current_entry, key->hash) {if (current_entry->key.hash == key->hash&&emc_entry_alive(current_entry)&& emc_flow_key_equal_mf(&current_entry->key, &key->mf)) {/*We found the entry with the 'key->mf' miniflow*/
            return current_entry->flow;

        }

    }returnNULL;

}#define EM_FLOW_HASH_SHIFT 13
#define EM_FLOW_HASH_ENTRIES (1u << EM_FLOW_HASH_SHIFT)
#define EM_FLOW_HASH_MASK (EM_FLOW_HASH_ENTRIES - 1)
#define EM_FLOW_HASH_SEGS 2
#define EMC_FOR_EACH_POS_WITH_HASH(EMC, CURRENT_ENTRY, HASH)                 \
    for (uint32_t i__ = 0, srch_hash__ =(HASH);                             \

         (CURRENT_ENTRY)= &(EMC)->entries[srch_hash__ &EM_FLOW_HASH_MASK], \

         i__<EM_FLOW_HASH_SEGS;                                            \

         i__++, srch_hash__ >>=EM_FLOW_HASH_SHIFT)//比较miniflow是否相同
static inline boolemc_flow_key_equal_mf(const struct netdev_flow_key *key,const struct miniflow *mf)

{return !memcmp(&key->mf, mf, key->len);

}

EMC
查询函数执行：

1.2 smc查询：smc_lookup_batch()

static inline voidsmc_lookup_batch(struct dp_netdev_pmd_thread *pmd,struct netdev_flow_key *keys,struct netdev_flow_key **missed_keys,struct dp_packet_batch *packets_,const intcnt,struct dp_packet_flow_map *flow_map,

            uint8_t*index_map)

{inti;struct dp_packet *packet;

    size_t n_smc_hit= 0, n_missed = 0;struct dfc_cache *cache = &pmd->flow_cache;struct smc_cache *smc_cache = &cache->smc_cache;const struct cmap_node *flow_node;intrecv_idx;

    uint16_t tcp_flags;/*Prefetch buckets for all packets*/
    for (i = 0; i < cnt; i++) {

        OVS_PREFETCH(&smc_cache->buckets[keys[i].hash &SMC_MASK]);

    }



    DP_PACKET_BATCH_REFILL_FOR_EACH (i, cnt, packet, packets_) {struct dp_netdev_flow *flow =NULL;//找到hash相同的flow链表的头节点
        flow_node =smc_entry_get(pmd, keys[i].hash);bool hit = false;/*Get the original order of this packet in received batch.*/recv_idx=index_map[i];if (OVS_LIKELY(flow_node !=NULL)) {//遍历一下看看哪一个是相同的，这个通过offsetof找到存放该cmap结构体的首地址//dp_netdev_flow里面的首地址就是，
CMAP_NODE_FOR_EACH (flow, node, flow_node) {/*Since we dont have per-port megaflow to check the port

                 * number, we need to  verify that the input ports match.*/
                if (OVS_LIKELY(dpcls_rule_matches_key(&flow->cr, &keys[i]) &&flow->flow.in_port.odp_port == packet->md.in_port.odp_port)) {

                    tcp_flags= miniflow_get_tcp_flags(&keys[i].mf);

                    keys[i].len=netdev_flow_key_size(miniflow_n_values(&keys[i].mf));if (emc_probabilistic_insert(pmd, &keys[i], flow)) {if (flow->status ==OFFLOAD_NONE) {

                            queue_netdev_flow_put(pmd->dp->dp_flow_offload, \

                                    pmd->dp->class, \

                                    flow, NULL, DP_NETDEV_FLOW_OFFLOAD_OP_ADD);

                        }

                    }

                    packet_enqueue_to_flow_map(packet, flow, tcp_flags,

                                               flow_map, recv_idx);

                    n_smc_hit++;

                    hit= true;break;

                }

            }if(hit) {continue;

            }

        }//SMC也miss了，和之前一样，把miss的放packets_里，从0开始放
dp_packet_batch_refill(packets_, packet, i);

        index_map[n_missed]=recv_idx;

        missed_keys[n_missed++] = &keys[i];

    }



    pmd_perf_update_counter(&pmd->perf_stats, PMD_STAT_SMC_HIT, n_smc_hit);

}

查找
hash
相同的链表头：smc_entry_get()

static inline const struct cmap_node *smc_entry_get(struct dp_netdev_pmd_thread *pmd, constuint32_t hash)

{struct smc_cache *cache = &(pmd->flow_cache).smc_cache;//smc_cache桶的大小是（1<<18），SMC_MASK=（1<<18）- 1//先通过后hash的后18位定位到桶
    struct smc_bucket *bucket = &cache->buckets[hash &SMC_MASK];//一个桶有4个16位的sig，存key->hash前16位，正好是64位//遍历4个元素看那个匹配，获得匹配后的cmap的下标
    uint16_t sig = hash >> 16;

    uint16_t index=UINT16_MAX;for (int i = 0; i < SMC_ENTRY_PER_BUCKET; i++) {if (bucket->sig[i] ==sig) {

            index= bucket->flow_idx[i];break;

        }

    }//通过index找到在dpcls里的桶位置
    if (index !=UINT16_MAX) {return cmap_find_by_index(&pmd->flow_table, index);

    }returnNULL;

}

1.3 更新emc：
emc_probabilistic_insert()

命中
SMC
后，插入回上一层cache（
EMC
）里：emc_probabilistic_insert()

插入EMC的条件:

默认插入流表的概率是1%，可以通过ovs-vsctl set Open_vSwitch . other_config:emc-insert-prob=10 设置概率，表示平均10条流表有1条插入，当为0时禁用EMC，当为1的时候，百分百插入。设置后会在代码里设置
emc_insert_min
字段为uint_max/10，插入的时候生成一个uint_random()，如果随机数小于
emc_insert_min
才会插入。

static inline boolemc_probabilistic_insert(struct dp_netdev_pmd_thread *pmd,const struct netdev_flow_key *key,struct dp_netdev_flow *flow)

{/*Insert an entry into the EMC based on probability value 'min'. By

     * default the value is UINT32_MAX / 100 which yields an insertion

     * probability of 1/100 ie. 1%*/uint32_t min= pmd->ctx.emc_insert_min;if (min && random_uint32() <=min) {

        emc_insert(&(pmd->flow_cache).emc_cache, key, flow);return true;

    }return false;

}

emc_insert同样有我在内核查询里的问题，如果cache里没有该miniflow，会找一个hash值小的entry，覆盖这个entry，那如果有一个hash很大的flow被插入了，但是这个flow之后就没用过了，那岂不是这个entry就浪费了，不会被用到。

找到了合适的emc_entry。则将报文对应的netdev_dev_flow key信息存储到该表项中。而对于这个表项，原有的emc_entry.flow有可能还有指向一条旧的流表，需要将这条流表的引用计数减1，如果减1后达到0，则释放该流表空间。同时更新emc_entry.flow重新指向新的流表。到此为止，EMC表项更新完毕。

static inline voidemc_insert(struct emc_cache *cache, const struct netdev_flow_key *key,struct dp_netdev_flow *flow)

{struct emc_entry *to_be_replaced =NULL;struct emc_entry *current_entry;



    EMC_FOR_EACH_POS_WITH_HASH(cache, current_entry, key->hash) {if (netdev_flow_key_equal(&current_entry->key, key)) {/*We found the entry with the 'mf' miniflow*/emc_change_entry(current_entry, flow, NULL);return;

        }/*Replacement policy: put the flow in an empty (not alive) entry, or

         * in the first entry where it can be*/

        if (!to_be_replaced||(emc_entry_alive(to_be_replaced)&& !emc_entry_alive(current_entry))|| current_entry->key.hash < to_be_replaced->key.hash) {//这个黄色判断就是我迷惑的地方
            to_be_replaced =current_entry;

        }

    }/*We didn't find the miniflow in the cache.

     * The 'to_be_replaced' entry is where the new flow will be stored*/emc_change_entry(to_be_replaced, flow, key);

}

1.4 EMC的轮训更新

在pmd_thread_main()里面：

if (lc++ > 1024) {

    lc= 0;



    coverage_try_clear();//这里的optimize是排序一下TSS
dp_netdev_pmd_try_optimize(pmd, poll_list, poll_cnt); 

    dp_netdev_pmd_hook_idle_run(pmd);

#ifdef ENABLE_EMCif (!ovsrcu_try_quiesce()) {

        emc_cache_slow_sweep(pmd->dp, &((pmd->flow_cache).emc_cache));

    }#elseovsrcu_try_quiesce();#endif

    for (i = 0; i < poll_cnt; i++) {

        uint64_t current_seq=netdev_get_change_seq(poll_list[i].rxq->port->netdev);if (poll_list[i].change_seq !=current_seq) {

            poll_list[i].change_seq=current_seq;

            poll_list[i].rxq_enabled=netdev_rxq_enabled(poll_list[i].rxq->rx);

        }

    }

}

1.5 承上启下：OVS的TSS算法

dpcls是megaflow的查询过程，使用TSS算法，是个很老的算法了，看源码之前，先讲一下ovs里面的TSS，之前内核已经讲过，但是没有讲OVS里做的优化，下边再说一次，然后建议再去看一下这个有很多图的博客
OVS-DPDK Datapath Classifier
，这样之后对整个dpcls流程就有所了解了。

TSS算法原理

OVS 在内核态使用了元组空间搜索算法（Tuple Space Search，简称 TSS）进行流表查找，
元组空间搜索算法的核心思想是，把所
有规则按照每个字段的前缀长度进行组合，并划分为不同的元组中，然后在这些元组集合中进行
哈希
查找
。我们举例说明，假设现有 10 条规则以及 3 个匹配字段，每个匹配字段长度均为 4：

我们将每条规则各匹配字段的前缀长度提取出来，按照前缀长度进行组合，并根据前缀长度组合进行分组：

我们将每个前缀长度组合称为
元组
，每个元组对应于哈希表的一个桶，同一前缀长度组合内的所有规则放置在同一个哈希桶内：

10 条规则被划分为 4 个元组，因此最多只需要四次查找，就可以找到对应的规则。

算法优缺点

为什么OVS选择TSS，而不选择其他查找算法？论文给出了以下三点解释：

（1）在虚拟化数据中心环境下，流的添加删除比较频繁，TSS支持高效的、常数时间的表项更新； （2）TSS支持任意匹配域的组合； （3）TSS存储空间随着流的数量线性增长，空间复杂度为 O(N)，N 为规则数目。

元组空间搜索算法的缺点是，由于基于哈希表实现，因此查找的时间复杂度不能确定。当所有规则各个字段的前缀长度组合数目过多时，查找性能会大大降低，最坏情况下需要查找所有规则。

OVS里做的排序优化

查找的过程需要从前向后遍历所有元组，命中了就不用往后查了。OVS给每个元组加了一个命中次数，命中次数越多，元组这个链表越靠前，这样就可以减少了查表次数。

2. dpcls查询

2.1 dpcls相关数据结构

//线程安全的
#define OVSRCU_TYPE(TYPE) struct { ATOMIC(TYPE) p; }

structcmap {

    OVSRCU_TYPE(struct cmap_impl *) impl;

};/*The implementation of a concurrent hash map.*/
structcmap_impl {//补齐64字节
PADDED_MEMBERS_CACHELINE_MARKER(CACHE_LINE_SIZE, cacheline0,

        unsignedint n;             /*Number of in-use elements.*/unsignedint max_n;         /*Max elements before enlarging.*/unsignedint min_n;         /*Min elements before shrinking.*/uint32_t mask;/*Number of 'buckets', minus one.*/uint32_t basis;/*Basis for rehashing client's

                                       hash values.*/);

    PADDED_MEMBERS_CACHELINE_MARKER(CACHE_LINE_SIZE, cacheline1,struct cmap_bucket buckets[1];

    );

};structcmap_bucket {/*Padding to make cmap_bucket exactly one cache line long.*/PADDED_MEMBERS(CACHE_LINE_SIZE,//锁机制，读和写都会+1，读的时候等到变成偶数再去读，保证安全
atomic_uint32_t counter;//桶中的每个槽用(hashs[i], nodes[i])元组来表示
uint32_t hashes[CMAP_K];structcmap_node nodes[CMAP_K];

    );

};structcmap_node {

    OVSRCU_TYPE(struct cmap_node *) next; /*Next node with same hash.*/};/*二级匹配表.每个报文接收端口对应一个*/
structdpcls {struct cmap_node node; /*链表节点*/odp_port_t in_port;/*报文接收端口*/
    struct cmap subtables_map; //管理下边subtables的索引，用于遍历
    struct pvector subtables;  //上文TSS算法所说的元组表
}

 structpvector {//指向具体子表信息
    OVSRCU_TYPE(struct pvector_impl *) impl;//平时,temp都是为NULL.只有当pvector扩充时,temp才用来临时缓存数据.//待排好序后,再拷贝到impl中,temp再置NULL
    struct pvector_impl *temp;

};//相当于vector<pvector_entry>
structpvector_impl {

    size_t size;/*Number of entries in the vector*/size_t allocated;/*Number allocted entries*/
    /*初始化的时候只有4个元素.后续可能会扩充*/
    structpvector_entry vector[];

}

 structpvector_entry {//pvector_impl中的vector是按照priority从小到大排序的//pmd_thread_main里会把priority赋值为hit_cnt，然后排序
        intpriority;/*实际指向了struct dpcls_subtable结构*/
        void *ptr;

}

 //子表信息
structdpcls_subtable {/*The fields are only used by writers.*/
    struct cmap_node cmap_node OVS_GUARDED; /*Within dpcls 'subtables_map'.*/

    struct cmap rules; //该表的bucket内容
    uint32_t hit_cnt;  //命中该子表的次数//下边是mask的miniflow前两个的bits里1的个数
uint8_t mf_bits_set_unit0; 

    uint8_t mf_bits_set_unit1;//根据mf_bits_set_unit01选择最合适的查找算法
dpcls_subtable_lookup_func lookup_func;/*Caches the masks to match a packet to, reducing runtime calculations.*/uint64_t*mf_masks; //由下边的mask->mf->bits[01]得来的，
    struct netdev_flow_key mask; //该表的掩码信息
};



关于上边的mf_masks与mask，举个例子

mf_bits_set_unit0= 4， mf_bits_set_unit1 = 0netdev_flow_key.mf.bits[0] = 666666010(2进制)

mf_masks= [1, 666666, 6666661, 66666611]  （2进制）

三个图对应他们的关系，链表三用于遍历的，查找过程中并不会通过链表三方式搜索。查找的时候走的就是链表二的流程。

2.2 dpcls查询入口：fast_path_processing->dpcls_lookup()

static booldpcls_lookup(struct dpcls *cls, const struct netdev_flow_key *keys[],struct dpcls_rule **rules, constsize_t cnt,int *num_lookups_p)

{#define MAP_BITS (sizeof(uint32_t) * CHAR_BIT)BUILD_ASSERT_DECL(MAP_BITS>=NETDEV_MAX_BURST);struct dpcls_subtable *subtable;

    uint32_t keys_map= TYPE_MAXIMUM(uint32_t); /*Set all bits.*/

    if (cnt !=MAP_BITS) {/*keys_map中置1位数为包的总数，并且第i位对应第i个包*/keys_map>>= MAP_BITS - cnt; /*Clear extra bits.*/}

    memset(rules,0, cnt * sizeof *rules);int lookups_match = 0, subtable_pos = 1;

    uint32_t found_map;



    PVECTOR_FOR_EACH (subtable,&cls->subtables) {//查找函数，对应下边的lookup_generic()
        found_map = subtable->lookup_func(subtable, keys_map, keys, rules);

        

        uint32_t pkts_matched=count_1bits(found_map);//搜索的子表个数，加上的是当前这几个key找了多少个表
        lookups_match += pkts_matched *subtable_pos;



        keys_map&= ~found_map;if (!keys_map) {if(num_lookups_p) {*num_lookups_p =lookups_match;

            }//全找到了
            return true;

        }

        subtable_pos++;

    }if(num_lookups_p) {*num_lookups_p =lookups_match;

    }//没有全找到
    return false;

}

lookup_generic()

ovs-dpdk里面有avx512-gather.c，使用avx512优化了look_up，整体逻辑还是一样的，这里只说dpif-netdev-lookup-generic

入口在这里，往下走,传进去subtable有效字段有多大staticuint32_t

dpcls_subtable_lookup_generic(struct dpcls_subtable *subtable,

                              uint32_t keys_map,const struct netdev_flow_key *keys[],struct dpcls_rule **rules)

{returnlookup_generic_impl(subtable, keys_map, keys, rules,

                               subtable->mf_bits_set_unit0,

                               subtable->mf_bits_set_unit1);

}staticinline uint32_t ALWAYS_INLINE

lookup_generic_impl(struct dpcls_subtable *subtable,      //当前的subtable
                    uint32_t keys_map,                    //miss_bit_map
                    const struct netdev_flow_key *keys[], //miss_key
                    struct dpcls_rule **rules,            //save hit_rule
                    constuint32_t bit_count_u0,constuint32_t bit_count_u1)

{//有几个包
    const uint32_t n_pkts =count_1bits(keys_map);

    ovs_assert(NETDEV_MAX_BURST>=n_pkts);

    uint32_t hashes[NETDEV_MAX_BURST];//根据mask字段的大小开空间
    const uint32_t bit_count_total = bit_count_u0 +bit_count_u1;//一个batch最大是NETDEV_MAX_BURST
    const uint32_t block_count_required = bit_count_total *NETDEV_MAX_BURST;

    uint64_t*mf_masks = subtable->mf_masks;inti;//申请存储一个batch报文信息的数组，存放
    uint64_t *blocks_scratch =get_blocks_scratch(block_count_required);//获得每个key与当前表的mask“与运算”的结果
ULLONG_FOR_EACH_1 (i, keys_map) {

            netdev_flow_key_flatten(keys[i],&subtable->mask, //该表的掩码信息
                                    mf_masks,  //由subtable->mask处理后的mask
                                    &blocks_scratch[i *bit_count_total],

                                    bit_count_u0,

                                    bit_count_u1);

    }//算出来每一个key在该subtable里的hash值，该hash值由“mask字节数，key和mask与运算结果”得出
ULLONG_FOR_EACH_1 (i, keys_map) {

        uint64_t*block_ptr = &blocks_scratch[i *bit_count_total];

        uint32_t hash= hash_add_words64(0, block_ptr, bit_count_total); 

        hashes[i]= hash_finish(hash, bit_count_total * 8); 

    }



    uint32_t found_map;const struct cmap_node *nodes[NETDEV_MAX_BURST];//找到每个key在该subtable里的cmap，并且返回每个key有没有被找到，第i位是1则找到
    found_map = cmap_find_batch(&subtable->rules, keys_map, hashes, nodes);



    ULLONG_FOR_EACH_1 (i, found_map) {struct dpcls_rule *rule;//可能不同的rule有相同的hash，看那个是匹配的
CMAP_NODE_FOR_EACH (rule, cmap_node, nodes[i]) {const uint32_t cidx = i *bit_count_total;/*rule->mask & keys[i]的值与rule->flow相比较*/uint32_t match=netdev_rule_matches_key(rule, bit_count_total,&blocks_scratch[cidx]);if(OVS_LIKELY(match)) {

                rules[i]=rule;

                subtable->hit_cnt++;gotonext;

            }

        }

        ULLONG_SET0(found_map, i);/*Did not match.*/next:

        ;/*Keep Sparse happy.*/}returnfound_map;

}

掩码运算netdev_flow_key_flatten()

//这个函数对应dpif-netdev.c里面的dpcls_flow_key_gen_masks()
static inline voidnetdev_flow_key_flatten(const struct netdev_flow_key *key,  //要查找的miss_key
                        const struct netdev_flow_key *mask,const uint64_t *mf_masks,

                        uint64_t*blocks_scratch,constuint32_t u0_count,constuint32_t u1_count)

{/*Load mask from subtable, mask with packet mf, popcount to get idx.*/
    const uint64_t *pkt_blocks = miniflow_get_values(&key->mf);const uint64_t *tbl_blocks = miniflow_get_values(&mask->mf); //获取miss_key和mask的miniflow

    /*Packet miniflow bits to be masked by pre-calculated mf_masks.*/
    const uint64_t pkt_bits_u0 = key->mf.map.bits[0];const uint32_t pkt_bits_u0_pop =count_1bits(pkt_bits_u0);const uint64_t pkt_bits_u1 = key->mf.map.bits[1];//这个函数就是把miss_key与subtable的掩码进行&运算//会运算出该mask在意字段结果，放到blocks_scratch里
    netdev_flow_key_flatten_unit(&pkt_blocks[0],   //key-mf的数据段
                                 &tbl_blocks[0],   //mask->mf的数据段
                                 &mf_masks[0],     //mask->mf->bits得来mask
                                 &blocks_scratch[0], //存放的地址
                                 pkt_bits_u0,      //key->mf里的bits[0]
                                 u0_count);        //mask->mf->bits[0]里1的个数
netdev_flow_key_flatten_unit(&pkt_blocks[pkt_bits_u0_pop], //上边bits[0]的已经算过了，从bits[1]开始算
                                 &tbl_blocks[u0_count],&mf_masks[u0_count],&blocks_scratch[u0_count],

                                 pkt_bits_u1,

                                 u1_count);

}static inline voidnetdev_flow_key_flatten_unit(const uint64_t *pkt_blocks, //key-mf的数据段
                             const uint64_t *tbl_blocks, //mask->mf里的数据段
                             const uint64_t *mf_masks,   //mask->mf->bits得来mask
                             uint64_t *blocks_scratch,   //存放到这里
                             const uint64_t pkt_mf_bits, //key->mf里的bits[01]
                             const uint32_t count)       //mask->mf->bits[0]里1的个数
{//说一下意思，这个我们流程就是用key和subtable的mask与运算，肯定只需要与运算mask里//不为0的字段，其他的mask不关心，然后这个操作就是为了得到key对应字段是key->mf的第几位，//比如mask的bits[0]=66666611, key的bits[0] = 10100, mask里的第3个1在key里面是第1个//这一位与的结果就是tbl_blocks[2]&pkt_blocks[0], 也就是怎么找到key里的下标0//就看key当前位之前有几个1就行了。这里这样做的1010666666，//蓝色1之前有count_1bits(1010666666 & 0006666661) = 3//对上边的mask举个例子 count = 4;//mask->mf->bits[0] = 666666010 (2进制)//mf_masks = [1, 666666, 6666661, 66666611] (2进制);//pkt_mf_bits = 010100//blocks_scratch = [0,0,0,0,pkt_blocks[1]&tbl_blocks[4],0]
uint32_t i;for (i = 0; i < count; i++) {//拿i=2举例
        uint64_t mf_mask = mf_masks[i];             //mf_mask = 006666661
        uint64_t idx_bits = mf_mask & pkt_mf_bits;  //idx_bits = 000100
        const uint32_t pkt_idx = count_1bits(idx_bits); //pkt_idx = 1
uint64_t pkt_has_mf_bit= (mf_mask + 1) & pkt_mf_bits;  //pkt_has_mf_bit = 010000//是否求掩码：mask当前位对应的key的字段，如果key在当前位是0，下边算掩码就会变成0
        uint64_t no_bit = ((!pkt_has_mf_bit) > 0) - 1; //2^64 - 1//mask里第i个字段与运算key对应的字段
        blocks_scratch[i] = pkt_blocks[pkt_idx] & tbl_blocks[i] & no_bit; // 
}

}

key对应的cmap：cmap_find_batch()

unsigned longcmap_find_batch(const struct cmap *cmap, unsigned longmap,

                uint32_t hashes[],const struct cmap_node *nodes[])

{const struct cmap_impl *impl =cmap_get_impl(cmap); 

    unsignedlong result =map;inti;//每一位就是一个包，一字节8个包
    uint32_t h1s[sizeof map *CHAR_BIT];const struct cmap_bucket *b1s[sizeof map *CHAR_BIT];const struct cmap_bucket *b2s[sizeof map *CHAR_BIT];

    uint32_t c1s[sizeof map *CHAR_BIT];//每个impl里桶的数量为impl->mask+1//为什么mask是桶的个数减1:因为下标从0开始，找下表的时候直接(hash & impl->mask)就行了//至于为什么开两个？因为buckets存放的方法也是一个值对应两个hash//第一次hash1 = rehash(impl->basis, hash), 找buckets[hash1 & impl->mask], 遍历里面CMAP_K个元素//第二次hash2 = other_hash(hash1), 找buckets[hash2 & impl->mask], 遍历里面CMAP_K个元素
    

    /*Compute hashes and prefetch 1st buckets.*/ULLONG_FOR_EACH_1(i, map) {

        h1s[i]=rehash(impl, hashes[i]);            

        b1s[i]= &impl->buckets[h1s[i] & impl->mask];

        OVS_PREFETCH(b1s[i]);

    }/*Lookups, Round 1. Only look up at the first bucket.*/ULLONG_FOR_EACH_1(i, map) {

        uint32_t c1;const struct cmap_bucket *b1 =b1s[i];const struct cmap_node *node;do{

            c1=read_even_counter(b1);//找一下这个cmap_bucket里面有没有相同hash的
            node =cmap_find_in_bucket(b1, hashes[i]);

        }while(OVS_UNLIKELY(counter_changed(b1, c1)));if (!node) {/*Not found (yet); Prefetch the 2nd bucket.*/b2s[i]= &impl->buckets[other_hash(h1s[i]) & impl->mask];

            OVS_PREFETCH(b2s[i]);

            c1s[i]= c1; /*We may need to check this after Round 2.*/
            continue;

        }/*Found.*/ULLONG_SET0(map, i);/*Ignore this on round 2.*/OVS_PREFETCH(node);

        nodes[i]=node;

    }/*Round 2. Look into the 2nd bucket, if needed.*/ULLONG_FOR_EACH_1(i, map) {

        uint32_t c2;const struct cmap_bucket *b2 =b2s[i];const struct cmap_node *node;do{

            c2=read_even_counter(b2);

            node=cmap_find_in_bucket(b2, hashes[i]);

        }while(OVS_UNLIKELY(counter_changed(b2, c2)));if (!node) {//可能被修改了，
            if(OVS_UNLIKELY(counter_changed(b1s[i], c1s[i]))) {

                node=cmap_find__(b1s[i], b2s[i], hashes[i]);if(node) {gotofound;

                }

            }/*Not found.*/ULLONG_SET0(result, i);/*Fix the result.*/
            continue;

        }

found:

        OVS_PREFETCH(node);

        nodes[i]=node;

    }returnresult;

}

2.3 fast_path_processing()

static inline voidfast_path_processing(struct dp_netdev_pmd_thread *pmd,struct dp_packet_batch *packets_,struct netdev_flow_key **keys,struct dp_packet_flow_map *flow_map,

                     uint8_t*index_map,

                     odp_port_t in_port)

{const size_t cnt =dp_packet_batch_size(packets_);#if !defined(__CHECKER__) && !defined(_WIN32)
    const size_t PKT_ARRAY_SIZE =cnt;#else
    /*Sparse or MSVC doesn't like variable length array.*/
    enum { PKT_ARRAY_SIZE =NETDEV_MAX_BURST };#endif
    struct dp_packet *packet;struct dpcls *cls;struct dpcls_rule *rules[PKT_ARRAY_SIZE];struct dp_netdev *dp = pmd->dp;int upcall_ok_cnt = 0, upcall_fail_cnt = 0;int lookup_cnt = 0, add_lookup_cnt;boolany_miss;for (size_t i = 0; i < cnt; i++) {/*Key length is needed in all the cases, hash computed on demand.*/keys[i]->len = netdev_flow_key_size(miniflow_n_values(&keys[i]->mf));

    }/*Get the classifier for the in_port*/
    //找到端口对应的dpcls结构，每个port有自己的dpcls，因为每个port收到的报文会更相似
    cls =dp_netdev_pmd_lookup_dpcls(pmd, in_port);if(OVS_LIKELY(cls)) {//调用dpcls_lookup进行匹配
        any_miss = !dpcls_lookup(cls, (const struct netdev_flow_key **)keys,

                                rules, cnt,&lookup_cnt);

    }else{

        any_miss= true;

        memset(rules,0, sizeof(rules));

    }//如果有miss的，则需要进行openflow流表查询
    if (OVS_UNLIKELY(any_miss) && !fat_rwlock_tryrdlock(&dp->upcall_rwlock)) {

        uint64_t actions_stub[512 / 8], slow_stub[512 / 8];structofpbuf actions, put_actions;



        ofpbuf_use_stub(&actions, actions_stub, sizeofactions_stub);

        ofpbuf_use_stub(&put_actions, slow_stub, sizeofslow_stub);



        DP_PACKET_BATCH_FOR_EACH (i, packet, packets_) {struct dp_netdev_flow *netdev_flow;if(OVS_LIKELY(rules[i])) {continue;

            }//此时可能已经更新了，在进入upcall之前如果再查一次，如果能够查到，会比upcall消耗的少得多
            netdev_flow =dp_netdev_pmd_lookup_flow(pmd, keys[i],&add_lookup_cnt);if(netdev_flow) {

                lookup_cnt+=add_lookup_cnt;

                rules[i]= &netdev_flow->cr;continue;

            }//第一级和第二级流表查找失败后，就要查找第三级流表了，即openflow流表，这也称为upcall调用。//在普通ovs下是通过netlink实现的，在ovs+dpdk下，直接在pmd线程中调用upcall_cb即可。//开始查找openflow流表。如果查找openflow流表成功并需要下发到dpcls时，需要判断是否超出最大流表限制
            int error =handle_packet_upcall(pmd, packet, keys[i],&actions, &put_actions);if(OVS_UNLIKELY(error)) {

                upcall_fail_cnt++;

            }else{

                upcall_ok_cnt++;

            }

        }



        ofpbuf_uninit(&actions);

        ofpbuf_uninit(&put_actions);

        fat_rwlock_unlock(&dp->upcall_rwlock);

    }else if(OVS_UNLIKELY(any_miss)) {

        DP_PACKET_BATCH_FOR_EACH (i, packet, packets_) {if (OVS_UNLIKELY(!rules[i])) {

                dp_packet_delete(packet);

                COVERAGE_INC(datapath_drop_lock_error);

                upcall_fail_cnt++;

            }

        }

    }



    DP_PACKET_BATCH_FOR_EACH (i, packet, packets_) {struct dp_netdev_flow *flow;/*Get the original order of this packet in received batch.*/
        int recv_idx =index_map[i];

        uint16_t tcp_flags;if (OVS_UNLIKELY(!rules[i])) {continue;

        }



        flow=dp_netdev_flow_cast(rules[i]);bool hook_cached = false;if (pmd->cached_hook &&\

                pmd->cached_hook_pmd &&\

                pmd->cached_hook->hook_flow_miss) {

            hook_cached= pmd->cached_hook->hook_flow_miss(pmd->cached_hook_pmd, packet, flow);

        }if (!hook_cached) {boolsmc_enable_db;

            atomic_read_relaxed(&pmd->dp->smc_enable_db, &smc_enable_db);//查找到了packet，如果开启了smc，更新smc
            if(smc_enable_db) {

                uint32_t hash=  dp_netdev_flow_hash(&flow->ufid);

                smc_insert(pmd, keys[i], hash);

            }//查到了packet，看是否写会更新上一层cache（EMC）
            if(emc_probabilistic_insert(pmd, keys[i], flow)) {if (flow->status ==OFFLOAD_NONE) {

                    queue_netdev_flow_put(pmd->dp->dp_flow_offload, \

                            pmd->dp->class, \

                            flow, NULL, DP_NETDEV_FLOW_OFFLOAD_OP_ADD);

                }

            }

        }/*Add these packets into the flow map in the same order

         * as received.*/tcp_flags= miniflow_get_tcp_flags(&keys[i]->mf);

        packet_enqueue_to_flow_map(packet, flow, tcp_flags,

                                   flow_map, recv_idx);

    }//更新各个信息
    pmd_perf_update_counter(&pmd->perf_stats, PMD_STAT_MASKED_HIT,

                            cnt- upcall_ok_cnt -upcall_fail_cnt);

    pmd_perf_update_counter(&pmd->perf_stats, PMD_STAT_MASKED_LOOKUP,

                            lookup_cnt);

    pmd_perf_update_counter(&pmd->perf_stats, PMD_STAT_MISS,

                            upcall_ok_cnt);

    pmd_perf_update_counter(&pmd->perf_stats, PMD_STAT_LOST,

                            upcall_fail_cnt);

}

2.4 smc更新smc_insert()

static inline voidsmc_insert(struct dp_netdev_pmd_thread *pmd,const struct netdev_flow_key *key,

           uint32_t hash)

{struct smc_cache *smc_cache = &(pmd->flow_cache).smc_cache;struct smc_bucket *bucket = &smc_cache->buckets[key->hash &SMC_MASK];

    uint16_t index;

    uint32_t cmap_index;inti;//布谷鸟算法
    cmap_index = cmap_find_index(&pmd->flow_table, hash);

    index= (cmap_index >= UINT16_MAX) ?UINT16_MAX : (uint16_t)cmap_index;/*If the index is larger than SMC can handle (uint16_t), we don't

     * insert*/
    if (index ==UINT16_MAX) {//表明找到了
        return;

    }/*If an entry with same signature already exists, update the index*/uint16_t sig= key->hash >> 16;for (i = 0; i < SMC_ENTRY_PER_BUCKET; i++) {if (bucket->sig[i] ==sig) {

            bucket->flow_idx[i] =index;return;

        }

    }/*If there is an empty entry, occupy it.*/
    for (i = 0; i < SMC_ENTRY_PER_BUCKET; i++) {if (bucket->flow_idx[i] ==UINT16_MAX) {

            bucket->sig[i] =sig;

            bucket->flow_idx[i] =index;return;

        }

    }/*Otherwise, pick a random entry.*/i= random_uint32() %SMC_ENTRY_PER_BUCKET;

    bucket->sig[i] =sig;

    bucket->flow_idx[i] =index;

}

3. upcall到openflow查找，然后更新dpcls

这里就不讲具体代码了，讲一下大概：到openflow查找后会更新dpcls，执行dp_netdev_flow_add() --> dpcls_insert() -->
dpcls_find_subtable()
--> cmap_insert()

dpcls_find_subtable()：

找一下是否存在相同mask的subtable，存在返回这个subtable，不存在就创建一个subtable，创建的时候会调用dpcls_create_subtable，里面有个dpcls_subtable_get_best_impl会根据mask的miniflow的bits[0]和bits[1]选择的查找算法。

cmap_insert里hash算法用的就是布谷鸟hash，hash两次，插入的核心代码：

static boolcmap_try_insert(struct cmap_impl *impl, struct cmap_node *node, uint32_t hash)

{

    uint32_t h1=rehash(impl, hash);

    uint32_t h2=other_hash(h1);//hash两次找到两个桶
    struct cmap_bucket *b1 = &impl->buckets[h1 & impl->mask];struct cmap_bucket *b2 = &impl->buckets[h2 & impl->mask];//插入规则：//1.是否有相同hash的node，就插到对应链上//2.没有相同hash，就看有没有空的node//3.都不行就通过bfs，看能否让b1,b2空出来一个，把这个放进去//都不行就插入失败
    return (OVS_UNLIKELY(cmap_insert_dup(node, hash, b1) ||cmap_insert_dup(node, hash, b2))||OVS_LIKELY(cmap_insert_bucket(node, hash, b1)||cmap_insert_bucket(node, hash, b2))||cmap_insert_bfs(impl, node, hash, b1, b2)); 

}

参考博客：

OVS-DPDK Datapath Classifier
：这个是理论上的流程，看完就知道这个算法流程了

ovs分类器 flow和miniflow
：很重要的结构体miniflow

OVS-DPDK DataPath Classifier反向设计
：这个有很多详细的解释，但不怎么流畅

什么是jsRPC

说实话在介绍 JSRPC 我向大家推荐一个库 Selenium-wire 感觉和JSrpc的原理很像
RPC指远程过程调用，APP里面的RPC大家比较熟悉了。
那什么是jsRPC，顾名思义，就是指利用js手段实现代码远程控制浏览器控制台的一种手段，需要借助websocket实现。
下面我们先简单的写一个小例子

var ws = require('nodejs-websocket');
console.log('开始建立连接...');

var cached = {}

var server = ws.createServer(function (conn) {
  conn.on('text', function (msg) {
    var key = conn.key
    console.log("将要发送代码：【" + msg + "】至浏览器/python");
    server.connections.forEach(conn => {
      conn.send(msg);
    })
  })
  conn.on('close', function (code, reason) {
    console.log('关闭连接', code, reason)
  });
  conn.on('error', function (code, reason) {
    console.log('异常关闭', code, reason)
  });
}).listen(8015)
console.log('WebSocket建立完毕');

python

# -*- coding: utf-8 -*-
# @Time    : 2022/12/2 14:17
# @Author  : lzc
# @Email   : hybpjx@163.com
# @Software: PyCharm
# @blog    : https://www.cnblogs.com/zichliang

import websocket

ws = websocket.WebSocketApp("ws://127.0.0.1:8015")

def on_message(ws,message):
    if message.split("_")[0]!= 'js':
        print(message)


ws.on_message = on_message
ws.run_forever()

在浏览器中注入

(function () {
    if (window.WebSocket) {
        ws = new WebSocket("ws://localhost:8015");
        ws.onopen = function (e) {
        };
        ws.onclose = function (e) {
            console.log("server closed", e)
        }
        ws.onerror = function (e) {
            "connection error"
        }
        ws.onmessage = function (e) {
            console.log(e)
        }
    }
})()

在浏览器中对其进行操作
比如发送什么数据

在python代码中就能接收到这些数据

基于sekiro的isRPC应用

简介: sekiro是基于长链接和代码注入的Android private API暴露框架

sekiro在安卓方面的应用请看课程 第六章
github:
https://github.com/virjar/sekiro
作者:邓维佳 virjar 渣总[都是一个人]
sekiro 在 web方向，使用 websocket协议注入
官方文档内容:
Sekiro 已支持 websocket 协议，使用本功能可以支持注入js 到浏览器后，调用浏览器环境的is 代码
Web 环境基于 WebSocket 实现
文档:
https://sekiro.iinti.cn/sekiro-doc/01_user_manual/1.quickstart.html
下载地址：
Go HTTP File Server (iinti.cn)

下载这个 到本地
并且下载好 JDK java的编译器

然后运行响应版本
注意：

使用

使用方法：将js文件
sekiroWebMini.js
注入到浏览器环境 ,
之后通过当前接口访问自己的浏览器服务:

ws://127.0.0.1:5612

然后注入我们的代码

var client = new SekiroClient("ws://127.0.0.1:5612/business/register?group=demo-ws&clientId="+Math.random());
client.registerAction("clientTime",function(request, resolve,reject ){
            resolve(""+new Date());
})

请注意，Sekiro JS RPC是Sekiro的扩展，依然需要遵循Sekiro的各种规则定义。

注入完成之后 使用我们的浏览器访问
http://127.0.0.1:5612/business/invoke?group=demo-ws&action=clientTime

必须与我们注入的代码相对应 不然就会失败 成功之后是这样的

可以看到 这里返回的就是我们resolve(""+new Date()); 这段代码 也就意味着可以进行任何的操作。

对于sekiro jsRPC的理解

提供了一种浏览器与代码以及接口交互的一种可行性理论上只要是网站都可以通过这种方案在is层面突破反爬限制，且稳定性要优于目前的一切自动化框架

使用描述:
可能性1:针对解密函数的接口暴露
可能性2:针对数据接口的接口暴露
可能性3:针对自动化采集的接口暴露
可能性4:其他的一切想你所想....

优势

sekiro驱动jsRPC的优势与缺陷分析

1. 与传统自动化框架相比，稳定性更高且不需要考虑浏览器指纹。操作得当的话，性能不在一个数量级
2. 可以通过js直接获取加密参数，对于全局变量的参数暴露具有极佳的体验。[一剑破光阴]
3. 可以直接获取接口数据，对于获取数据接口暴露明显的网站具有极佳的体验。
4. 不考虑风控的情况下，能达到高并发。

缺陷

1. 内置不可更改的浏览器环境不可更改，对于风控突破而言是个极大的隐患[浏览器沙箱可能会解决这个问题]
2. 需要对js有较深理解，需要寻找合适的注入时机和注入位置[这个算缺陷么? 菜是原罪]。
3. 需要注入脚本，所以要依托于油猴脚本/autoResponse/手动注入[如果使用抓包工具需要注意端口问题]
4. 对于全局js混淆、风控严格、is参数分布散乱、鼠标轨迹指纹、变量作用域复杂的网址的支持较差。
5. 需要保持浏览器开启，所以多少会占用些资源。
6. 如果不取参数，而是只取接口内容 或者 参数与IP绑定的加密，对代理池接入的支持不佳。