【K3S】10 - 微服务拆分之MongoDB

如遇图片无法加载请使用代理访问

前言

在上文中，我成功拆分了单体服务为多个SpringBoot微服务，本文将主要讲述在数据库中，MySQL到MongDB的转型

创建MongoDB

MongoDB有多种创建方式，这里我选择的是比较稳妥的主从Replicate，版本是比较旧的4.4，因为机器的内存和带宽小，再加上网络延迟大，这里没有选择新版本的分片

巨页（Huge Pages）是一种Linux内核特性，它允许将连续的物理页面组合成一个大页面。每个大页面可以包含多个传统大小的页面，通常为2MB或1GB。

使用巨页可以提高系统的内存管理效率和性能。在一些内存密集型的应用场景中（例如数据库），使用巨页可以减少内存碎片，并且仅使用更少的页表项来映射相同数量的物理内存，从而降低了内存访问的延迟和CPU开销。

不过，使用巨页需要操作系统和应用程序的支持，并且可能需要进行一些额外的配置工作。因此它只适用于特定的应用场景，而不是所有的应用都会受益于使用巨页。

MongoDB官方文档中指出，在很多情况下，使用巨页并不能带来明显的性能提升，而且还可能会导致一些稳定性问题。具体来说，可能会遇到以下问题：

• 巨页分配过程中可能会导致更多的内存碎片，从而影响内存管理效率。
• 可能会遇到一些操作系统和硬件相关的问题，例如无法访问巨页、巨页分配失败等。
• 在某些场景下，可能会导致页面交换等性能问题。

由于MongoDB建议关闭掉 Transparent Hugepage，所以我创建一个DaemonSet来管理需要部署的节点

hostvm-ds.yaml内容如下：

# hostvm-ds.yaml
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: hostvm-configurer
  labels:
    app: startup-script
spec:
  selector:
    matchLabels:
      app: startup-script
  template:
    metadata:
      labels:
        app: startup-script
    spec:
      hostPID: true
      containers:
      - name: hostvm-configurer
        image: cnych/startup-script:v1
        securityContext:
          privileged: true #提升特权
        env:
        - name: STARTUP_SCRIPT
          value: |
            #! /bin/bash
            # 表示如果任何语句返回非零值(错误)，则立即退出脚本
            set -o errexit
            # 表示如果管道中的任何一个子命令执行失败，则整个管道命令应被视为失败，而不是只处理管道的最后一个命令的状态码。由此可以确保在管道执行过程中所有的子命令都成功完成
            set -o pipefail
            # 表示对于任何没有声明过的变量，将会输出错误信息并退出脚本执行
            set -o nounset
            
            # Disable hugepages
            echo 'never' > /sys/kernel/mm/transparent_hugepage/enabled #禁用巨页
            echo 'never' > /sys/kernel/mm/transparent_hugepage/defrag #禁用巨页

mongo.yaml内容如下：

这里仅提供没有用户名密码校验的yaml，由于我多次尝试使用key密钥验证集群模式下的auth，可惜多次尝试失败，后续有解决方法再贴出来吧，对于网上铺天盖地的在env配置用户名密码亲测无效

# mongo.yaml
apiVersion: v1
kind: Namespace #新建命名空间
metadata:
  name: mongo
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: mongo
  namespace: mongo
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: mongo
subjects:
  - kind: ServiceAccount
    name: mongo
    namespace: mongo
roleRef:
  kind: ClusterRole
  name: cluster-admin
  apiGroup: rbac.authorization.k8s.io
---
apiVersion: v1
kind: Service
metadata:
 name: mongo
 namespace: mongo
 labels:
   name: mongo
spec:
 ports:
 - port: 27017
   targetPort: 27017
 clusterIP: None
 selector:
   role: mongo
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: mongo
  namespace: mongo
spec:
  serviceName: mongo
  replicas: 3
  selector:
    matchLabels:
      role: mongo
      environment: staging
  template:
    metadata:
      labels:
        role: mongo
        environment: staging
        replicaset: MainRepSet
    spec:
      affinity:
        podAntiAffinity:  # 添加 Pod 反亲和性，将副本打散在不同的节点
          preferredDuringSchedulingIgnoredDuringExecution:  # 软策略
          - weight: 100
            podAffinityTerm:
              labelSelector:
                matchExpressions:
                - key: replicaset
                  operator: In
                  values:
                  - MainRepSet
              topologyKey: kubernetes.io/hostname
      terminationGracePeriodSeconds: 10
      serviceAccountName: mongo
      containers:
        - name: mongo
          image: mongo:4.4
          command:
            - mongod
            - "--wiredTigerCacheSizeGB"
            - "0.25"
            - "--bind_ip"
            - "0.0.0.0"
            - "--replSet"
            - MainRepSet
          ports:
            - containerPort: 27017
          volumeMounts:
            - name: mongo-data
              mountPath: /data/db  
          resources:
            requests:
              cpu: 1
              memory: 0.5Gi
        - name: mongo-sidecar
          image: cvallance/mongo-k8s-sidecar
          env:
            - name: MONGO_SIDECAR_POD_LABELS
              value: "role=mongo,environment=staging"
            - name: KUBE_NAMESPACE
              value: "mongo"
            - name: KUBERNETES_MONGO_SERVICE_NAME
              value: "mongo"
  volumeClaimTemplates:
  - metadata:
      name: mongo-data
    spec:
      accessModes: [ "ReadWriteOnce" ]
      storageClassName: local-path  # 提供一个可用的 Storageclass
      resources:
        requests:
          storage: 10Gi
---
apiVersion: v1
kind: Service
metadata:
  name: mongo-0-nodeport
  namespace: mongo
spec:
  selector:
    statefulset.kubernetes.io/pod-name: mongo-0
  ports:
    - name: tcp
      port: 27017
      protocol: TCP
      targetPort: 27017
      nodePort: 30007
  sessionAffinity: None
  type: NodePort
---
apiVersion: v1
kind: Service
metadata:
  name: mongo-1-nodeport
  namespace: mongo
spec:
  selector:
    statefulset.kubernetes.io/pod-name: mongo-1
  ports:
    - name: tcp
      port: 27017
      protocol: TCP
      targetPort: 27017
      nodePort: 30008
  sessionAffinity: None
  type: NodePort
---
apiVersion: v1
kind: Service
metadata:
  name: mongo-2-nodeport
  namespace: mongo
spec:
  selector:
    statefulset.kubernetes.io/pod-name: mongo-2
  ports:
    - name: tcp
      port: 27017
      protocol: TCP
      targetPort: 27017
      nodePort: 30009
  sessionAffinity: None
  type: NodePort

这里我们给 Mongo 的 Pod 添加了一个 sidecar 容器，主要用于副本集的配置，该 sidecar 会每5s检查一次新成员。通过几个环境变量配置指定了 Pod 的标签、命名空间和 Service。

为了保证应用的稳定性，我们通过 podAntiAffinity 指定了 Pod 的反亲和性，这样可以保证不会有两个副本出现在同一个节点上。

此外需要提供一个可用的 StorageClass，这样可以保证不同的副本数据持久化到不同的 PV。

直接运行上面的两个资源清单文件即可：

kubectl apply -f hostvm-ds.yaml
kubectl apply -f mongo.yaml

随后你会看到，主节点启动后，会根据sidecar，创建副本集

leopold-mongo-starter

这个starter主要用于spring data mongodb，套路和我们前文用的mysql starter没什么区别，除了proto定义外，这里我想强调如下几个细节：

1. ObjectId的序列化

   由于我们传输的是json序列化的字符串，但ObjectId的序列化并不是我们所想的String，所以这种特殊的类型（LocalDateTime）需要指定序列化的方式才行

2. 文档型数据库的多变，导致我们不仅仅只会等值查询，很多时候还会使用范围查询，模糊查询，这里的查询方式和mysql又不太一样，如果写法错误，可能你根本查不到

BaseDocument

对于特殊的序列化方式，我们在声明上配置我们写好的Adapter即可，比如常用的BaseDocument：

public class BaseDocument {
    @Indexed(direction = IndexDirection.ASCENDING)
    private Integer isDeleted = 0;
    @JsonAdapter(AdapterLocalDateTime.class)
    @CreatedDate
    private LocalDateTime createDate;
    @JsonAdapter(AdapterLocalDateTime.class)
    @LastModifiedDate
    private LocalDateTime updateDate;
    private String creator;
    private String updator;
}    

ObjectId

我们通过@JsonAdapter来指定序列的方式即可，对于 ObjectId 而言，我们只需要在序列化时，new一个ObjectId对象即可，例如：

import com.google.gson.TypeAdapter;
import com.google.gson.stream.JsonReader;
import com.google.gson.stream.JsonToken;
import com.google.gson.stream.JsonWriter;
import org.bson.types.ObjectId;
import org.springframework.util.ObjectUtils;

import java.io.IOException;

public class AdapterObjectId extends TypeAdapter<ObjectId> {

    @Override
    public void write(JsonWriter out, ObjectId value) throws IOException {
        if (value == null || ObjectUtils.isEmpty(value.toString())) {
            out.nullValue();
            return;
        }
        out.value(value.toString());
    }

    @Override
    public ObjectId read(JsonReader in) throws IOException {
        if (in.peek() == JsonToken.NULL) {
            in.nextNull();
            return null;
        }
        String objectIdString = in.nextString();
        if (ObjectUtils.isEmpty(objectIdString)) {
            return null;
        }
        return new ObjectId(objectIdString);
    }
}	

然后在实力类中添加这个注解即可，例如：

@Document("resource")
public class ResourceDocument extends BaseDocument {

    @Id
    @JsonAdapter(AdapterObjectId.class)
    private ObjectId id;

    ...

}

LocalDateTime

对于 LocalDateTime 而言，我们只需要在序列化时，按照LocalDateTime的序列化方式即可，例如：

import com.google.gson.TypeAdapter;
import com.google.gson.stream.JsonReader;
import com.google.gson.stream.JsonWriter;
import org.springframework.util.ObjectUtils;

import java.io.IOException;
import java.time.LocalDateTime;

public class AdapterLocalDateTime extends TypeAdapter<LocalDateTime> {
    @Override
    public void write(JsonWriter out, LocalDateTime value) throws IOException {
        if (value != null) {
            out.value(value.toString());
        }
    }

    @Override
    public LocalDateTime read(JsonReader in) throws IOException {
        String s = in.nextString();
        if (ObjectUtils.isEmpty(s)) {
            return null;
        }
        return LocalDateTime.parse(s);
    }
}

Sort

对于排序：

Query query = new Query();
List<Sort.Order> orderList = List.of(Sort.Order.asc(field), Sort.Order.desc(field));
Sort sort = Sort.by(orderList);
query.with(sort);
List<T> findList = mongoTemplate.find(query, clazz);

Page

对于分页：

int page = 1;
int limit = 10;
Query query = new Query();
query.skip((page - 1) * limit).limit(limit);
List<T> findList = mongoTemplate.find(query, clazz);

Count

对于count：

Query query = new Query();
...
long count = mongoTemplate.count(query, clazz);

Criteria

对于等值、范围判断：

Query query = new Query();
List<Criteria> criteriaList = new ArrayList<>();

// 一般等值
criteriaList.add(Criteria.where(k).is(Boolean.parseBoolean(v)));

// 一般范围
criteriaList.add(Criteria.where(k).In(Boolean.parseBoolean(v)));

// 模糊且不区分大小写
criteriaList.add(Criteria.where(k).regex(Pattern.compile(String.format(".*%s.*", v), Pattern.CASE_INSENSITIVE)));

// 模糊且区分大小写
criteriaList.add(Criteria.where(k).regex(Pattern.compile(String.format(".*%s.*", v))));

// 不模糊且不区分大小写
criteriaList.add(Criteria.where(k).regex(Pattern.compile(String.format("^%s$.*", v), Pattern.CASE_INSENSITIVE)));

// 不模糊且区分大小写 = 一般等值
criteriaList.add(Criteria.where(k).is(v));

// 时间范围查询
criteriaList.add(Criteria.where(key).gte(startLocalDateTime).lte(endLocalDateTime));
criteriaList.add(Criteria.where(key).lt(startLocalDateTime));
criteriaList.add(Criteria.where(key).gt(startLocalDateTime));
criteriaList.add(Criteria.where(key).lte(startLocalDateTime));
criteriaList.add(Criteria.where(key).gte(startLocalDateTime));

// 以上所有条件均为and
Criteria eq = new Criteria().andOperator(criteriaList.toArray(new Criteria[0]));

// 以上所有条件均为or
Criteria eq = new Criteria().orOperator(criteriaList.toArray(new Criteria[0]));

// 填充query
query.addCriteria(eq);
List<T> findList = mongoTemplate.find(query, clazz);

假删除

对于假删除（其实就是部分字段更新）：

Query query = new Query();
...
Update update = new Update();
update.set("isDeleted", 1);
update.set("updateDate", LocalDateTime.now(ZoneId.systemDefault()));
UpdateResult upsert = mongoTemplate.updateMulti(query, update, clazz);

事务

由于分布式的MongoDB，开启事务时，查询必须在主节点执行，但我同时想让普通的查询（不含事务）在从节点执行，就需要做如下配置，指定事务执行的节点：

import com.mongodb.ClientSessionOptions;
import com.mongodb.ReadPreference;
import com.mongodb.TransactionOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.MongoDatabaseFactory;
import org.springframework.data.mongodb.MongoTransactionManager;

@Configuration
public class MongoTransactionConfig {

    @Bean
    public MongoTransactionManager transactionManager(MongoDatabaseFactory mongoDatabaseFactory) {
        MongoDatabaseFactory mongoDatabaseFactory1 = mongoDatabaseFactory.withSession(ClientSessionOptions.builder().defaultTransactionOptions(TransactionOptions.builder()
                .readPreference(ReadPreference.primary())
                .build()).build());
        return new MongoTransactionManager(mongoDatabaseFactory1);
    }
}

然后使用Spring的事务即可：

@Override
@Transactional(rollbackFor = Exception.class)
public void smartSave(SmartSaveReqProto.SmartSaveReq request, StreamObserver<SmartSaveRespProto.SmartSaveResp> responseObserver) {
    try {
        ...
    } catch (Exception e) {
        ...
        TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
    }
}

启动类开启事务：

...
@EnableTransactionManagement
public class TestApplication extends SpringBootServletInitializer {
    public static void main(String[] args) {
        SpringApplication.run(TestApplication.class, args);
    }
}

封装

在BaseService中，我们通过泛型来实现：

public class BaseService<T extends BaseDocument, C extends MongoRepository<T, ObjectId>> {
    private MongoTemplate mongoTemplate;
    private C repository;
    public BaseService(MongoTemplate mongoTemplate, C repository) {
        this.mongoTemplate = mongoTemplate;
        this.repository = repository;
    }
    public C getRepository() {
        return repository;
    }

    public void setRepository(C repository) {
        this.repository = repository;
    }

    public MongoTemplate getMongoTemplate() {
        return mongoTemplate;
    }

    public void setMongoTemplate(MongoTemplate mongoTemplate) {
        this.mongoTemplate = mongoTemplate;
    }
}    

封装好后，在各个service继承此类，并字段注入即可：

@Service
public class ResourceDocumentService extends BaseService<ResourceDocument, ResourceDocumentRepository> {
    public ResourceDocumentService(@Autowired MongoTemplate mongoTemplate, @Autowired ResourceDocumentRepository repository) {
        super(mongoTemplate, repository);
    }
}

由于这里我自定义的代码比较多，就不贴源码了，原理都写在这里了。

最后你可以在启动类配置审计功能，具体原理需要搭配BaseDocument，这里就不细说了：

...
@EnableMongoAuditing
public class MyServiceApplication extends SpringBootServletInitializer {
    ...
}

坑点

1. andOperator

对于 where a=b and c=d and e=f这种查询，我们不能按照一般思维去写：

// 错误写法
new Criteria()
    .andOperator(Criteria.where("a").is("b"))
    .andOperator(Criteria.where("c").is("d"))
    .andOperator(Criteria.where("e").is("f"));

// 正确写法
new Criteria().andOperator(Criteria.where("a").is("b"), 
                           Criteria.where("c").is("d"), 
                           Criteria.where("e").is("f"));
// 多个andOperator写法
new Criteria().andOperator(
        new Criteria().andOperator(Criteria.where("a").is("b")),
        new Criteria().andOperator(Criteria.where("c").is("d"), Criteria.where("e").is("f"))
);

// orOperator同理
new Criteria().andOperator(
        new Criteria().andOperator(Criteria.where("a").is("b")),
        new Criteria().orOperator(Criteria.where("c").is("d"), Criteria.where("e").is("f"))
);

这就导致我们在封装上会有困难，这里贴一下我封装好的参考下：

private Criteria spec(SmartDTO smartDTO) {
    Criteria timeCriteria = new Criteria();
    if (!ObjectUtils.isEmpty(smartDTO.getTimeSettingDTOList())) {
        timeCriteria.andOperator(time(smartDTO.getTimeSettingDTOList()).toArray(new Criteria[0]));
    }
    Criteria eqCriteria = new Criteria();
    if (!ObjectUtils.isEmpty(smartDTO.getEqSettingDTOList())) {
        List<Criteria> criteriaList = new ArrayList<>(eq(smartDTO.getEqSettingDTOList()));
        if (smartDTO.getEqOr()) {
            eqCriteria = eqCriteria.orOperator(criteriaList.toArray(new Criteria[0]));
        } else {
            eqCriteria = eqCriteria.andOperator(criteriaList.toArray(new Criteria[0]));
        }
    }
    return new Criteria().andOperator(eqCriteria, timeCriteria);
}

其他

这个月我成功整合了MongoDB，并将MySQL中的表转为了MongoDB的集合，现在每个集合已经很健壮，其他服务更改文档时也不像原先MySQL那样复杂，现在有了json类型的文档，日后对于写入es整合efk埋下了一个伏笔。

当然，如果这个月只完成了这些，对于已经整合了MySQL starter的我们来说已经慢了，下一篇文章中，我将记录，我是如何将300MB的java程序转型为3MB的go程序，这是一个很大的进步，包含了很多的坑和go语言的学习~~

参考资料

[1] 在 Kubernetes 上编排 MongoDB 集群-腾讯云开发者社区-腾讯云 (tencent.com)