# iptables基础管理shell脚本


这是一个管理iptables防火墙基础规则的shell脚本。主要功能包括：

```
1. 查看当前iptables的规则。
2. 清空所有iptables的规则
3. 放行指定IP访问。
4. 封堵常见端口。
5. 自定义规则。
6. 删除单条iptables规则。
7. 关闭selinux。8. 退出脚本。
```

```
#!/bin/bash
tables(){
while true
do
echo "
 (1) 查看iptables
 (2) 清空所有规则
 (3) 放行指定IP访问
 (4) 封堵常见端口
 (5) 自定义规则
 (6) 删除单条规则
 (7) 关闭selinux
 (0) 退出"
read -p "选择要执行的项:  " put
case $put in

1)
echo "************************************************************************"
iptables -nvL
echo "************************************************************************"
echo "*************************************NAT********************************"
iptables -nL -t nat
echo "*************************************NAT********************************"
;;
2)
iptables -F
iptables -t nat -F
;;
3)
read -p "请输入要放行的IP: " a 
#ipadd=$(ifconfig |grep "broadcast"|awk '{print $2}')
iptables -I INPUT -s $a   -j ACCEPT
[ $? -eq 0 ]
echo "已放行$a"
sleep 2
;;
4)
read -p "请务必先放行IP再执行此操作，否则会导致ssh无法登录设备yes/exit：" waring
if [ "$waring" = "yes" ];then
iptables -A INPUT -p tcp -m multiport  --dport 22,23 -j DROP
[ $? -eq 0 ]
echo "ssh,telnet端口已禁止所有IP连接"
sleep 2       
iptables -A INPUT -p tcp --dport 3306 -j DROP
[ $? -eq 0 ]   
 echo "mysql端口已禁止所有IP连接"
sleep 2
iptables -A INPUT -p tcp -m multiport  --dport 21,20 -j DROP #-m multiport  同时封堵多个端口
[ $? -eq 0 ] 
   echo "ftp端口已禁用所有IP连接"
sleep 2     
iptables -A INPUT -p tcp -m multiport  --dport 445,139,135,137,138,1434  -j DROP
  [ $? -eq 0 ]
    echo "共享服务端口已禁用所有IP连接"
iptables -A INPUT -p tcp -m multiport --dport 25,53
[ $? -eq 0 ]
echo "邮件传输协议端口已禁用所有IP连接"
elif [ "waring" = "exit" ];then
  exit 0
fi
;;
5)
#IP=$(ifconfig |grep "broadcast"|awk '{print $2}')
echo "输入1禁止访问所有端口，输入2指定端口禁止访问,exit退出"
read -p "输入策略写入方式：" A
if [ "$A" = 1 ];then
 read -p "请输入要放行的IP：" IP
 iptables -A INPUT -j DROP
 echo "已禁止所有连接"
 iptables -I INPUT -s $IP -j ACCEPT
 echo "允许$IP访问"
elif [ "$A" = 2 ];then
 read -p "请输入放行的IP：" IPADD
 read -p "请输入端口协议(tcp/udp): " TCP
 #read -p "请输入允许访问的端口：" PORT
 read -p "请输入禁止访问的端口：" DRPORT
 iptables -A INPUT -p $TCP -m multiport  --dport $DRPORT -j DROP       
 echo "$DRPORT已禁止所有连接"
 iptables -I INPUT -s $IPADD   -j ACCEPT   
 echo "允许$IPADD访问"
elif [ "$A" = "exit" ];then
 continue
fi
;;
6)
iptables -L -n --line-number              
read -p "请输入要删除的规则序号：" A
iptables -D INPUT $A
;;
7)
echo -e "\033[34m 输入1临时关闭firewalld和selinux，输入2永久关闭 \033[0m"
read -p "请输入要操作的功能：" PUT
if [ "$PUT" = 1 ];then
  systemctl stop firewalld.service
  setenforce 0
  echo -e "\033[31m 已临时关闭firewalld和selinux \033[0m"
elif [ "$PUT" = 2 ];then
  systemctl stop firewalld.service
  systemctl disable firewalld.service
  sed -i "s/SELINUX=enforcing/SELINUX=disabled/" /etc/selinux/config    
  echo -e "\033[31m 已永久关闭firewalld和selinux,需重启系统生效 \033[0m"
fi
sleep 3
echo -e "\033[34m 重新启用firewalld输入1/启用selinux输入2/查看状态请输入"status" \033[0m"
read -p "请输入你的选择:  " input

if [ "$input" = 1 ];then
     systemctl start firewalld.service
     systemctl enable firewalld.service
     echo "已启用firewalld"

elif [ "$input" = 2 ];then
 setenforce 1
 sed -i "s/SELINUX=disabled/SELINUX=enforcing/" /etc/selinux/config
    echo "已启用selinux"

  elif [ "$input" = "status" ];then
echo -e "\033[34m 查看firewalld状态: \033[0m"
echo ""
firewall-cmd --state
echo ""
echo -e "\033[34m 查看selinux状态: \033[0m"
echo ""
/usr/sbin/sestatus -v
fi
;;
0)
break
exit 0
;;
*)
echo -e "\033[31m 您的输入有误,请重新输入[0~7]的数字 \033[0m"
;;
esac
done
}
tables
```

iptables基础管理shell脚本

# 部署K8S集群的高可用性



接上篇，为确保 K8s 集群的高可用性，可以配置 haproxy 和 keepalived。这样可以避免单个节点故障导致服务不可用，从而实现集群的高可用性。

## 安装haproxy和keepalived,在两台Master节点上安装

```
yum install haproxy keepalived -y
```
## 修改配置文件，添加master节点配置负载均衡算法为轮询

```
vim /etc/haproxy/haproxy.cfg
 
frontend k8s  #定义一个前端命名为k8s，负责接收客户端的请求，并将其转发到后端
    bind *:16443 #监听16443端口
    mode tcp  #以TCP模式运行
    option tcplog #启用TCP日志记录
    tcp-request inspect-delay 5s #请求延迟为5秒
    default_backend             k8s  #关联后端名

backend k8s   #定义一个后端，将前端的请求转发到后端的真实服务器中
    mode tcp  
    option tcplog
    option tcp-check
    balance roundrobin  #设置负载均衡规则为轮询，按顺序将流量分配给后端服务器
    default-server inter 10s downinter 5s rise 2 fa11 2 slowstart 60s maxconn 250 maxqueue 256 weight 100  #设置服务器健康检查间隔为10秒，如果服务器被标记为不可用则5秒检查一次，服务器需连续两次健康检查通过才被标记为可用状态，服务器在健康检查失败两次后标记为不可用，请求队列最大为256，权重为100
    server  k8s-server 192.168.3.1:6443 check  #定义后端真实服务器+端口，启用check健康检查
    server  k8s-node2  192.168.3.2:6443 check
```
## 将配置文件复制到另一台master节点上
```
scp /etc/haproxy/haproxy.cfg k8s-node1:/etc/haproxy/
```
## 启动服务,并设置为开机自启,在所有master节点上都要操作

```
systemctl start haproxy.service
systemctl enable haproxy.service
```
## 配置keepalived，关联监测脚本

```
vim /etc/keepalived/keepalived.conf

! Configuration File for keepalived

global_defs {
        router_id K8S-2
script_user root
        enable_script_security
}
vrrp_script check_server{

        script "/etc/keepalived/check.sh"
        interval 3  #脚本运行的间隔时间为 3 秒。
        weight -10  #如果脚本失败，权重减少10。
        fall 2   #连续两次脚本失败后，状态被认为是不可用。
  rise 1   #脚本成功一次后，状态被认为是可用。
}

vrrp_instance VI_1{
        state BACKUP     #设置 VRRP 实例的初始状态为 BACKUP,主节点则设置为MASTER
        interface ens224    #指定 VRRP 实例使用的网络接口为 ens224。
        mcast_src_ip 192.168.3.2  #设置为本机IP
        virtual_router_id 51   #设置虚拟组ID 为51，主备服务器需在同一个组里
        priority 50      #设置优先级
        advert_int 2     #设置心跳检测时间。
        authentication{     #配置主备之间的认证
                auth_type PASS
                auth_pass 123
        }
        virtual_ipaddress{    #设置VIP
                192.168.3.254/24
        }
        track_script{     #关联检测脚本
                check_server
        }

}
```
编写检测脚本，将其放在keepalived目录下用于监控haproxy服务进程,当该进程不在时则关闭当前节点的keepalived服务
```
#!/usr/bin/env bash
if pgrep haproxy 1>/dev/null ;then
  true
else
        systemctl stop keepalived.service
        exit 1
fi
```
将配置文件与脚本复制一份给另一台Master服务器，以此组成K8S的高可用

```
scp /etc/keepalived/keepalived.conf k8s-node1:/etc/keepalived/
scp /etc/keepalived/check.sh k8s-node1:/etc/keepalived/
```
故障切换测试手动关闭MASTER节点的haproxy服务，查看VIP是否会切换至备份节点中

```
systemctl stop haproxy
```
关闭后VIP自动切换

![image.png](/static/img/22ed99fa99bb4c377b37be4db8fb8e0e.image.webp)
在备份节点中查看,VIP已经切换过来了

![image.png](/static/img/6766bcb44541b637f0822d357e87174d.image.webp)
查询节点状态,命令可以正常执行
```

![image.png](/static/img/a7ed494bd07cd899112be93a393df9b9.image.webp)
```
优化脚本

由于在检测到haproxy服务关闭后会将keepalived服务关闭，且服务恢复后VIP不会自动切换回来。可使用定时任务+脚本的方式，循环检测，当haproxy服务恢复后检查keepalived服务，如果未启动则执行启动命令以此来恢复MASTER身份

```
#!/usr/bin/env bash
if pgrep haproxy 1>/dev/null ;then
        if systemctl status keepalived.service |grep "inactive" 1>/dev/null;then
                systemctl start keepalived.service
        else
                exit 0
        fi
fi
```
编写定时任务

```
*/1 * * * * bash /mnt/k8s_dir/check.sh
```
手动恢复haproxy服务

```
systemctl start haproxy
```
当触发定时任务时，VIP也自动切换回来了

![image.png](/static/img/b58997a1edb8bd4ac38e4eb70c180f5d.image.webp)

![image.png](/static/img/42cd496f0a5ea1bc36cb60d2a0bea28e.image.webp)

部署K8S集群的高可用性

# 基于kraft的kafka集群部署

“ Kafka 3.0版本之后，可以脱离对zookeeper的依赖，由kraft来对集群的controller进行选举，大大增加了kafka的独立性，本文将mark一下基于kafka的kafka集群搭建过程”

![image.png](/static/img/b2d8d4d69abc5e5802e849160dc7f3cc.image.webp)

# 01 kafka的基本构成
  kafka集群是由多个kafka broker节点构成，在多个kafka的broker节点中，将会被选举出一个controller节点，该controller是一个特殊的角色，主要用来管理和协调整个kafka集群，对kafka的定时任务、事件队列、事件线程处理、zk监听等进行管理，具体主要包括：更新集群元数据信息、创建topic、删除topic、分区重分配、leader副本选举、topic分区扩展、broker加入集群、broker崩溃、受控关闭、controller leader选举等。

而这些元数据，kafka本身并不保存，而是通过zookeeper进行存储，即zookeeper在kafka中主要承担了：
- 管理Kafka broker节点的状态信息，比如broker的上下线状态、topic分区信息、副本信息等；
- 存储和管理Kafka集群的元数据信息和配置信息，
- 监控Kafka broker节点的状态信息，帮助Kafka集群实现自动故障转移和负载均衡等功能。
- Kafka的broker集群中的controller的选择，通过zk的临时节点争抢获得。
因此，kafka的正常运行，无法脱离zookeeper。
但在kafka 3.0以上的版本中，kafka自身引入了kraft机制，其相关的元数据也可以保存到controller的节点中了。因此，我们也可以认为，在kafka 3.0版本后，它可以不依赖于zookeeper进行执行。


# 02—kafka集群部署规划

在版本2.8之前，kafka集群部署必须依赖zookeeper。在构建整个kafka集群的时候，我们可以不对broker的数量进行限制，但是需要保证提前创建一个奇数个节点的zookeeper集群；    

在3.0之后，kafka引入kraft模式，在该模式下则无需通过zookeeper进行管理，但是为了保证集群的可靠性，我们需要保证kafka集群主机的数量为奇数个。

集群规划：
![image.png](/static/img/6c12876a9d887d0ee4e788575825493e.image.webp)


# 03—kafka集群部署步骤
## 解压文件
```
tar -xvzf kafka_2.13-3.6.0.tgz -C /apps
```    
## 修改配置文件

```
cd /apps/kafka_2.13-3.6.0/config/kraft
vi server.properties
需要修改的内容包括：
# 服务器的角色。这个设置将会使该节点进入kraft模式，其中角色可以分为controller和broker，即该节点可以作为broker，同时也可以作为controller
process.roles=broker,controller

# 设置节点的id，每个节点的id必须不一样
node.id=1

# 设置controller集群（默认管理节点端口为9093，这里我们设置为19093）
controller.quorum.voters=1@192.168.1.1:19093,2@192.168.1.2:19093,3@192.168.1.3:19093

# 带有PLAINTEXT监听名称使用端口19092，带有CONTROLLER的监听名称使用19093
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:19092#     或者#     listeners = PLAINTEXT://your.host.name:19092,CONTROLLER://192.168.1.1:19093

#不同服务器绑定的端口listeners=PLAINTEXT://192.168.1.1:19092,CONTROLLER://192.168.1.1:19093

# 监听器名称、主机名和代理将向客户端公布的端口.(broker 对外暴露的地址)
# 如果未设置，则使用"listeners"的值.
advertised.listeners=PLAINTEXT://192.168.1.1:19092
其他参数可根据实际去需求进行更改。
```
## 其他节点修改
### node2:

```
# 服务器的角色。这个设置将会使该节点进入kraft模式，其中角色可以分为controller和broker，即该节点可以作为broker，同时也可以作为
controllerprocess.roles=broker,controller

# 设置节点的id，每个节点的id必须不一样
node.id=2
# 设置controller集群（默认管理节点端口为9093，这里我们设置为19093）

controller.quorum.voters=1@192.168.1.1:19093,2@192.168.1.2:19093,3@192.168.1.3:19093

# 带有PLAINTEXT监听名称使用端口19092，带有CONTROLLER的监听名称使用19093

#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:19092
#     或者
#     listeners = LAINTEXT://your.host.name:19092,CONTROLLER://192.168.1.1:19093

#不同服务器绑定的端口listeners=PLAINTEXT://192.168.1.2:19092,CONTROLLER://192.168.1.2:19093
# 监听器名称、主机名和代理将向客户端公布的端口.(broker 对外暴露的地址)
# 如果未设置，则使用"listeners"的值.

advertised.listeners=PLAINTEXT://192.168.1.2:19092
其他参数可根据实际去需求进行更改。
```
### node3:

```
# 服务器的角色。这个设置将会使该节点进入kraft模式，其中角色可以分为controller和broker，即该节点可以作为broker，同时也可以作为

controllerprocess.roles=broker,controller
# 设置节点的id，每个节点的id必须不一样
node.id=3

# 设置controller集群（默认管理节点端口为9093，这里我们设置为19093）
controller.quorum.voters=1@192.168.1.1:19093,2@192.168.1.2:19093,3@192.168.1.3:19093

# 带有PLAINTEXT监听名称使用端口19092，带有CONTROLLER的监听名称使用19093
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:19092
#     或者
#     listeners = PLAINTEXT://your.host.name:19092,CONTROLLER://192.168.1.1:19093#不同服务器绑定的端口listeners=PLAINTEXT://192.168.1.3:19092,CONTROLLER://192.168.1.3:19093
# 监听器名称、主机名和代理将向客户端公布的端口.(broker 对外暴露的地址)
# 如果未设置，则使用"listeners"的值.
advertised.listeners=PLAINTEXT://192.168.1.3:19092
其他参数可根据实际去需求进行更改。
```
## 生成随机uuid

```
/apps/kafka_2.13-3.6.0/bin/kafka-storage.sh random-uuid
会随机输出一个uuid值，比如：
9WbxA-qWMNFv-WAbLQ53Mx
```
## 注册集群服务(每个节点执行)
用生成的uuid格式化目录，让三个节点配置认定为同一个集群(在每一个节点执行)：

```
/apps/kafka_2.13-3.6.0/bin/kafka-storage.sh format -t 9WbxA-qWMNFv-WAbLQ53Mx -c /apps/kafka_2.13-3.6.0/config/kraft/server.properties
```
## 启动服务启动服务，在各个节点执行：

```
/apps/kafka_2.13-3.6.0/bin/kafka-server-start.sh -daemon /apps/kafka_2.13-3.6.0/config/kraft/server.properties

也可以通过nohup的方式启动：
nohup sh /apps/kafka_2.13-3.6.0/bin/kafka-server-start.sh /apps/kafka_2.13-3.6.0/config/kraft/server.properties > /apps/kafka_2.13-3.6.0/startkafka.out &
```
## 检查状态    

对于状态检查，可以通过kafka-cluster.sh、kafka-metadata-quorum.sh脚本进行检查，该脚本主要是用于检查kraft模式下集群元数据的状态。

```
查看集群id
/apps/kafka_2.13-3.6.0/bin/kafka-cluster.sh --bootstrap-server 192.168.1.1:19092,192.168.1.2:19092,192.168.1.3:19092 cluster-id
查看集群信息以及状态
/apps/kafka_2.13-3.6.0/bin/kafka-metadata-server.sh --bootstrap-server 192.168.1.1:19092,192.168.1.2:19092,192.168.1.3:19092 describe --status
查看节点角色
/apps/kafka_2.13-3.6.0/bin/kafka-metadata-server.sh --bootstrap-server 192.168.1.1:19092,192.168.1.2:19092,192.168.1.3:19092 --replication
```
## 创建topic

```
./kafka-topics.sh --bootstrap-server 192.168.1.1:19092,192.168.1.2:19092,192.168.1.3:19092 --create --topic socketTopic --partitions 3   --replication-factor 2
```
## 查看topic

```
查看存在的topic列表：
./kafka-topics.sh --bootstrap-server 192.168.1.1:19092,192.168.1.2:19092,192.168.1.3:19092 --list
查看topic的详细信息：
./kafka-topics.sh --bootstrap-server 192.168.1.1:19092,192.168.1.2:19092,192.168.1.3:19092 --describe
```

基于kraft的kafka集群部署

## mongodb
因为mongodb是NOSQL，所以他的数据也是比较重要的，所以做个持久化，也就是将容器中的数据映射到机器上；

### 创建存放数据、日志、配置文件的目录：

```
mkdir /data/mongo/config -p
mkdir /data/mongo/data
mkdir /data/mongo/logs
```

创建配置文件：

```
touch /data/mongo/config/mongod.conf
chmod 777 /data/mongo
```

修改配置文件：

```
vim /data/mongo/config/mongod.conf 
```

```
# 数据库存储路径
dbpath=/data/mongo/data
 
# 日志文件路径
logpath=/data/mongo/logs/mongod.log
 
# 监听的端口
port=27017
 
# 允许所有的 IP 地址连接
bind_ip=0.0.0.0
 
# 启用日志记录
journal=true
 
# 是否后台运行
fork=true
 
# 启用身份验证
#auth=true
```


启动容器：

```
[root@docker ~/docker-com]$ cat mongo.yaml 
version: '3.8'

services:
  mongo:
    container_name: mongo
    image: mongo:3.4
    ports:
      - "17017:27017"
    volumes:
      - /data/mongo/config/mongo.conf:/etc/mongod.conf
      - /data/mongo/data:/data/db
      - /data/mongo/logs:/var/log/mongodb
    environment:
      MONGO_INITDB_ROOT_USERNAME: admin
      MONGO_INITDB_ROOT_PASSWORD: 123456
    restart: always
```
命令解释：

```
    docker run：运行 Docker 容器的命令。
    -itd：以守护进程模式运行容器，并在容器内分配一个伪终端。
    --name mongo：为容器指定一个名称。
    -p 17017:27017：将容器的 27017 端口映射到主机的 17017 端口，允许外部访问 MongoDB 数据库。
    -v /data/mongo/config/mongod.conf:/etc/mongod.conf：将主机上的/data/mongo/config/mongod.conf文件挂载到容器的/etc/mongod.conf位置，作为 MongoDB 的配置文件。
    -v /data/mongo/data:/data/db：将主机上的/data/mongo/data目录挂载到容器的/data/db位置，作为 MongoDB 的数据存储目录。
    -v /data/mongo/logs:/var/log/mongodb：将主机上的/data/mongo/logs目录挂载到容器的/var/log/mongodb位置，作为 MongoDB 的日志存储目录。
    -e MONGO_INITDB_ROOT_USERNAME=admin：设置 MongoDB 初始化时的 root 用户名为admin。
    -e MONGO_INITDB_ROOT_PASSWORD=123456：设置 MongoDB 初始化时的 root 密码为123456。
    --restart=always：容器在退出后总是自动重启。
    mongo：指定要运行的 Docker 镜像为 MongoDB。
```
进入容器中：
```
docker exec -it mongo /bin/bash
```

![image.png](/static/img/0af8663ac13367a0e8ef9706445a5c2e.image.webp)


```
mongo --host mongo -u admin -p 123456 --authenticationDatabase admin
```
使用admin账户 连接：

```
use admin
```
切换完之后，开始进行用户认证：

```
db.auth("admin","123456")
```

请按照以下步骤重新创建用户：

确保您连接到 admin 数据库：

```
docker exec -it mongo mongo -u admin -p 123456 --authenticationDatabase admin
```

创建 goldmanUser 用户并授予其在 goldman 数据库上的 readWrite 权限：


```
use goldman
db.createUser({
    user: "goldmanUser",
    pwd: "mongo",
    roles: [{ role: "readWrite", db: "goldman" }]
});
```

之后，您可以再次检查 goldman 数据库中的用户：

```
db.getUsers()
```

尝试使用 goldmanUser 进行连接：

```
docker exec -it mongo mongo -u goldmanUser -p mongo --authenticationDatabase goldman
```

mongodb

# docker-compose安装方式
```
version: '3.8'

services:
  mysql:
    image: mysql:5.7.43
    container_name: mysql_5_7_43
    environment:
      MYSQL_ROOT_PASSWORD: '密码'
    ports:
      - "3306:3306"
    volumes:
      - /root/mysql5.7.43:/var/lib/mysql
    restart: always
```
# 授权对外开放和设置密码

```
docker exec -it mysql_5_7_43 /bin/bash
mysql -uroot -p密码
ALTER USER 'root'@'%' IDENTIFIED BY '新密码';
# 刷新权限
FLUSH PRIVILEGES;
```
# 导出

```
# 本机数据库
mysqldump -u 账号 -p密码 -P 端口 --databases single_edu > single_edu.sql
# 远端数据库
mysqldump -h 远程数据库ip -u shield -p密码 -P 端口 --databases social_study_user > social_study_user.sql
# 导出的时候忽略sys_log表
mysqldump -h 远程数据库ip -u shield -p密码 -P 端口 --databases social_study_user --ignore-table=social_study_user.sys_log > social_study_user.sql
```
- -u 账号：指定 MySQL 用户名。
- -p密码：指定 MySQL 密码（注意密码和 -p 之间没有空格）。
- -P 端口：指定 MySQL 端口号。
- --databases single_edu：指定要导出的数据库名（single_edu）。
- single_edu.sql：将导出的数据库内容保存到 single_edu.sql 文件中。

## tar打包 rsync加速传输 bbcp传输
```
tar czvf social_study_edu.tar.gz social_study_edu.sql

rsync -avP social_study_edu.tar.gz root@ip:/root/mysql-data

bbcp social_study_edu.sql root@192.168.1.53:/var/lib/docker
```
## 导入docker

```
docker cp /root/mysql-data/social_study_user.sql  mysql_5_7_43:/root/social_study_user.sql
```
## 导入数据库

```
mysql -u root -p密码 < social_study_edu.sql
mysql -u root -p密码 --databases single_edu  < social_study_edu.sql
```
## 进程监控

```
# mysql进程监控
 SHOW PROCESSLIST;
```

# 报错问题
Caused by: java.sql.SQLException: null,  message from server: "Host '10.100.20.204' is blocked because of many connection errors; unblock with 'mysqladmin flush-hosts'"
增加线程池

```
max_connections = 500
```

## 数据库拉取
前提条件，mysql版本一致
需求，手动从生产库拉去指定数据库到目标库
```
[root]$ cat xxx.py 
import pymysql
import requests
from concurrent.futures import ThreadPoolExecutor
import gc  # 引入垃圾回收模块

def send_dingtalk_notification(message):
    url = "钉钉的通知webhook"
    headers = {"Content-Type": "application/json"}
    data = {
        "msgtype": "text",
        "text": {
            "content": message
        }
    }
    response = requests.post(url, json=data, headers=headers)
    if response.status_code == 200:
        print("钉钉通知发送成功")
    else:
        print("钉钉通知发送失败", response.text)

def insert_data(target_config, table_name, rows):
    target_db = pymysql.connect(**target_config)
    try:
        target_cursor = target_db.cursor()

        # 插入数据
        if rows:
            insert_query = f"""
            INSERT INTO {table_name} ({', '.join(rows[0].keys())})
            VALUES ({', '.join(['%s'] * len(rows[0]))})
            ON DUPLICATE KEY UPDATE 
            {', '.join([f"{col}=VALUES({col})" for col in rows[0].keys()])};
            """
            target_cursor.executemany(insert_query, [tuple(row.values()) for row in rows])
            target_db.commit()
            print(f"成功插入 {len(rows)} 条数据到表 {table_name}。")

        # 强制垃圾回收
        gc.collect()

    except Exception as e:
        print(f"插入数据时出现错误: {e}")
        target_db.rollback()
    finally:
        target_cursor.close()
        target_db.close()

def sync_table(source_config, target_config, table_name):
    if table_name in ['sys_log', 'user_login_log']:
        print(f"跳过表 {table_name} 的同步。")
        return

    source_db = pymysql.connect(**source_config)
    try:
        source_cursor = source_db.cursor()

        # 统计源表中的总记录数
        source_cursor.execute(f"SELECT COUNT(*) FROM {table_name};")
        total_count = source_cursor.fetchone()[0]
        print(f"表 {table_name} 中总共有 {total_count} 条数据。")

        batch_size = 100000  # 每批处理的大小
        inserted_count = 0  # 记录已插入的总数

        # 获取列名
        source_cursor.execute(f"SHOW COLUMNS FROM {table_name};")
        columns = [col[0] for col in source_cursor.fetchall()]

        # 使用生成器逐批获取数据
        for offset in range(0, total_count, batch_size):
            query = f"SELECT * FROM {table_name} LIMIT {offset}, {batch_size};"
            source_cursor.execute(query)
            rows = source_cursor.fetchall()

            if not rows:
                break  # 没有更多数据

            # 将元组转换为字典
            rows_dict = [dict(zip(columns, row)) for row in rows]

            # 使用线程池进行并行插入
            with ThreadPoolExecutor(max_workers=5) as executor:
                executor.submit(insert_data, target_config, table_name, rows_dict)
                inserted_count += len(rows)

                # 打印当前进度
                print(f"已同步 {inserted_count} / {total_count} 条数据 ({(inserted_count / total_count) * 100:.2f}%)")

    except Exception as e:
        print(f"同步表 {table_name} 时出现错误: {e}")

    finally:
        source_cursor.close()
        source_db.close()

def sync_databases(source_config, target_config):
    source_db = pymysql.connect(**source_config)
    
    try:
        source_cursor = source_db.cursor()
        source_cursor.execute("SHOW TABLES;")
        tables = source_cursor.fetchall()

        for table in tables:
            table_name = table[0]
            print(f"开始同步表 {table_name}...")
            sync_table(source_config, target_config, table_name)
            print(f"完成表 {table_name} 的同步。")

    except Exception as e:
        print(f"数据库操作过程中出现错误: {e}")

    finally:
        source_cursor.close()
        source_db.close()

if __name__ == "__main__":
    source_config = {
        "host": "",   ip
        "user": "shield",  #账号
        "password": "",    #密码
        "port": 3306,
        "database": ""  # 源数据库名称（需要拉的数据库名称）
    }
    
    target_config = {
        "host": "",    #ip
        "user": "",     #账号
        "password": "",  # 密码
        "port": 3306,    
        "database": ""  # 目标数据库名称
    }
    
    sync_databases(source_config, target_config)
    send_dingtalk_notification(f"{target_config['database']} 同步完成")
```
如果出现缓存内存溢出执行

```
echo 1 > /proc/sys/vm/drop_caches
```



## mysql8

```
cat mysql-kaoshizhongxin.yaml 


version: '3.8'

services:
  mysql_kaoshi:
    image: mysql:8.0.33
    container_name: kaoshi_mysql_8.0.33
    restart: always
    environment:
      MYSQL_ROOT_PASSWORD: 111         # MySQL root 用户密码
      MYSQL_DATABASE: test                # 要创建的数据库名称
      MYSQL_USER: test                        # 普通用户
      MYSQL_PASSWORD: 111              # 普通用户的密码
      TZ: "Asia/Shanghai"                     # 设置时区为上海
    ports:
      - "23406:3306"                             # 将主机的 3306 端口映射到容器的 3306 端口
    volumes:
      - "/home/mysql/data:/var/lib/mysql"   
      - "/home/mysql/my.cnf:/etc/mysql/my.cnf"
      - "/home/mysql/config:/etc/mysql/conf.d"
    networks:
      - mysql_kaoshi_network                   # 使用定义的网络

networks:
  mysql_kaoshi_network:
    driver: bridge                         # 新建网络使用 bridge 驱动
```
### 创建用户

```
CREATE USER 'test'@'%' IDENTIFIED BY '111';
GRANT ALL PRIVILEGES ON *.* TO 'test'@'%' WITH GRANT OPTION;
FLUSH PRIVILEGES;
SELECT User, Host FROM mysql.user;
```
### 更新对外

```
UPDATE mysql.user SET Host='%' WHERE User='test' AND Host='localhost'
```
## my.cnf

``
# 默认存在地址
datadir=/var/lib/mysql
socket=/var/run/mysqld/mysqld.sock
```