Storm的分组策略

柔情只为你懂 2022-12-28 00:34 172阅读 0赞

# **1、分组策略（Stream Grouping）** #

### stream grouping用来定义一个stream应该如何分配给Bolts上面的多个Executors（多线程、多并发）。 ###

Storm里面有7种类型的stream grouping

**1****）Shuffle Grouping: 随机分组**，轮询，平均分配。随机派发stream里面的tuple，保证每个bolt接收到的tuple数目大致相同。

**2****）Fields Grouping：按字段分组**，比如按userid来分组，具有同样userid的tuple会被分到相同的Bolts里的一个task，而不同的userid则会被分配到不同的bolts里的task。

**3****）All Grouping：广播发送**，对于每一个tuple，所有的bolts都会收到。

**4****）Global Grouping：全局分组**，这个tuple被分配到storm中的一个bolt的其中一个task。再具体一点就是分配给id值最低的那个task。

**5****）Non Grouping：不分组**，这stream grouping个分组的意思是说stream不关心到底谁会收到它的tuple。**目前这种分组和Shuffle grouping是一样的效果。在多线程情况****下不平均分配****。**

**6****）Direct Grouping：直接分组**，这是一种比较特别的分组方法，用这种分组意味着消息的发送者指定由消息接收者的哪个task处理这个消息。只有被声明为Direct Stream的消息流可以声明这种分组方法。而且这种消息tuple必须使用emitDirect方法来发射。消息处理者可以通过TopologyContext来获取处理它的消息的task的id （OutputCollector.emit方法也会返回task的id）。

**7****）Local or shuffle grouping：**如果目标bolt有一个或者多个task在同一个工作进程中，tuple将会被随机发送给这些tasks。否则，和普通的Shuffle Grouping行为一致。

# 2、测试 #

### **2.1、 Shuffle Grouping** ###

**ProduceWebLog.java：**

package storm.util;
    
    import org.apache.storm.shade.io.netty.buffer.ByteBuf;
    
    import java.io.File;
    import java.io.FileOutputStream;
    import java.io.IOException;
    import java.nio.ByteBuffer;
    import java.nio.channels.FileChannel;
    import java.util.Random;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/12 15:36
     * @Version: 1.0
     * @Description:
     */
    public class ProduceWebLog {
        public static void main(String[] args) throws Exception {
    
            String logFilePath = "D:/mine/weblog.txt";
            FileOutputStream logFileOutPutStream = new FileOutputStream(logFilePath);
            // 1 网站名称
            String[] hosts = { "ljxwtl.cn" , "miaov.com"};
            // 2 会话id
            String[] sessionId = { "ABYH6Y4V4SCVXTG6DPB4VH9U123", "XXYH6YCGFJYERTT834R52FDXV9U34",
                    "BBYH61456FGHHJ7JL89RG5VV9UYU7", "CYYH6Y2345GHI899OFG4V9U567", "VVVYH6Y4V4SFXZ56JIPDPB4V678" };
            // 3 访问网站时间
            String[] time = { "2017-08-07 08:40:50", "2017-08-07 08:40:51", "2017-08-07 08:40:52", "2017-08-07 08:40:53",
                    "2017-08-07 09:40:49", "2017-08-07 10:40:49", "2017-08-07 11:40:49", "2017-08-07 12:40:49" };
    
            // 4 拼接网站访问日志
            StringBuilder stringBuilder = new StringBuilder();
            Random random = new Random();
            for (int i = 0; i < 30; i++) {
                stringBuilder.append(hosts[random.nextInt(hosts.length)]).append("\t").append(sessionId[random.nextInt(sessionId.length)]).append("\t").append(time[random.nextInt(time.length)]).append("\n");
            }
    
            File logFile = new File(logFilePath);
            if (!logFile.exists()){
                logFile.createNewFile();
            }
    
            FileChannel fileChannel = logFileOutPutStream.getChannel();
            ByteBuffer byteBuffer = ByteBuffer.allocateDirect(1024 * 1024);
    
            byteBuffer.put(stringBuilder.toString().getBytes());
            byteBuffer.flip();
            fileChannel.write(byteBuffer);
            byteBuffer.clear();
    
            fileChannel.close();
            logFileOutPutStream.close();
        }
    }

**WebSpout.java：**

package storm;
    
    import lombok.SneakyThrows;
    import org.apache.storm.spout.SpoutOutputCollector;
    import org.apache.storm.task.TopologyContext;
    import org.apache.storm.topology.IRichSpout;
    import org.apache.storm.topology.OutputFieldsDeclarer;
    import org.apache.storm.tuple.Fields;
    import org.apache.storm.tuple.Values;
    
    import java.io.BufferedReader;
    import java.io.FileReader;
    import java.util.Map;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/12 15:33
     * @Version: 1.0
     * @Description:
     */
    public class WebSpout implements IRichSpout {
    
        private TopologyContext topologyContext;
        private SpoutOutputCollector spoutOutputCollector;
        private BufferedReader reader;
    
        @SneakyThrows
        @Override
        public void open(Map<String, Object> map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
            this.topologyContext = topologyContext;
            this.spoutOutputCollector = spoutOutputCollector;
            reader = new BufferedReader(new FileReader("D:/mine/weblog.txt"));
        }
    
        @Override
        public void close() {
    
        }
    
        @Override
        public void activate() {
    
        }
    
        @Override
        public void deactivate() {
    
        }
    
        @SneakyThrows
        @Override
        public void nextTuple() {
            String line = null;
            while ((line = reader.readLine())!=null){
                spoutOutputCollector.emit(new Values(line));
                TimeUnit.SECONDS.sleep(1);
            }
        }
    
        @Override
        public void ack(Object o) {
    
        }
    
        @Override
        public void fail(Object o) {
    
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
            outputFieldsDeclarer.declare(new Fields("line"));
        }
    
        @Override
        public Map<String, Object> getComponentConfiguration() {
            return null;
        }
    }

**WebSlot.java：**

package storm;
    
    import org.apache.storm.task.OutputCollector;
    import org.apache.storm.task.TopologyContext;
    import org.apache.storm.topology.IRichBolt;
    import org.apache.storm.topology.OutputFieldsDeclarer;
    import org.apache.storm.tuple.Tuple;
    import storm.util.OSUtil;
    
    import java.util.Map;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/12 15:58
     * @Version: 1.0
     * @Description:
     */
    public class WebSlot implements IRichBolt {
        private TopologyContext topologyContext;
        private OutputCollector outputCollector;
    
        private int lineNumber = 0;
        @Override
        public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {
            this.topologyContext = topologyContext;
            this.outputCollector = outputCollector;
        }
    
        @Override
        public void execute(Tuple tuple) {
            String line = tuple.getStringByField("line");
            String sessionId = line.split("\t")[1];
            lineNumber ++;
    
            OSUtil.sendMessage(this,line + "" + lineNumber);
        }
    
        @Override
        public void cleanup() {
    
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    
        }
    
        @Override
        public Map<String, Object> getComponentConfiguration() {
            return null;
        }
    }

**WebLogTopology.java： **

package storm;
    
    import org.apache.storm.Config;
    import org.apache.storm.LocalCluster;
    import org.apache.storm.generated.StormTopology;
    import org.apache.storm.thrift.TException;
    import org.apache.storm.topology.TopologyBuilder;
    import org.apache.storm.tuple.Fields;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/12 16:07
     * @Version: 1.0
     * @Description:
     */
    public class WebLogTopology {
        public static void main(String[] args) throws Exception {
            TopologyBuilder topologyBuilder = new TopologyBuilder();
            topologyBuilder.setSpout("webLogSpout",new WebSpout());
            topologyBuilder.setBolt("webLogSlot",new WebSlot(),2).shuffleGrouping("webLogSpout");
    
            //创建任务
            StormTopology job = topologyBuilder.createTopology();
    
            Config config = new Config();
    
            //运行任务有两种模式
            //1 本地模式   2 集群模式
    
            //1、本地模式
            LocalCluster localCluster = new LocalCluster();
            localCluster.submitTopology("MyWebLog",config,job);
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70][]

### **2.1、 Fields Grouping** ###

**WebSpout.java：**

package storm;
    
    import lombok.SneakyThrows;
    import org.apache.storm.spout.SpoutOutputCollector;
    import org.apache.storm.task.TopologyContext;
    import org.apache.storm.topology.IRichSpout;
    import org.apache.storm.topology.OutputFieldsDeclarer;
    import org.apache.storm.tuple.Fields;
    import org.apache.storm.tuple.Values;
    
    import java.io.BufferedReader;
    import java.io.FileReader;
    import java.util.Map;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/12 15:33
     * @Version: 1.0
     * @Description:
     */
    public class WebSpout implements IRichSpout {
    
        private TopologyContext topologyContext;
        private SpoutOutputCollector spoutOutputCollector;
        private BufferedReader reader;
    
        @SneakyThrows
        @Override
        public void open(Map<String, Object> map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
            this.topologyContext = topologyContext;
            this.spoutOutputCollector = spoutOutputCollector;
            reader = new BufferedReader(new FileReader("D:/mine/weblog.txt"));
        }
    
        @Override
        public void close() {
    
        }
    
        @Override
        public void activate() {
    
        }
    
        @Override
        public void deactivate() {
    
        }
    
        @SneakyThrows
        @Override
        public void nextTuple() {
            String line = null;
            while ((line = reader.readLine())!=null){
                String[] splits = line.split("\t");
                spoutOutputCollector.emit(new Values(splits[0],line));
                TimeUnit.SECONDS.sleep(1);
            }
        }
    
        @Override
        public void ack(Object o) {
    
        }
    
        @Override
        public void fail(Object o) {
    
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
            outputFieldsDeclarer.declare(new Fields("domain","line"));
        }
    
        @Override
        public Map<String, Object> getComponentConfiguration() {
            return null;
        }
    }

**WebSlot.java：**

package storm;
    
    import org.apache.storm.task.OutputCollector;
    import org.apache.storm.task.TopologyContext;
    import org.apache.storm.topology.IRichBolt;
    import org.apache.storm.topology.OutputFieldsDeclarer;
    import org.apache.storm.tuple.Tuple;
    import storm.util.OSUtil;
    
    import java.util.Map;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/12 15:58
     * @Version: 1.0
     * @Description:
     */
    public class WebSlot implements IRichBolt {
        private TopologyContext topologyContext;
        private OutputCollector outputCollector;
    
        private int lineNumber = 0;
        @Override
        public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {
            this.topologyContext = topologyContext;
            this.outputCollector = outputCollector;
        }
    
        @Override
        public void execute(Tuple tuple) {
            String domain = tuple.getStringByField("domain");
            String line = tuple.getStringByField("line");
            String sessionId = line.split("\t")[1];
            lineNumber ++;
    
            OSUtil.sendMessage(this,domain + "======" + lineNumber);
        }
    
        @Override
        public void cleanup() {
    
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    
        }
    
        @Override
        public Map<String, Object> getComponentConfiguration() {
            return null;
        }
    }

**WebLogTopology.java： **

package storm;
    
    import org.apache.storm.Config;
    import org.apache.storm.LocalCluster;
    import org.apache.storm.generated.StormTopology;
    import org.apache.storm.thrift.TException;
    import org.apache.storm.topology.TopologyBuilder;
    import org.apache.storm.tuple.Fields;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/12 16:07
     * @Version: 1.0
     * @Description:
     */
    public class WebLogTopology {
        public static void main(String[] args) throws Exception {
            TopologyBuilder topologyBuilder = new TopologyBuilder();
            topologyBuilder.setSpout("webLogSpout",new WebSpout());
            topologyBuilder.setBolt("webLogSlot",new WebSlot(),2).fieldsGrouping("webLogSpout",new Fields("domain"));
    
            //创建任务
            StormTopology job = topologyBuilder.createTopology();
    
            Config config = new Config();
    
            //运行任务有两种模式
            //1 本地模式   2 集群模式
    
            //1、本地模式
            LocalCluster localCluster = new LocalCluster();
            localCluster.submitTopology("MyWebLog",config,job);
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 1][]

### **2.3、** **All Grouping** ###

package storm;
    
    import org.apache.storm.Config;
    import org.apache.storm.LocalCluster;
    import org.apache.storm.generated.StormTopology;
    import org.apache.storm.thrift.TException;
    import org.apache.storm.topology.TopologyBuilder;
    import org.apache.storm.tuple.Fields;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/12 16:07
     * @Version: 1.0
     * @Description:
     */
    public class WebLogTopology {
        public static void main(String[] args) throws Exception {
            TopologyBuilder topologyBuilder = new TopologyBuilder();
            topologyBuilder.setSpout("webLogSpout",new WebSpout());
            topologyBuilder.setBolt("webLogSlot",new WebSlot(),3).allGrouping("webLogSpout");
    
            //创建任务
            StormTopology job = topologyBuilder.createTopology();
    
            Config config = new Config();
    
            //运行任务有两种模式
            //1 本地模式   2 集群模式
    
            //1、本地模式
            LocalCluster localCluster = new LocalCluster();
            localCluster.submitTopology("MyWebLog",config,job);
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 2][]

### **2.4、** **Global Grouping** ###

package storm;
    
    import org.apache.storm.Config;
    import org.apache.storm.LocalCluster;
    import org.apache.storm.generated.StormTopology;
    import org.apache.storm.thrift.TException;
    import org.apache.storm.topology.TopologyBuilder;
    import org.apache.storm.tuple.Fields;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/12 16:07
     * @Version: 1.0
     * @Description:
     */
    public class WebLogTopology {
        public static void main(String[] args) throws Exception {
            TopologyBuilder topologyBuilder = new TopologyBuilder();
            topologyBuilder.setSpout("webLogSpout",new WebSpout());
            topologyBuilder.setBolt("webLogSlot",new WebSlot(),3).globalGrouping("webLogSpout");
    
            //创建任务
            StormTopology job = topologyBuilder.createTopology();
    
            Config config = new Config();
    
            //运行任务有两种模式
            //1 本地模式   2 集群模式
    
            //1、本地模式
            LocalCluster localCluster = new LocalCluster();
            localCluster.submitTopology("MyWebLog",config,job);
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 3][]

### **2.5、 Direct Grouping** ###

**要使用Direct Grouping，必须在emit使用emitDirect方式进行发送！！！**

package storm;
    
    import lombok.SneakyThrows;
    import org.apache.storm.spout.SpoutOutputCollector;
    import org.apache.storm.task.TopologyContext;
    import org.apache.storm.topology.IRichSpout;
    import org.apache.storm.topology.OutputFieldsDeclarer;
    import org.apache.storm.tuple.Fields;
    import org.apache.storm.tuple.Values;
    
    import java.io.BufferedReader;
    import java.io.FileReader;
    import java.util.Map;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/12 15:33
     * @Version: 1.0
     * @Description:
     */
    public class WebSpout implements IRichSpout {
    
        private TopologyContext topologyContext;
        private SpoutOutputCollector spoutOutputCollector;
        private BufferedReader reader;
    
        @SneakyThrows
        @Override
        public void open(Map<String, Object> map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
            this.topologyContext = topologyContext;
            this.spoutOutputCollector = spoutOutputCollector;
            reader = new BufferedReader(new FileReader("D:/mine/weblog.txt"));
        }
    
        @Override
        public void close() {
    
        }
    
        @Override
        public void activate() {
    
        }
    
        @Override
        public void deactivate() {
    
        }
    
        @SneakyThrows
        @Override
        public void nextTuple() {
            int taskId = -1;
            Map<Integer, String> taskToComponent = topologyContext.getTaskToComponent();
            for (Map.Entry<Integer,String> entry : taskToComponent.entrySet()){
                if (entry.getValue().contains("webLogSlot")){
                    taskId = entry.getKey();
                    break;
                }
            }
            String line = null;
            while ((line = reader.readLine())!=null){
                String[] splits = line.split("\t");
                spoutOutputCollector.emitDirect(taskId,new Values(splits[0],line));
                TimeUnit.SECONDS.sleep(1);
            }
        }
    
        @Override
        public void ack(Object o) {
    
        }
    
        @Override
        public void fail(Object o) {
    
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
            outputFieldsDeclarer.declare(new Fields("domain","line"));
        }
    
        @Override
        public Map<String, Object> getComponentConfiguration() {
            return null;
        }
    }

** WebSlot.java：**

package storm;
    
    import org.apache.storm.task.OutputCollector;
    import org.apache.storm.task.TopologyContext;
    import org.apache.storm.topology.IRichBolt;
    import org.apache.storm.topology.OutputFieldsDeclarer;
    import org.apache.storm.tuple.Tuple;
    import storm.util.OSUtil;
    
    import java.util.Map;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/12 15:58
     * @Version: 1.0
     * @Description:
     */
    public class WebSlot implements IRichBolt {
        private TopologyContext topologyContext;
        private OutputCollector outputCollector;
    
        private int lineNumber = 0;
        @Override
        public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {
            this.topologyContext = topologyContext;
            this.outputCollector = outputCollector;
        }
    
        @Override
        public void execute(Tuple tuple) {
            String domain = tuple.getStringByField("domain");
            String line = tuple.getStringByField("line");
            String sessionId = line.split("\t")[1];
            lineNumber ++;
    
            OSUtil.sendMessage(this,topologyContext.getThisComponentId() + "======" + lineNumber);
        }
    
        @Override
        public void cleanup() {
    
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    
        }
    
        @Override
        public Map<String, Object> getComponentConfiguration() {
            return null;
        }
    }

package storm;
    
    import org.apache.storm.Config;
    import org.apache.storm.LocalCluster;
    import org.apache.storm.generated.StormTopology;
    import org.apache.storm.thrift.TException;
    import org.apache.storm.topology.TopologyBuilder;
    import org.apache.storm.tuple.Fields;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/12 16:07
     * @Version: 1.0
     * @Description:
     */
    public class WebLogTopology {
        public static void main(String[] args) throws Exception {
            TopologyBuilder topologyBuilder = new TopologyBuilder();
            topologyBuilder.setSpout("webLogSpout",new WebSpout());
            topologyBuilder.setBolt("webLogSlot",new WebSlot(),3).directGrouping("webLogSpout");
    
            //创建任务
            StormTopology job = topologyBuilder.createTopology();
    
            Config config = new Config();
    
            //运行任务有两种模式
            //1 本地模式   2 集群模式
    
            //1、本地模式
            LocalCluster localCluster = new LocalCluster();
            localCluster.submitTopology("MyWebLog",config,job);
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 4][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70]: /images/20221120/e1d11c05356442489f21af5b5adadba4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 1]: /images/20221120/ea548115812c46ca841e50ab5002902b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20201212164120990.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20201212164352110.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20201212175021611.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI=,size_16,color_FFFFFF,t_70