小项目不想引入 MQ？试试 Debezium！

作者：是奉壹呀
链接：https://juejin.cn/post/7264791359839223823
奥卡姆剃刀原理，“如无必要，勿增实体"。
在一些小型项目当中，没有引入消息中间件，也不想引入，但有一些业务逻辑想要解耦异步，那怎么办呢？
我们的web项目，单独内网部署，由于大数据配景，公司消息中间件统一利用的kafka，在一些小项目上kafka就显得很笨重。 引入rocketmq或rabittmq也没必要。 变乱或多线程也不适合。
详细一点的，之前对接的一个体系，一张记录表有10+以上的范例状态，新的需求是，针对每种状态做出对应的差别的操作。
之前写入这张记录表的时间，方式也是五花八门，有的是单条记录写入，有的是批量写入，有的调用了统一的service,有的呢直接调用了DAO层mapper直接写入。
所以想找到一个统一入口进行切入处理，就不行了。
这个时间就算引入消息队列，也必要在差别的业务方法里进行写入消息的操作。业务方也不太乐意配合改。
可以利用触发器，但它是属于上个时代的产物，槽点太多。（这里并不是完全不主张利用触发器，技术永远是为业务服务的，只要评估觉得可行，就可以利用）那么这个时间，CDC技术就可以粉墨登场了。
CDC(change data capture)数据更改捕获。
常见的数据更改捕获都是通过数据库比如mysql的binlog来达到目的。
我们可以监控mysql binlog日记，当写入一条数据的时间，吸收到数据变更日记，做出相应的操作。这样的好处是，只需导入依靠，不额外引入组件，同时无需改动之前的代码。 两边完全解耦，互不干扰。
常见的CDC框架，比如，canal (非Camel)

• canal [kə'næl]，译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日记剖析，提供增量数据订阅和消耗 早期阿里巴巴因为杭州和美国双机房部署，存在跨机房同步的业务需求，实现方式主要是基于业务 trigger 获取增量变更。 从 2010 年开始，业务逐步尝试数据库日记剖析获取增量变更进行同步，由此衍生出了大量的数据库增量订阅和消耗业务。
  
• 它是基于日记增量订阅和消耗的业务，包括
  

数据库镜像
数据库实时备份
索引构建和实时维护(拆分异构索引、倒排索引等)
业务 cache 刷新
带业务逻辑的增量数据处理

它的原理：

• canal 模拟 MySQL slave 的交互协议，伪装本身为 MySQL slave ，向 MySQL master 发送dump 协议
  
• MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
  
• canal 剖析 binary log 对象(原始为 byte 流)
  

再比如，debezium(音同 dbzm 滴BZ姆)很多人可能不太了解. 包括databus,maxwell，flink cdc（大数据范畴）等等，它们同属CDC捕获数据更改(change data capture)类的技术。

为什么是debezium

这么多技术框架，为什么选debezium？
看起来很多。但一一排除下来就debezium和canal。
sqoop,kettle,datax之类的工具，属于前大数据时代的产物，地位类似于web范畴的structs2。而且，它们基于查询而非binlog日记，实在不属于CDC。首先排除。
flink cdc是大数据范畴的框架，一般web项目的数据量属于大材小用了。
同时databus,maxwell相对比力冷门，用得比力少。
最后不用canal的缘故原由有以下几点。

• canal必要安装，这违背了“如非必要，勿增实体”的原则。
  
• canal只能对MYSQL进行CDC监控。有很大的局限性。
  
• 大数据范畴非常流行的flink cdc(阿里团队主导)底层利用的也是debezium，而非同是阿里出品的canal。
  
• debezium可借助kafka组件，将变动的数据发到kafka topic，后续的读取操作只需读取kafka,可有用淘汰数据库的读取压力。可保证一次语义，至少一次语义。
  同时，也可基于内嵌部署模式，无需我们手动部署kafka集群，可满足”如非必要，勿增实体“的原则。
  

Debezium是一个捕获数据更改(CDC)平台，并且利用Kafka和Kafka Connect实现了本身的持久性、可靠性和容错性。
每一个部署在Kafka Connect分布式的、可扩展的、容错性的服务中的connector监控一个上游数据库服务器，捕获所有的数据库更改，然后记录到一个或者多个Kafka topic(通常一个数据库表对应一个kafka topic)。
Kafka确保所有这些数据更改变乱都能够多副本并且总体上有序(Kafka只能保证一个topic的单个分区内有序)，这样，更多的客户端可以独立消耗同样的数据更改变乱而对上游数据库体系造成的影响降到很小(如果N个应用都直接去监控数据库更改，对数据库的压力为N，而用debezium报告数据库更改变乱到kafka，所有的应用都去消耗kafka中的消息，可以把对数据库的压力降到1)。
另外，客户端可以随时停止消耗，然后重启，从前次停止消耗的地方接着消耗。每个客户端可以自行决定他们是否必要exactly-once或者at-least-once消息交付语义保证，并且所有的数据库或者表的更改变乱是按照上游数据库发生的顺序被交付的。

对于不必要或者不想要这种容错级别、性能、可扩展性、可靠性的应用，他们可以利用内嵌的Debezium connector引擎来直接在应用内部运行connector。
这种应用仍必要消耗数据库更改变乱，但更盼望connector直接传递给它，而不是持久化到Kafka里。

简介
Debezium是一个开源项目，为捕获数据更改(change data capture,CDC)提供了一个低延迟的流式处理平台。你可以安装并且设置Debezium去监控你的数据库，然后你的应用就可以消耗对数据库的每一个行级别(row-level)的更改。只有已提交的更改才是可见的，所以你的应用不用担心事务(transaction)或者更改被回滚(roll back)。Debezium为所有的数据库更改变乱提供了一个统一的模子，所以你的应用不用担心每一种数据库管理体系的错综复杂性。另外，由于Debezium用持久化的、有副本备份的日记来记录数据库数据变化的历史，因此，你的应用可以随时停止再重启，而不会错过它停止运行时发生的变乱，保证了所有的变乱都能被正确地、完全地处理掉。
监控数据库，并且在数据变动的时间获得通知一直是很复杂的变乱。关系型数据库的触发器可以做到，但是只对特定的数据库有用，而且通常只能更新数据库内的状态(无法和外部的进程通信)。一些数据库提供了监控数据变动的API或者框架，但是没有一个标准，每种数据库的实现方式都是差别的，并且必要大量特定的知识和明白特定的代码才能运用。确保以相同的顺序查看和处理所有更改，同时最小化影响数据库仍然非常具有挑战性。
Debezium提供了模块为你做这些复杂的工作。一些模块是通用的，并且能够适用多种数据库管理体系，但在功能和性能方面仍有一些限制。另一些模块是为特定的数据库管理体系定制的，所以他们通常可以更多地利用数据库体系本身的特性来提供更多功能。
github官网上摆列的一些
典型应用场景：

• 缓存失效(Cache invalidation)
  经典问题 Redis与MySQL双写一致性如何保证？Debezium利用kafka单分区的有序性(忽略mysql binlog本身可能的延迟和乱序)，可完全解决此问题。
  在缓存中缓存的条目(entry)在源头被更改或者被删除的时间立刻让缓存中的条目失效。 如果缓存在一个独立的进程中运行(例如Redis，Memcache，Infinispan或者其他的)，那么简单的缓存失效逻辑可以放在独立的进程或服务中， 从而简化主应用的逻辑。在一些场景中，缓存失效逻辑可以更复杂一点，让它利用更改变乱中的更新数据去更新缓存中受影响的条目。
  

• 简化单体应用(Simplifying monolithic applications) 许多应用更新数据库，然后在数据库中的更改被提交后，做一些额外的工作：更新搜索索引，更新缓存，发送通知，运行业务逻辑，等等。 这种情况通常称为双写(dual-writes)，因为应用没有在一个事务内写多个体系。这样不但应用逻辑复杂难以维护， 而且双写容易丢失数据或者在一些体系更新乐成而另一些体系没有更新乐成的时间造成差别体系之间的状态不一致。利用捕获更改数据技术(change data capture,CDC)， 在源数据库的数据更改提交后，这些额外的工作可以被放在独立的线程或者进程(服务)中完成。这种实现方式的容错性更好，不会丢失变乱，容易扩展，并且更容易支持升级。
  

• 共享数据库(Sharing databases) 当多个应用共用同一个数据库的时间，一个应用提交的更改通常要被另一个应用感知到。一种实现方式是利用消息总线， 尽管非事务性(non-transactional)的消息总线总会受上面提到的双写(dual-writes)影响。但是，另一种实现方式，即Debezium，变得很直接：每个应用可以直接监控数据库的更改，并且相应更改。
  

• 数据集成(Data integration) 数据通常被存储在多个地方，尤其是当数据被用于差别的目的的时间，会有差别的情势。保持多体系的同步是很有挑战性的， 但是可以通过利用Debezium加上简单的变乱处理逻辑来实现简单的ETL范例的解决方案。
  

• 命令查询职责分离(CQRS) 在命令查询职责分离 Command Query Responsibility Separation (CQRS) 架构模式中，更新数据利用了一种数据模子， 读数据利用了一种或者多种数据模子。由于数据更改被记录在更新侧(update-side)，这些更改将被处理以更新各种读展示。 所以CQRS应用通常更复杂，尤其是他们必要保证可靠性和全序(totally-ordered)处理。Debezium和CDC可以使这种方式更可行： 写操作被正常记录，但是Debezium捕获数据更改，并且持久化到全序流里，然后供那些必要异步更新只读视图的服务消耗。 写侧(write-side)表可以表示面向范畴的实体(domain-oriented entities)，或者当CQRS和 Event Sourcing 联合的时间，写侧表仅仅用做追加操作命令变乱的日记。
  

springboot 整合 Debezium

推荐一个开源免费的 Spring Boot 实战项目：

https://github.com/javastacks/spring-boot-best-practice

依靠

1. <debezium.version>1.7.0.Final</debezium.version>
2. <mysql.connector.version>8.0.26</mysql.connector.version>
4. <dependency>
5.     <groupId>mysql</groupId>
6.     <artifactId>mysql-connector-java</artifactId>
7.     <version>${mysql.connector.version}</version>
8.     <scope>runtime</scope>
9. </dependency>
10. <dependency>
11.     <groupId>io.debezium</groupId>
12.     <artifactId>debezium-api</artifactId>
13.     <version>${debezium.version}</version>
14. </dependency>
15. <dependency>
16.     <groupId>io.debezium</groupId>
17.     <artifactId>debezium-embedded</artifactId>
18.     <version>${debezium.version}</version>
19. </dependency>
20. <dependency>
21.     <groupId>io.debezium</groupId>
22.     <artifactId>debezium-connector-mysql</artifactId>
23.     <version>${debezium.version}</version>
24.     <exclusions>
25.         <exclusion>
26.             <groupId>mysql</groupId>
27.             <artifactId>mysql-connector-java</artifactId>
28.         </exclusion>
29.     </exclusions>
30. </dependency>

复制代码

注意debezium版本为1.7.0.Final，对应mysql驱动为8.0.26,低于这个版本会报兼容错误。
设置

相应的设置

1. debezium.datasource.hostname = localhost
2. debezium.datasource.port = 3306
3. debezium.datasource.user = root
4. debezium.datasource.password = 123456
5. debezium.datasource.tableWhitelist = test.test
6. debezium.datasource.storageFile = E:/debezium/test/offsets/offset.dat
7. debezium.datasource.historyFile = E:/debezium/test/history/custom-file-db-history.dat
8. debezium.datasource.flushInterval = 10000
9. debezium.datasource.serverId = 1
10. debezium.datasource.serverName = name-1

复制代码

然后进行设置初始化。
主要的设置项：

1. connector.class

复制代码

• 监控的数据库范例，这里选mysql。
  

1. offset.storage

复制代码

• 选择FileOffsetBackingStore时,意思把读取进度存到当地文件，因为我们不用kafka，当利用kafka时，选KafkaOffsetBackingStore 。
  

1. offset.storage.file.filename

复制代码

• 存放读取进度的当地文件地址。
  

1. offset.flush.interval.ms

复制代码

• 读取进度刷新保存频率，默认1分钟。如果不依靠kafka的话，应该就没有exactly once只读取一次语义，应该是至少读取一次。意味着可能重复读取。如果web容器挂了，最新的读取进度没有刷新到文件里，下次重启时，就会重复读取binlog。
  

1. table.whitelist

复制代码

• 监控的表名白名单,发起设置此值，只监控这些表的binlog。
  

1. database.whitelist

复制代码

• 监控的数据库白名单,如果选此值，会忽略table.whitelist，然后监控此db下所有表的binlog。
  

1. import io.debezium.connector.mysql.MySqlConnector;
2. import io.debezium.relational.history.FileDatabaseHistory;
3. import lombok.Data;
4. import org.apache.kafka.connect.storage.FileOffsetBackingStore;
5. import org.springframework.boot.context.properties.ConfigurationProperties;
6. import org.springframework.context.annotation.Bean;
7. import org.springframework.context.annotation.Configuration;
9. import java.io.File;
10. import java.io.IOException;
12. /**
13. * @className: MysqlConfig
14. * @author: nyp
15. * @description: TODO
16. * @date: 2023/8/7 13:53
17. * @version: 1.0
18. */
19. @Configuration
20. @ConfigurationProperties(prefix ="debezium.datasource")
21. @Data
22. public class MysqlBinlogConfig {
24.     private String hostname;
25.     private String port;
26.     private String user;
27.     private String password;
28.     private String tableWhitelist;
29.     private String storageFile;
30.     private String historyFile;
31.     private Long flushInterval;
32.     private String serverId;
33.     private String serverName;
35.     @Bean
36.     public io.debezium.config.Configuration MysqlBinlogConfig () throws Exception {
37.         checkFile();
38.         io.debezium.config.Configuration configuration = io.debezium.config.Configuration.create()
39.                 .with("name", "mysql_connector")
40.                 .with("connector.class", MySqlConnector.class)
41.                 // .with("offset.storage", KafkaOffsetBackingStore.class)
42.                 .with("offset.storage", FileOffsetBackingStore.class)
43.                 .with("offset.storage.file.filename", storageFile)
44.                 .with("offset.flush.interval.ms", flushInterval)
45.                 .with("database.history", FileDatabaseHistory.class.getName())
46.                 .with("database.history.file.filename", historyFile)
47.                 .with("snapshot.mode", "Schema_only")
48.                 .with("database.server.id", serverId)
49.                 .with("database.server.name", serverName)
50.                 .with("database.hostname", hostname)
51. //                .with("database.dbname", dbname)
52.                 .with("database.port", port)
53.                 .with("database.user", user)
54.                 .with("database.password", password)
55. //                .with("database.whitelist", "test")
56.                 .with("table.whitelist", tableWhitelist)
57.                 .build();
58.         return configuration;
60.     }
62.     private void checkFile() throws IOException {
63.         String dir = storageFile.substring(0, storageFile.lastIndexOf("/"));
64.         File dirFile = new File(dir);
65.         if(!dirFile.exists()){
66.             dirFile.mkdirs();
67.         }
68.         File file = new File(storageFile);
69.         if(!file.exists()){
70.             file.createNewFile();
71.         }
72.     }
73. }

复制代码

snapshot.mode 快照模式，指定连接器启动时运行快照的条件。可能的设置有:

• initial 只有在没有为逻辑服务器名记录偏移量时，连接器才运行快照。
  
• When_needed 当连接器认为有必要时，它会在启动时运行快照。也就是说，当没有可用的偏移量时，或者当先前记录的偏移量指定了服务器中不可用的binlog位置或GTID时。
  
• Never 连接器从不利用快照。在第一次利用逻辑服务器名启动时，连接器从binlog的开头读取。谨慎设置此行为。只有当binlog保证包含数据库的整个历史记录时，它才有用。
  
• Schema_only 连接器运行模式而不是数据的快照。当您不必要主题包含数据的一致快照，而只必要主题包含自连接器启动以来的更改时，此设置非常有用。
  
• Schema_only_recovery 这是已经捕获更改的连接器的规复设置。当您重新启动连接器时，此设置允许规复损坏或丢失的数据库历史主题。您可以定期将其设置为“清理”意外增长的数据库历史主题。数据库历史主题必要无穷保留。
  

1. database.server.id

复制代码

• 伪装成slave的Debezium服务的id，自界说，有多个Debezium服务不能重复，如果重复的话会报以下异常。
  

1. io.debezium.DebeziumException: A slave with the same server_uuid/server_id as this slave has connected to the master; the first event 'binlog.000013' at 46647257, the last event read from './binlog.000013' at 125, the last byte read from './binlog.000013' at 46647257. Error code: 1236; SQLSTATE: HY000.
2.         at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.wrap(MySqlStreamingChangeEventSource.java:1167)
3.         at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$ReaderThreadLifecycleListener.onCommunicationFailure(MySqlStreamingChangeEventSource.java:1212)
4.         at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:980)
5.         at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:599)
6.         at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:857)
7.         at java.lang.Thread.run(Thread.java:750)
8. Caused by: com.github.shyiko.mysql.binlog.network.ServerException: A slave with the same server_uuid/server_id as this slave has connected to the master; the first event 'binlog.000013' at 46647257, the last event read from './binlog.000013' at 125, the last byte read from './binlog.000013' at 46647257.
9.         at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:944)
10.         ... 3 common frames omitted

复制代码

监听

设置监听服务

1. import com.alibaba.fastjson.JSON;
2. import io.debezium.config.Configuration;
3. import io.debezium.data.Envelope;
4. import io.debezium.engine.ChangeEvent;
5. import io.debezium.engine.DebeziumEngine;
6. import io.debezium.engine.format.Json;
7. import lombok.Builder;
8. import lombok.Data;
9. import lombok.extern.slf4j.Slf4j;
10. import org.apache.commons.lang3.StringUtils;
11. import org.springframework.beans.factory.annotation.Qualifier;
12. import org.springframework.stereotype.Component;
14. import javax.annotation.PostConstruct;
15. import javax.annotation.PreDestroy;
16. import javax.annotation.Resource;
17. import java.io.IOException;
18. import java.util.ArrayList;
19. import java.util.List;
20. import java.util.Map;
21. import java.util.Objects;
22. import java.util.concurrent.Executor;
24. /**
25. * @projectName: test
26. * @package: com.test.config
27. * @className: MysqlBinlogListener
28. * @author: nyp
29. * @description: TODO
30. * @date: 2023/8/7 13:56
31. * @version: 1.0
32. */
33. @Component
34. @Slf4j
35. public class MysqlBinlogListener {
37.     @Resource
38.     private Executor taskExecutor;
40.     private final List<DebeziumEngine<ChangeEvent<String, String>>> engineList = new ArrayList<>();
42.     private MysqlBinlogListener (@Qualifier("mysqlConnector") Configuration configuration) {
43.             this.engineList.add(DebeziumEngine.create(Json.class)
44.                     .using(configuration.asProperties())
45.                     .notifying(record -> receiveChangeEvent(record.value()))
46.                     .build());
47.     }
49.     private void receiveChangeEvent(String value) {
50.         if (Objects.nonNull(value)) {
51.             Map<String, Object> payload = getPayload(value);
52.             String op = JSON.parseObject(JSON.toJSONString(payload.get("op")), String.class);
53.             if (!(StringUtils.isBlank(op) || Envelope.Operation.READ.equals(op))) {
54.                 ChangeData changeData = getChangeData(payload);
55.                // 这里抛出异常会导致后面的日志监听失败
56.                 try {
57.                     mysqlBinlogService.service(changeData);
58.                 }catch (Exception e){
59.                     log.error("binlog处理异常，原数据: " + changeData, e);
60.                 }
62.             }
63.         }
64.     }
66.     @PostConstruct
67.     private void start() {
68.         for (DebeziumEngine<ChangeEvent<String, String>> engine : engineList) {
69.             taskExecutor.execute(engine);
70.         }
71.     }
73.     @PreDestroy
74.     private void stop() {
75.         for (DebeziumEngine<ChangeEvent<String, String>> engine : engineList) {
76.             if (engine != null) {
77.                 try {
78.                     engine.close();
79.                 } catch (IOException e) {
80.                     log.error("", e);
81.                 }
82.             }
83.         }
84.     }
86.     public static Map<String, Object> getPayload(String value) {
87.         Map<String, Object> map = JSON.parseObject(value, Map.class);
88.         Map<String, Object> payload = JSON.parseObject(JSON.toJSONString(map.get("payload")), Map.class);
89.         return payload;
90.     }
92.     public static ChangeData getChangeData(Map<String, Object> payload) {
93.         Map<String, Object> source = JSON.parseObject(JSON.toJSONString(payload.get("source")), Map.class);
94.         return ChangeData.builder()
95.                 .op(payload.get("op").toString())
96.                 .table(source.get("table").toString())
97.                 .after(JSON.parseObject(JSON.toJSONString(payload.get("after")), Map.class))
98.                 .source(JSON.parseObject(JSON.toJSONString(payload.get("source")), Map.class))
99.                 .before(JSON.parseObject(JSON.toJSONString(payload.get("before")), Map.class))
100.                 .build();
101.     }
103.     @Data
104.     @Builder
105.     public static class ChangeData {
106.         /**
107.          * 更改前数据
108.          */
109.         private Map<String, Object> after;
110.         private Map<String, Object> source;
111.         /**
112.          * 更改后数据
113.          */
114.         private Map<String, Object> before;
115.         /**
116.          * 更改的表名
117.          */
118.         private String table;
119.         /**
120.          * 操作类型, 枚举 Envelope.Operation
121.          */
122.         private String op;
123.     }
125. }

复制代码

将监听到的binlog日记封装为ChangeData对象，包括表名，更改前后的数据，
以及操作范例

1. READ("r"),
2. CREATE("c"),
3. UPDATE("u"),
4. DELETE("d"),
5. TRUNCATE("t");

复制代码

测试

update操作输出

1. MysqlListener.ChangeData(after = {
2.         name = Suzuki Mio2,
3.         id = 1
4. }, source = {
5.         file = binlog .000013,
6.         connector = mysql,
7.         pos = 42587833,
8.         name = test - 1,
9.         row = 0,
10.         server_id = 1,
11.         version = 1.7 .0.Final,
12.         ts_ms = 1691458956000,
13.         snapshot = false,
14.         db = test
15.         table = test
16. }, before = {
17.         name = Suzuki Mio,
18.         id = 1
19. }, table = test, op = u)
20. data = {
21.         name = Suzuki Mio2,
22.         id = 1
23. }

复制代码

新增操作输出

1. MysqlListener.ChangeData(after = {
2.         name = 王五,
3.         id = 0
4. }, source = {
5.         file = binlog .000013,
6.         connector = mysql,
7.         pos = 42588175,
8.         name = test - 1,
9.         row = 0,
10.         server_id = 1,
11.         version = 1.7 .0.Final,
12.         ts_ms = 1691459066000,
13.         snapshot = false,
14.         db = test,
15.         table = test
16. }, before = null, table = test, op = c)

复制代码

删除操作输出

1. MysqlListener.ChangeData(after = null, source = {
2.         file = binlog .000013,
3.         connector = mysql,
4.         pos = 42588959,
5.         name = test - 1,
6.         row = 0,
7.         server_id = 1,
8.         version = 1.7 .0.Final,
9.         ts_ms = 1691459104000,
10.         snapshot = false,
11.         db = test
12.         table = test
13. }, before = {
14.         name = 王五,
15.         id = 0
16. }, table = test, op = d)

复制代码

我们之前设置的保存读取进度的文件storageFile，类似于kafka的偏移量，记录的内容如下：

停止服务，对数据库进行操作，再次重启，会根据进度重新读取。
小结

本文先容了debezium，更多的时间，我们一谈到CDC，第一想到的是大量数据同步的工具。 但实在也可以利用其数据变更捕获的特性，来达到一部份消息队列的作用。
但其毕竟不能完全替换消息队列。大家理性对待与选择。
本文的重点在先容一种思路，详细的某项技术反而不那么紧张。
更多文章推荐：
1.Spring Boot 3.x 教程，太全了！
2.2,000+ 道 Java口试题及答案整理(2024最新版)
3.免费获取 IDEA 激活码的 7 种方式（2024最新版）
觉得不错，别忘了顺手点赞+转发哦！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。