ABP 框架 数据库底层迁移 Mysql 集群

技术交流,请加QQ群:538327407

我的各种github 开源项目和代码:https://github.com/linbin524

背景

笔者 目前架构的IOT 项目是使用abp 框架作为后台,虽然abp的框架适用于中小型项目框架,但由于架构优美,笔者认为还是可以经过改造,作为大型项目中使用。但IOT 的这个项目目前刚上线不久,十几天数据库已经有了上百GB,而且由于实施检查设备状态,调用设备状态维护表,审计日志压力很大,单单审计日志一天的数据量就有几十万,目前在架构上,笔者做了几个优化处理;

1、针对审计日志,笔者重写了Abp 原有的 IAuditingStore,实现mongodb和redis 两种转移,并且针对审计日志内容做了过滤,DisableAuditing特性标记指定的类或方法不进行记录。

ps:abp 虽然有mongodb 的封装,但它的出发点是和EF 同一个模式,左右系统唯一的ORM,如果要使用abp 的mongo 封装,必须要替代EF,或者重写ABP UnitOfWorkOptions,否则直接用会出现工作单元转换失败的问题。

2、站点层面使用nginx 做了反向代理,进行多站点服务,通信模式由原来的队列、改为服务化,EventBus等方式

3、数据库底层 做了Percona XtraDB Cluster—MySQL 集群处理迁移。

思考评估:1、审计日志这样处理,从源头做了缩减,并且进行Nosql拆分,有助于缓解数据库压力。

                 2、中间层的处理是一般IOT 中间件各种脚手架的组合,成熟,也有经过多年生产环境的检验。

                 3、数据库底层 使用Percona XtraDB Cluster,是因为它支持集群,可以缓解数据库请求压力,又支持abp的事务;

               但从真正大系统考虑,其实最理性的模式应该是分片,结合SOA、或者微服务才能真正解决底层压力,目前考量了Tidb(张善友 张队推荐的)、oceanbase(淘宝 自有数据库,生产环境十年)、mycat中间件(听说这个坑多)等,

               为了暂时不做大改造,只能先使用 Percona XtraDB Cluster,后续可能使用Orleans(Azure 云框架)、akka.net(大型的框架) 或者 Service Fabric(微服务框架)

二、Percona XtraDB Cluster 评估

优点如下:

1.当执行一个查询时,在本地节点上执行。因为所有数据都在本地,无需远程访问。

2.无需集中管理。可以在任何时间点失去任何节点,但是集群将照常工作。

3.良好的读负载扩展,任意节点都可以查询。

缺点如下:

1.加入新节点,开销大。需要复制完整的数据。

2.不能有效的解决写缩放问题,所有的写操作都将发生在所有节点上。

3.有多少个节点就有多少重复的数据。

Percona XtraDB Cluster是MySQL高可用性和可扩展性的解决方案.

Percona XtraDB Cluster提供的特性有:

1.同步复制,事务要么在所有节点提交或不提交。

2.多主复制,可以在任意节点进行写操作。

3.在从服务器上并行应用事件,真正意义上的并行复制。

4.节点自动配置。

5.数据一致性,不再是异步复制。

Percona XtraDB Cluster完全兼容MySQL和Percona Server,表现在:

1.数据的兼容性

2.应用程序的兼容性:无需更改应用程序

1.集群是有节点组成的,推荐配置至少3个节点,但是也可以运行在2个节点上。

2.每个节点都是普通的mysql/percona服务器,可以将现有的数据库服务器组成集群,反之,也可以将集群拆分成单独的服务器。

3.每个节点都包含完整的数据副本。

三、部署流程

1、环境准备

  在腾讯云上开设三个测试服务器,系统 镜像 CentOS 7.5 64

用远程工具连接三台测试服务器,完成如下操作

(1)  关闭firewalld防火墙

# systemctl disable firewalld --now

关闭防火墙或者允许3306, 4444, 4567和4568四个端口的连接

(2)关闭SELINUX

# setenforce 0
# sed -i 's,^SELINUX=enforcing,SELINUX=disabled,g' /etc/selinux/config

2、主节点部署

(1)安装PXC yum源

yum install http://www.percona.com/downloads/percona-release/redhat/0.1-3/percona-release-0.1-3.noarch.rpm

(2) 安装PXC

# yum install Percona-XtraDB-Cluster-56

最终下载下来的版本是Percona-XtraDB-Cluster-56-5.6.30

(3) 修改    /etc/my.cnf

     vim  /etc/my.cnf
[mysqld]

datadir=/var/lib/mysql
user=mysql

wsrep_provider=/usr/lib64/galera3/libgalera_smm.so
#集群的ip
wsrep_cluster_address=gcomm://节点ip1,节点ip2,节点ip3

binlog_format=ROW

default_storage_engine=InnoDB

innodb_autoinc_lock_mode=2

#当前主节点的ip
wsrep_node_address=当前节点ip


wsrep_sst_method=xtrabackup-v2

wsrep_cluster_name=my_centos_cluster
#初始化一个mysql的用户和密码
wsrep_sst_auth="admin:123456"

(4)启动主节点

 systemctl start mysql@bootstrap.service

(5)进入mysql

登录 (初始化状态,无密码,遇到要输密码直接回车)

mysql -uroot -p

(6) 登录客户端查看数据库的状态,在进行权限配置允许ip访问,默认无法远程访问,但是我们需要远程通过图形化等界面查看,所以要做如下配置

mysql> show status like 'wsrep%';

CREATE USER 'admin'@'localhost' IDENTIFIED BY '123456';//如果这里报错,看一下是否有 用户存在了

GRANT RELOAD, LOCK TABLES, REPLICATION CLIENT ON *.* TO 'admin'@'localhost'; 

FLUSH PRIVILEGES;

完成后可以用Navicat For mysql 连接看一下是否可以成功访问

3、其他两个节点的配置

(1)安装PXC yum源

yum install http://www.percona.com/downloads/percona-release/redhat/0.1-3/percona-release-0.1-3.noarch.rpm

(2) 安装PXC

# yum install Percona-XtraDB-Cluster-56

(3) 修改    /etc/my.cnf

     vim  /etc/my.cnf
[mysqld]

datadir=/var/lib/mysql
user=mysql

wsrep_provider=/usr/lib64/galera3/libgalera_smm.so
#集群的ip
wsrep_cluster_address=gcomm://节点ip1,节点ip2,节点ip3

binlog_format=ROW

default_storage_engine=InnoDB

innodb_autoinc_lock_mode=2

#当前主节点的ip
wsrep_node_address=当前节点ip


wsrep_sst_method=xtrabackup-v2

wsrep_cluster_name=my_centos_cluster
#初始化一个mysql的用户和密码
wsrep_sst_auth="admin:123456"

 (4)启动当前节点(这一步和主节点不一样)

systemctl start mysql

(5)进入mysql

登录 (初始化状态,无密码,遇到要输密码直接回车)

mysql -uroot -p

(6) 登录客户端查看数据库的状态,在进行权限配置允许ip访问,默认无法远程访问,但是我们需要远程通过图形化等界面查看,所以要做如下配置

mysql> show status like 'wsrep%';

CREATE USER 'admin'@'localhost' IDENTIFIED BY '123456';//如果这里报错,看一下是否有 用户存在了

GRANT RELOAD, LOCK TABLES, REPLICATION CLIENT ON *.* TO 'admin'@'localhost'; 

FLUSH PRIVILEGES;

完成后可以用Navicat For mysql 连接看一下是否可以成功访问

 (7)可以在mysql中执行如下命令查看

 show status like 'wsrep%';

 如果正常,可以出现如下界面,标识当前三个集群节点

(8)如果出现启动节点时候出现异常,可以查看提示的操作,看看日志,百度一下看看是什么错误,怎么解决,因为各种错误都有,就不好一一解释了。

比如笔者在操作过程中就出现如下错误

ecStop=/usr/bin/mysql-systemd stop (code=exited, status=2)

后面查找原因应该是 防火墙等问题,进行关闭拦截等操作,就是一开始 环境准备的后面那一步,关闭防火墙、SELINUX,

 主节点重启

systemctl stop mysql@bootstrap.service
systemctl start mysql@bootstrap.service

其他节点也再次启动

systemctl start mysql

4、abp 进行数据库迁移

(1)abp 想要进行mysql 支持,网上的教程有,我就不重复造轮子自己参考(不要要注意 组件的版本,如果出现差异可能会失败)

 https://www.jianshu.com/p/543e34da16a7?winzoom=1

(2) 将数据库连接字符串改为 主节点

 <add name="Default" connectionString="server=主节点ip;port=3306;database=abpzero4_6db;uid=admin;password=123456;" providerName="MySql.Data.MySqlClient" />

(3) 执行迁移 

 (4)查看对应的三台服务器集群都自动同步该数据库

(5)在Appservice 中建立测试服务进行增删改查、事务等测试

using Abp.Application.Services;
using Abp.Application.Services.Dto;
using Abp.AutoMapper;
using Abp.Domain.Repositories;
using Abp.Domain.Uow;
using AutoMapper;
using System;
using System.Collections.Generic;
using System.Data.Entity;
using System.Linq;
using System.Linq.Expressions;
using System.Text;
using System.Threading.Tasks;
using System.Linq.Dynamic;
using Abp.Linq.Extensions;

using MyCompanyName.AbpZeroTemplate;
using MyCompanyName.AbpZeroTemplate.ZLDB_Domain;
using MyCompanyName.AbpZeroTemplate.Authorization.Consignee.Exporting;
using MyCompanyName.AbpZeroTemplate.ZLDB_Domain.Dtos;
using MyCompanyName.AbpZeroTemplate.Dto;

namespace MyCompanyName.AbpZeroTemplate
{



    /// <summary>
    /// 收货地址 业务实现接口
    /// </summary>
    public class ConsigneeAppService : AbpZeroTemplateAppServiceBase, IConsigneeAppService
    {
        private readonly IRepository<Consignee, Guid> _consigneeRepository;
        private readonly IConsigneeListExcelExporter _iConsigneeListExcelExporter;


        /// <summary>
        /// 构造函数自动注入我们所需要的类或接口
        /// </summary>
        public ConsigneeAppService(IRepository<Consignee, Guid> consigneeRepository,
        IConsigneeListExcelExporter iConsigneeListExcelExporter)
        {
            _consigneeRepository = consigneeRepository;
            _iConsigneeListExcelExporter = iConsigneeListExcelExporter;
            //_consigneeMongoDbRepository = consigneeMongoDbRepository;
        }

        /// <summary>
        /// 获取所有数据列表
        /// </summary>
        /// <returns>返回数据集合</returns>
        public async Task<List<ConsigneeDto>> GetAllList(string guid = "")
        {
            //try
            //{
            //    var model = new Consignee() { DealerId = System.Guid.NewGuid() };
            //    var mr = _consigneeMongoDbRepository.Insert(model);
            //}
            //catch (Exception ex)
            //{

            //    throw;
            //}


          

            //调用Task仓储的特定方法GetAllWithPeople
            var resultList = await _consigneeRepository.GetAllListAsync();

        

            return Mapper.Map<List<ConsigneeDto>>(resultList).ToList();
        }

        /// <summary>
        /// 获取分页数据列表 分页具体代码需要适当修改,如orderby 需要匹配 创建时间 或者其他数据Id(int)
        /// </summary>
        /// <returns>返回数据集合</returns>
        public async Task<PagedResultDto<ConsigneeDto>> GetPagedListAsync(PagedAndFilteredInputDto input)
        {
            var query = _consigneeRepository.GetAll();
            //TODO:根据传入的参数添加过滤条件

            var resultCount = await query.CountAsync();
            var resultconsignee = await query
            .OrderBy(x => x.Id)
            .PageBy(input)
            .ToListAsync();

            var resultListDtos = resultconsignee.MapTo<List<ConsigneeDto>>();


            return new PagedResultDto<ConsigneeDto>(
            resultCount,
            resultListDtos
            );
        }

        /// <summary>
        /// 获取指定条件的数据列表  webapi 无法使用
        /// </summary>
        /// <returns>返回数据集合</returns>
        public async Task<List<ConsigneeDto>> GetListByCodition(Expression<Func<Consignee, bool>> predicate)
        {

            var resultList = await _consigneeRepository.GetAllListAsync(predicate);
            return Mapper.Map<List<ConsigneeDto>>(resultList).ToList();
        }


        /// <summary>
        /// 导出excel 具体方法
        /// </summary>
        /// <returns>excel文件</returns>
        /// public async Task<FileDto> GetConsigneeToExcel()
        ///{
        ///    var resultList = await _consigneeRepository.GetAllListAsync();
        ///   var consigneeDtos= Mapper.Map<List<ConsigneeDto>>(resultList).ToList();
        ///   return _iConsigneeListExcelExporter.ExportToFile(consigneeDtos);
        /// }

        /// <summary>
        /// 根据指定id 获取数据实体
        /// </summary>
        /// <param name="input">当前id</param>
        /// <returns></returns>
        public async Task<ConsigneeDto> GetConsigneeForEditAsync(NullableIdDto<System.Guid> input)
        {
            var output = new ConsigneeDto();

            ConsigneeDto consigneeEditDto;

            if (input.Id.HasValue)
            {
                var entity = await _consigneeRepository.GetAsync(input.Id.Value);
                consigneeEditDto = entity.MapTo<ConsigneeDto>();
            }
            else
            {
                consigneeEditDto = new ConsigneeDto();
            }

            output = consigneeEditDto;
            return output;
        }

        /// <summary>
        /// 根据Id创建或编辑操作
        /// </summary>
        /// <param name="input">实体</param>
        /// <returns></returns>
        public async Task CreateOrUpdateConsigneeAsync(ConsigneeDto input)
        {
            if (!string.IsNullOrWhiteSpace(input.Id))
            {
                await Update(input);
            }
            else
            {
                await Create(input);
            }
        }

        /// <summary>
        /// 新增 
        /// </summary>
        /// <param name="input">新增参数</param>
        /// <returns>新增实体</returns>
        public async Task<Guid> Create(ConsigneeDto input)
        {
            input.Id = new Consignee().Id.ToString();
            var resultObj = input.MapTo<Consignee>();
            var result = await _consigneeRepository.InsertAsync(resultObj);

            return result.Id;
        }

        /// <summary>
        /// 新增 
        /// </summary>
        /// <param name="input">新增参数</param>
        /// <returns>新增实体</returns>
        public async Task<int> CreateList(List<ConsigneeDto> list)
        {
            foreach (var input in list) {
                if (input.Contact.Contains("ex")) {
                    throw new Exception("测试分布式异常!");
                }
                input.Id = new Consignee().Id.ToString();

                var resultObj = input.MapTo<Consignee>();
                var result = await _consigneeRepository.InsertAsync(resultObj);
            }
            

            return list.Count();
        }

        /// <summary>
        /// 修改
        /// </summary>
        /// <param name="input">修改参数</param>
        /// <returns>修改实体</returns>
        public async Task<ConsigneeDto> Update(ConsigneeDto input)
        {
            Consignee obj = await _consigneeRepository.GetAsync(new Guid(input.Id));
            input.MapTo(obj);
            var result = await _consigneeRepository.UpdateAsync(obj);
            return obj.MapTo<ConsigneeDto>();
        }

        /// <summary>
        /// 删除
        /// </summary>
        /// <param name="input">删除Dto</param>
        /// <returns>无返回值</returns>
        public async System.Threading.Tasks.Task Delete(EntityDto<string> input)
        {
            await _consigneeRepository.DeleteAsync(new Guid(input.Id));
        }

        /// <summary>
        /// 删除 webapi 无法使用
        /// </summary>
        /// <param name="predicate">删除条件</param>
        /// <returns>无返回值</returns>
        public async System.Threading.Tasks.Task DeleteByCondition(Expression<Func<Consignee, bool>> predicate)
        {
            await _consigneeRepository.DeleteAsync(predicate);

        }
    }




}

在swagger ui中增删改查都已经正常,而且数据在三个数据库中正常同步

针对事务,做了人为异常处理,确认会实现回滚(abp 自带工作单元处理事务)

 五、后记

 这一次只是做了简单的实验性测试,后续需要在加强深入检测,才可以用生产环境中。

原创声明,本文系作者授权云+社区发表,未经许可,不得转载。

如有侵权,请联系 yunjia_community@tencent.com 删除。

编辑于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏菩提树下的杨过

[原创]web application中使用Profile应该注意的问题

1.如何在web application中正确使用Profile web application与website的一个不同之处在于,web applicat...

2188
来自专栏帘卷西风的专栏

linux安装valgrind

转载请注明出处:帘卷西风的专栏(http://blog.csdn.net/ljxfblog)

1281
来自专栏程序员同行者

redis状态查看

https://redis.readthedocs.org/en/latest/server/slowlog.html

1132
来自专栏技术博文

Redis持久化

Redis 提供了多种不同级别的持久化方式: RDB 持久化可以在指定的时间间隔内生成数据集的时间点快照(point-in-time snapshot)。 AO...

3534
来自专栏互联网技术栈

Redis核心概念

Redis 支持简单且易用的主从复制(master-slave replication)功能, 该功能可以让从服务器(slave server)成为主服务器(m...

1764
来自专栏FreeBuf

简析60度CMS的Cookies欺骗漏洞

前言 本篇文章只是为了和大家分享漏洞的挖掘经验和思路,同时让大家认识到 Cookies 欺骗漏洞的危害和严重性。 漏洞发现时间:2017.8.16,因产商无回应...

2758
来自专栏JetpropelledSnake

Linux学习笔记之Redis单机,伪集群,Sentinel主从复制的安装和配置

 Redis是一款开源的、高性能的键-值存储(key-value store)。它常被称作是一款数据结构服务器(data structure server)。

1111
来自专栏程序员的SOD蜜

使用微服务架构思想,设计部署OAuth2.0授权认证框架

3K3
来自专栏散尽浮华

redis持久化策略梳理及主从环境下的策略调整记录

redis是一个支持持久化的内存数据库,也就是说redis需要经常将内存中的数据同步到磁盘来保证持久化。可以不定期的通过异步方式保存到磁盘上(即“半持久化模式”...

46010
来自专栏.NET开发那点事

使用Microsoft Fakes进行单元测试(1)

一:什么是单元测试 单元测试是对软件进行准确性验证的步骤。单元测试并不进行整个软件功能的测试,仅仅是对于最小工作单元的测试。一般最小工作单元就是指方法/函数等...

2326

扫码关注云+社区

领取腾讯云代金券