最近折腾了一套自己的 AI 中转站，主要用到两个东西：New API 和 CLI Proxy API Plus。New API 负责统一管理渠道、模型、令牌、用户和日志；CLI Proxy API Plus 负责把一些 CLI/OAuth 类模型能力转换成 OpenAI 兼容接口（例如将chat gpt套餐转化成兼容接口提供给New API作为渠道使用）。以下是我自己整理的一套部署流程，尽量写成从零开始照着操作就能搭好的版本。 注意：文档里面的密码、密钥都用占位符，不要把真实密码直接写到博客里。功能描述简单来说，这套东西搭好后大概是这样：本地客户端 / Codex / 其他调用方 ↓ New API 统一入口 ↓ CLI Proxy API Plus ↓ 上游模型账号 / API Key / OAuth 登录态New API 主要做这些事：统一管理不同渠道。统一生成下游调用令牌。查看调用日志、用量、消耗。给不同用户、不同分组设置可用模型。对外暴露一个 OpenAI 兼容接口。CLI Proxy API Plus 主要做这些事：接收 OpenAI 兼容格式的请求。转发到 Codex、Claude、Gemini、OpenAI 兼容服务等上游。支持 API Key 和 OAuth 登录态。支持模型别名、模型转发、重试、管理面板等。我个人理解是： CLI Proxy API Plus 更像“底层转发器”，New API 更像“后台管理系统”。一、服务器准备我这里以一台普通云服务器为例，阿里云、腾讯云、雨云、RackNerd、Vultr 都差不多。1. 推荐配置如果只是自己用，配置不用太高：CPU：2 核 内存：2G 起步，4G 更舒服 硬盘：40G 起步 系统：Alibaba Cloud Linux / CentOS / Debian / Ubuntu 都可以 地区：尽量选择访问上游模型比较稳定的地区如果你后面要给多人使用，建议至少：CPU：2 核以上 内存：4G 以上 硬盘：60G 以上2. 需要开放的端口在云服务器安全组中开放以下端口：22 SSH 登录服务器 3000 New API 后台 8317 CLI Proxy API Plus 接口，按需开放 80 如果后面要配域名和 Nginx 443 如果后面要配 HTTPS注意：3306 MySQL 不建议对公网开放 6379 Redis 不建议对公网开放MySQL 和 Redis 给 Docker 内部用就行，不要随便暴露到公网。3. 登录服务器本地 Windows 可以用 Windows Terminal、XShell、FinalShell、Termius。SSH 登录方式如下：ssh root@你的服务器公网IP如果 SSH 端口不是 22，比如改成了 2222：ssh root@你的服务器公网IP -p 2222登录后大概会看到：[root@xxx ~]#这就说明已经进服务器了。二、安装基础环境先安装一些常用工具：yum install -y curl wget git vim unzip有些系统没有 yum，比如 Debian/Ubuntu，可以用：apt update apt install -y curl wget git vim unzip1. 安装 Docker直接使用官方安装脚本：curl -fsSL https://get.docker.com | bash启动 Docker：systemctl enable docker systemctl start docker检查 Docker 是否正常：docker version docker compose version能看到版本号就可以。2. 创建统一目录和 Docker 网络我建议把所有中转站相关的东西都放在 /opt/ai-gateway 下，后期找起来方便。mkdir -p /opt/ai-gateway cd /opt/ai-gateway创建一个公共 Docker 网络，让 New API 可以直接通过容器名访问 CLI Proxy API Plus：docker network create ai-gateway-network如果提示已经存在，不用管，继续下一步。三、部署 New APINew API 官方镜像是：calciumion/new-api:latest官方说明中支持 SQLite、MySQL、PostgreSQL，个人测试可以用 SQLite，但是长期使用建议直接 MySQL。 这里我用 MySQL + Redis + New API 的方式部署。1. 创建目录mkdir -p /opt/ai-gateway/new-api cd /opt/ai-gateway/new-api2. 创建 docker-compose.yml执行：vim docker-compose.yml写入以下内容：services: new-api: image: calciumion/new-api:latest container_name: new-api restart: always command: --log-dir /app/logs ports: - "3000:3000" volumes: - ./data:/data - ./logs:/app/logs environment: - TZ=Asia/Shanghai - SQL_DSN=root:你的MySQL密码@tcp(mysql:3306)/new-api - REDIS_CONN_STRING=redis://:你的Redis密码@redis:6379 - ERROR_LOG_ENABLED=true - BATCH_UPDATE_ENABLED=true - NODE_NAME=new-api-node-1 depends_on: - mysql - redis networks: - ai-gateway-network healthcheck: test: ["CMD-SHELL", "wget -q -O - http://localhost:3000/api/status | grep -o '\"success\":\\s*true' || exit 1"] interval: 30s timeout: 10s retries: 3 mysql: image: mysql:8.2 container_name: new-api-mysql restart: always environment: MYSQL_ROOT_PASSWORD: 你的MySQL密码 MYSQL_DATABASE: new-api volumes: - mysql_data:/var/lib/mysql networks: - ai-gateway-network redis: image: redis:latest container_name: new-api-redis restart: always command: ["redis-server", "--requirepass", "你的Redis密码"] networks: - ai-gateway-network volumes: mysql_data: networks: ai-gateway-network: external: true注意几点：1. docker-compose.yml 对缩进非常敏感，mysql、redis、new-api 必须同级缩进。 2. SQL_DSN 里的数据库名是 new-api，要和 MYSQL_DATABASE 保持一致。 3. 密码建议用复杂一点，但不要包含 @、#、&、?、/ 这种容易影响连接串解析的字符。 4. 如果密码里一定要用特殊符号，需要做 URL 编码，不然 New API 可能连不上 MySQL。3. 启动 New APIdocker compose up -d查看容器：docker ps查看日志：docker logs -f new-api如果没报错，浏览器访问：http://你的服务器公网IP:30004. 登录 New API常见默认账号密码：用户名：root 密码：123456如果默认密码不对，就按页面提示注册或初始化。第一次登录后先做三件事：1. 修改 root 密码 2. 关闭不需要的公开注册 3. 检查系统设置和站点名称New API 的登录页也可以直接访问：http://你的服务器公网IP:3000/login四、部署 CLI Proxy API PlusCLI Proxy API Plus 是一个 OpenAI 兼容代理，官方文档里默认端口是 8317，配置文件是 config.yaml。1. 创建目录mkdir -p /opt/ai-gateway/cli-proxy-api-plus cd /opt/ai-gateway/cli-proxy-api-plus2. 准备配置文件先生成两个密钥：openssl rand -hex 32 openssl rand -hex 32第一个作为：API_KEY第二个作为：MANAGEMENT_KEY创建配置文件：vim config.yaml写入：host: "" port: 8317 auth-dir: "/root/.cli-proxy-api" api-keys: - "替换成你的API_KEY" remote-management: allow-remote: true secret-key: "替换成你的MANAGEMENT_KEY" disable-control-panel: false debug: false request-retry: 3 routing: strategy: "round-robin" # 如果服务器访问上游模型不稳定，可以配置代理。 # 没有代理就保持为空。 proxy-url: "" # 示例：如果你有传统 API Key，可以写到这里。 # 没有就先注释掉，后面通过 OAuth 登录。 # # claude-api-key: # - api-key: "sk-ant-xxx" # # gemini-api-key: # - api-key: "AIzaSyxxx" # # openai-compatibility: # - name: "openrouter" # base-url: "https://openrouter.ai/api/v1" # api-key-entries: # - api-key: "sk-or-v1-xxx" # models: # - name: "openai/gpt-4o-mini" # alias: "gpt-4o-mini"注意：API_KEY 是给 New API 调用 CLI Proxy API Plus 用的。 MANAGEMENT_KEY 是进 CLI Proxy API Plus 管理接口用的。 这两个不要写成一样。3. 创建 docker-compose.ymlvim docker-compose.yml写入：services: cli-proxy-api-plus: image: eceasy/cli-proxy-api-plus:latest container_name: cli-proxy-api-plus restart: unless-stopped ports: - "127.0.0.1:18317:8317" volumes: - ./config.yaml:/CLIProxyAPI/config.yaml - ./auths:/root/.cli-proxy-api - ./logs:/CLIProxyAPI/logs networks: - ai-gateway-network networks: ai-gateway-network: external: true这里我把容器的 8317 映射到了服务器本机的 127.0.0.1:18317。原因是：1. New API 和 CLI Proxy API Plus 在同一个 Docker 网络里，可以直接访问 cli-proxy-api-plus:8317。 2. 服务器公网不直接暴露 8317，安全一点。 3. 如果后续确实要公网访问，再用 Nginx 单独代理。启动：docker compose up -d查看日志：docker logs -f cli-proxy-api-plus测试接口：curl -H "Authorization: Bearer 替换成你的API_KEY" \ http://127.0.0.1:18317/v1/models如果返回模型列表，说明 CLI Proxy API Plus 已经起来了。4. 如果镜像拉取失败有时候会遇到这种报错：pull access denied for eceasy/cli-proxy-api-plus这时候可以改成源码构建：cd /opt/ai-gateway git clone https://github.com/KooshaPari/cliproxyapi-plusplus.git cd cliproxyapi-plusplus docker build -t local/cli-proxy-api-plus:latest .然后把 /opt/ai-gateway/cli-proxy-api-plus/docker-compose.yml 里的镜像改成：image: local/cli-proxy-api-plus:latest再启动：cd /opt/ai-gateway/cli-proxy-api-plus docker compose up -d五、CLI Proxy API Plus 登录上游账号CLI Proxy API Plus 有两种上游方式：1. 直接填 API Key 2. 使用 OAuth 登录如果你有传统 API Key，直接写进 config.yaml 最简单。如果你要接 Codex、Claude Code、Gemini CLI 这类账号登录态，就需要运行登录命令。1. Codex / OpenAI OAuth 登录cd /opt/ai-gateway/cli-proxy-api-plus docker compose exec cli-proxy-api-plus /CLIProxyAPI/CLIProxyAPI -no-browser --codex-login它会输出一个登录地址。复制到浏览器打开，登录你的账号，完成授权后回到命令行看是否成功。2. Claude 登录cd /opt/ai-gateway/cli-proxy-api-plus docker compose exec cli-proxy-api-plus /CLIProxyAPI/CLIProxyAPI -no-browser --claude-login3. Gemini 登录cd /opt/ai-gateway/cli-proxy-api-plus docker compose exec cli-proxy-api-plus /CLIProxyAPI/CLIProxyAPI -no-browser --login注意：登录成功后，认证信息会保存在 ./auths 目录。 这个目录很重要，不要随便删除。 如果删除了，就要重新登录。六、把 CLI Proxy API Plus 接入 New API进入 New API 后台：http://你的服务器公网IP:3000进入：渠道管理 -> 添加渠道基础配置参考：类型：OpenAI 名称：CLI Proxy API Plus 密钥：CLI Proxy API Plus 配置中的 API_KEY API地址：http://cli-proxy-api-plus:8317/v1这里重点注意：如果 New API 和 CLI Proxy API Plus 在同一个 ai-gateway-network 里， API 地址就写： http://cli-proxy-api-plus:8317/v1 不要写： http://127.0.0.1:8317/v1因为在 New API 容器里面，127.0.0.1 指的是 New API 自己，不是宿主机，也不是 CLI Proxy API Plus。1. 模型怎么填模型必须填 CLI Proxy API Plus 实际支持的模型名。比如 CLI Proxy API Plus 里能返回：gpt-5 gpt-5-codex gpt-5.1 gpt-5.2那 New API 里就填这些模型。不要凭感觉乱填，比如：gpt-5.5-pro如果上游没有这个模型，调用时会报：unknown provider for model xxx这种 502 一般不是 New API 坏了，而是模型名没有在 CLI Proxy API Plus 里配置成功。2. 测试渠道保存渠道后，点测试。如果测试成功，就可以去：令牌管理 -> 新建令牌创建一个下游 API Key。之后客户端就可以调用 New API：curl http://你的服务器公网IP:3000/v1/chat/completions \ -H "Content-Type: application/json" \ -H "Authorization: Bearer 你的NewAPI令牌" \ -d '{ "model": "gpt-5", "messages": [ { "role": "user", "content": "你好" } ] }'七、如果要让 CLI Proxy API Plus 外网访问一般不建议直接开放 CLI Proxy API Plus。 如果只是给 New API 调用，不需要开放公网。如果你确实要在外网直接访问，可以用 Nginx 代理，并加一层 Basic Auth。1. 安装 Nginxyum install -y nginx httpd-tools systemctl enable nginx systemctl start nginx2. 创建 Basic Auth 密码htpasswd -bc /etc/nginx/.cliproxy_htpasswd admin 你的访问密码3. 添加 Nginx 配置vim /etc/nginx/conf.d/cli-proxy-api-plus.conf写入：server { listen 8317; server_name _; client_max_body_size 50m; location / { auth_basic "CLI Proxy API Plus"; auth_basic_user_file /etc/nginx/.cliproxy_htpasswd; proxy_pass http://127.0.0.1:18317; proxy_http_version 1.1; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; proxy_read_timeout 600s; proxy_send_timeout 600s; } }检查 Nginx 配置：nginx -t重载：systemctl reload nginx然后安全组开放 8317。访问：http://你的服务器公网IP:8317会先要求输入 Basic Auth 账号密码。八、客户端怎么用1. 普通 OpenAI 兼容客户端如果客户端支持填写 Base URL 和 API Key，就这样填：Base URL：http://你的服务器公网IP:3000/v1 API Key：New API 里创建的令牌 模型：New API 里配置好的模型名2. Codex CLI 直接接 CLI Proxy API Plus如果你不想走 New API，想让 Codex 直接连 CLI Proxy API Plus，可以改本地：~/.codex/config.toml示例：model_provider = "cliproxyapi" model = "gpt-5-codex" model_reasoning_effort = "high" [model_providers.cliproxyapi] name = "cliproxyapi" base_url = "http://你的服务器公网IP:8317/v1" wire_api = "responses"再改：~/.codex/auth.json示例：{ "OPENAI_API_KEY": "sk-dummy" }如果你走的是 New API，则可以把 Codex 的 base_url 改成：http://你的服务器公网IP:3000/v1然后 OPENAI_API_KEY 填 New API 创建的令牌。九、常用维护命令1. 查看容器docker ps2. 查看 New API 日志docker logs -f new-api3. 查看 CLI Proxy API Plus 日志docker logs -f cli-proxy-api-plus4. 重启 New APIcd /opt/ai-gateway/new-api docker compose restart5. 重启 CLI Proxy API Pluscd /opt/ai-gateway/cli-proxy-api-plus docker compose restart6. 更新镜像New API：cd /opt/ai-gateway/new-api docker compose pull docker compose up -dCLI Proxy API Plus：cd /opt/ai-gateway/cli-proxy-api-plus docker compose pull docker compose up -d如果 CLI Proxy API Plus 是源码构建的：cd /opt/ai-gateway/cliproxyapi-plusplus git pull docker build -t local/cli-proxy-api-plus:latest . cd /opt/ai-gateway/cli-proxy-api-plus docker compose up -d7. 备份建议至少备份这些目录：/opt/ai-gateway/new-api /opt/ai-gateway/cli-proxy-api-plus/config.yaml /opt/ai-gateway/cli-proxy-api-plus/auths打包备份：cd /opt tar -czvf ai-gateway-backup.tar.gz ai-gateway下载到本地保存。十、常见问题1. docker compose up -d 报 YAML 错误常见原因：1. 缩进不对。 2. mysql、redis 没有和 new-api 对齐。 3. 冒号后面少空格。 4. 密码里有特殊符号，导致解析异常。可以用：docker compose config检查 compose 文件是否能被正常解析。2. New API 打不开排查：docker ps docker logs -f new-api再检查安全组是否开放 3000。如果容器正常，但公网打不开，基本就是安全组或服务器防火墙问题。3. New API 连接不上 MySQL重点看 SQL_DSN：root:你的MySQL密码@tcp(mysql:3306)/new-api注意：mysql 是容器名，不是服务器 IP。 new-api 是数据库名。 密码不要乱带特殊符号。4. New API 渠道测试 502如果报：unknown provider for model xxx通常是模型名不对。处理方式：1. 先请求 CLI Proxy API Plus 的 /v1/models。 2. 看实际返回了哪些模型。 3. New API 里只填写真实存在的模型。5. CLI Proxy API Plus 登录后还是不能用检查：ls -la /opt/ai-gateway/cli-proxy-api-plus/auths docker logs -f cli-proxy-api-plus如果 auths 目录没有生成登录态文件，说明登录没有成功。重新执行对应登录命令即可。6. 服务器在新加坡，要不要开代理这个要看你的上游服务。如果服务器访问上游模型经常超时，可以在 config.yaml 里配置：proxy-url: "socks5://127.0.0.1:7890"如果服务器访问稳定，就不要多加代理，代理链路越长，越容易慢。7. MySQL 要不要开放 3306不建议。New API 和 MySQL 在同一个 Docker 网络里，内部访问即可：mysql:3306公网开放 3306 风险很高，除非你明确知道自己在做什么。十一、最终访问地址整理部署完成后常用地址：服务器登录： ssh root@你的服务器公网IP New API 后台： http://你的服务器公网IP:3000 New API 接口： http://你的服务器公网IP:3000/v1 CLI Proxy API Plus 内部地址： http://cli-proxy-api-plus:8317/v1 CLI Proxy API Plus 宿主机本地地址： http://127.0.0.1:18317/v1 CLI Proxy API Plus 外网地址，只有配置 Nginx 后才有： http://你的服务器公网IP:8317/v1登录方式：服务器：SSH 用户名 + 密码，或者 SSH 密钥 New API：root / 初始密码，首次登录后必须修改 CLI Proxy API Plus：API 调用使用 Bearer API_KEY，管理接口使用 MANAGEMENT_KEY Nginx 外网访问：Basic Auth 用户名 + 密码十二、我的建议如果只是自己用，推荐：New API 对公网开放 3000 CLI Proxy API Plus 不直接对公网开放 New API 通过 Docker 网络访问 CLI Proxy API Plus MySQL / Redis 不开放公网这样最清晰，也相对安全。如果后面要给公司同事或者多台电脑用，再考虑：1. 给 New API 配域名和 HTTPS 2. 关闭公开注册 3. 每个人单独创建 New API 用户和令牌 4. 给不同用户限制模型和额度 5. 定期看日志和异常请求这个东西最怕的不是部署不起来，而是部署起来后所有端口和密钥都裸奔。 能不暴露的端口就不要暴露，能用内部网络访问的就不要走公网。参考资料New API Docker 镜像说明：https://hub.docker.com/r/calciumion/new-apiNew API 环境变量说明：https://doc.newapi.pro/installation/environment-variables/New API 注册与登录说明：https://docs.newapi.pro/zh/docs/guide/feature-guide/user/authNew API docker-compose 示例：https://github.com/QuantumNous/new-api/blob/main/docker-compose.ymlCLI Proxy API Plus GitHub：https://github.com/KooshaPari/cliproxyapi-plusplusCLIProxyAPI Docker Compose 说明：https://help.router-for.me/cn/docker/docker-composeCLIProxyAPI 配置项说明：https://help.router-for.me/configuration/optionsCLIProxyAPI Codex 配置说明：https://help.router-for.me/cn/agent-client/codex
最近又来了个棘手的项目，时间紧任务重，要基于现有的软件，把MySQL数据库完全复制到Oracle数据库，查了相关数据库迁移的资料，最后决定使用dbswitch，以下是我使用时的一些经验。gitee地址：dbswitch功能描述一句话，dbswitch工具提供源端数据库向目的端数据库的批量迁移同步功能，支持数据的全量和增量方式同步。包括：结构迁移支持字段类型、主键信息、建表语句等的转换，并生成建表SQL语句。支持基于正则表达式转换的表名与字段名映射转换。数据同步。基于JDBC的分批次读取源端数据库数据，并基于insert/copy方式将数据分批次写入目的数据库。支持有主键表的 增量变更同步 （变化数据计算Change Data Calculate）功能(千万级以上数据量的性能尚需在生产环境验证)使用步骤及注意事项1.直接拉取代码，或者下载zip包到桌面2.双击build.cmd脚本文件即可编译打包JDK:>=1.8 （建议用JDK 1.8）maven:>=3.6要配置对应的环境遍历，按win+r，打开cmd，输入mvn -v可以出现maven版本号即可3.打包完成后，在同目录（target文件夹）下会出现：dbswitch-release-1.9.9.tar.gz压缩包，直接解压4.编辑dbswitch-release-1.9.9\conf文件夹下的application.yml文件server: port: 9088 spring: application: name: dbswitch-admin tomcat: uri-encoding: UTF-8 max-http-header-size: 8096 mvc: throw-exception-if-no-handler-found: false static-path-pattern: /statics/** datasource: driver-class-name: com.mysql.jdbc.Driver url: jdbc:mysql://你自己的数据库ip地址:3306/zhwy_fx?createDatabaseIfNotExist=true&useUnicode=true&characterEncoding=UTF8&autoReconnect=true&useSSL=false&allowMultiQueries=true&failOverReadOnly=false&connectTimeout=30000 username: 账号 password: 密码 validation-query: SELECT 1 test-on-borrow: true flyway: locations: classpath:db/migration baseline-on-migrate: true table: DBSWITCH_SCHEMA_HISTORY enabled: true mybatis: configuration: lazy-loading-enabled: true aggressive-lazy-loading: false map-underscore-to-camel-case: true #log-impl: org.apache.ibatis.logging.stdout.StdOutImpl #注意此处地址就是解压后的dbswitch-release-1.9.9文件夹中drivers的地址 dbswitch: configuration: drivers-base-path: C:\Users\22727\Desktop\dbswitch-release-1.9.9\drivers注意配置数据库的url，username，password，还有驱动包的地址drivers-base-path5.找到bin目录中的startup.cmd文件，双击启动即可如果启动的时候报错，说你配置的数据库中缺少什么表（不影响启动），就打开如下文件复制sql执行下6.启动成功后在浏览器中输入地址：http://localhost:9088，默认账号密码：admin/1234567.配置连接将目标库和源库信息添加进去以下是我的相关配置可作为参考（MySQL库导入到Oracle库）源库（MySQL）配置：目标库（Oracle）配置：注意红框中内容，别写错了，配置完后可以测试下是否连接成功，如果没找到自己的驱动，也可以在【连接管理-驱动配置】中添加8.创建同步任务在【数据接入-任务管理】中创建任务，我的配置如下：在配置目标端时，注意在【同步前置执行SQL脚本】中执行以下内容：CREATE PROCEDURE RemoveNotNullConstraintsExceptPrimaryKey() BEGIN DECLARE done INT DEFAULT FALSE; DECLARE tbl_name VARCHAR(255); DECLARE cur CURSOR FOR SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE(); DECLARE CONTINUE HANDLER FOR NOT FOUND SET done = TRUE; OPEN cur; read_loop: LOOP FETCH cur INTO tbl_name; IF done THEN LEAVE read_loop; END IF; -- 构建 ALTER 语句，排除主键列 SET @alter_sql = ( SELECT GROUP_CONCAT(CONCAT('MODIFY ', COLUMN_NAME, ' ', COLUMN_TYPE, ' NULL')) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = tbl_name AND IS_NULLABLE = 'NO' AND COLUMN_KEY != 'PRI' ); -- 执行修改表结构的 SQL IF @alter_sql IS NOT NULL THEN SET @full_alter_sql = CONCAT('ALTER TABLE ', tbl_name, ' ', @alter_sql); PREPARE stmt FROM @full_alter_sql; EXECUTE stmt; DEALLOCATE PREPARE stmt; END IF; END LOOP; CLOSE cur; END // DELIMITER ; -- 调用存储过程 SET FOREIGN_KEY_CHECKS = 0; CALL RemoveNotNullConstraintsExceptPrimaryKey(); 9.启动！然后就是等待执行即可，我这里基本上都执行完了，执行异常是因为我有个表的名称超长了，Oracle11中表名长度不能超过30个字符。出现异常可以点开日志进行排查（重要）
没想到在2024年，我还要用这种方式去更新程序，其中的辛酸和踩过的坑就不多说了。这里推荐使用是winRar解压缩工具，我也试过其他解压缩工具，例如BandiZip，但都没效果（可能是我不会操作）准备jar包解压jar先将jar包的后缀名改为.zip,得到压缩包如图解压压缩包解压后得到文件夹：这个时候可以根据需要替换或者修改其中的文件，之后再将它打包为jar重新打包为jar在文件夹内部 全部选中文件然后打包，注意是文件夹内部，不然打出来的包可能会多出来一层打包配置注意：如果原来的jar包中依赖的有其他的jar,如图：需要将这些依赖的jar包排除在压缩文件队列中，不然会造成二次压缩：压缩后得到一个zip格式的压缩包：直接改后缀为.jar这样就可以运行了错误示例：依赖的jar被重复压缩会出现下面的报错
什么是MQ？MQ全称为Message Queue，即消息队列。“消息队列”是在消息的传输过程中保存消息的容器。也是在消息的传输过程中保存消息的容器。用于分布式系统之间进行通信。典型的：生产者、消费者模型。MQ本身就相当于一个中间件，用于生产方通过MQ与消费方之间的交互作用。RabbitMQ应用场景场景一 、解耦上面我们提到MQ本身就相当于一个中间件，用于生产方通过MQ与消费方之间的交互作用。让我们看一下传统模式和这种中间件模式的对比：传统模式：系统间耦合性太强，如图所示，系统A在代码中直接调用系统B和系统C的代码，如果将来D系统接入，系统A还需要修改代码，过于麻烦！中间件模式：​ 将消息写入消息队列，需要消息的系统自己从消息队列中订阅，从而系统A不需要做任何修改。场景二 、异步传统模式:​ 一些非必要的业务逻辑以同步的方式运行，太耗费时间中间件模式:​ 将消息写入消息队列，非必要的业务逻辑以异步的方式运行，加快响应速度场景三、削峰传统模式：​ 并发量大的时候，所有的请求直接怼到数据库，造成数据库连接异常中间件模式:​ 系统A慢慢的按照数据库能处理的并发量，从消息队列中慢慢拉取消息RabbitMQ的使用RabbitMQ是一个消息代理：它接受和转发消息。 你可以把它想象成一个邮政信箱RabbitMQ与邮局的主要区别是它不处理纸张，而是接受，存储和转发数据消息的二进制数据块。如何避免消息堆积？1） 采用workqueue，多个消费者监听同一队列。2）接收到消息以后，而是通过线程池，异步消费。如何避免消息丢失？1） 消费者的ACK机制。可以防止消费者丢失消息。但是，如果在消费者消费之前，MQ就宕机了，消息就没了？2）可以将消息进行持久化。要将消息持久化，前提是：队列、Exchange都持久化RabbitMQ提供了6种消息模型，但是第6种是RPC，并不是MQ，因此不学习。那么也就剩下5种，3、4、5这三种都属于订阅模型，只不过进行路由的方式不同。1.Simple-简单模型P（producer/ publisher）：生产者，一个发送消息的用户应用程序。 C（consumer）：消费者，消费和接收有类似的意思，消费者是一个主要用来等待接收消息的用户应用程序队列（红色区域）：rabbitmq内部类似于邮箱的一个概念。虽然消息流经rabbitmq和你的应用程序，但是它们只能存储在队列中。队列只受主机的内存和磁盘限制，实质上是一个大的消息缓冲区。许多生产者可以发送消息到一个队列，许多消费者可以尝试从一个队列接收数据。引入依赖我们将用Java编写两个程序;发送单个消息的生产者，以及接收消息并将其打印出来的消费者。我们将详细介绍Java API中的一些细节，这是一个消息传递的“Hello World”。我们将调用我们的消息发布者（发送者）Send和我们的消息消费者（接收者）Recv。发布者将连接到RabbitMQ，发送一条消息，然后退出。新建一个spring boot项目,使用spring 2和jdk 8,引入下面的依赖 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> <version>3.1.5</version> </dependency>新建一个连接工具类:package com.test.demo.util; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; /** * rabbitMQ连接类 * * @author jg * @since 2023/11/7 15:27 */ public class ConnectionUtil { /** * 建立与RabbitMQ的连接 * * @return Connection * @throws Exception 异常 */ public static Connection getConnection() throws Exception { //定义连接工厂 ConnectionFactory factory = new ConnectionFactory(); //设置服务地址 factory.setHost("你的IP地址"); //端口,rabbitmq默认端口号为5672,阿里云等服务器记得开启防火墙和安全组 factory.setPort(5672); //设置账号信息，用户名、密码、vhost factory.setUsername("username"); factory.setPassword("password"); //虚拟主机路径,不同虚拟主机之间的数据不共享 factory.setVirtualHost("/"); // 通过工程获取连接 Connection connection = factory.newConnection(); return connection; } }生产者创建一个send类,充当简单模型生产者来发送消息package com.test.demo.util.simple; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.test.demo.util.ConnectionUtil; /** * 简单模式---生产者 * * @author jg * @since 2023/11/7 15:33 */ public class Send { private final static String QUEUE_NAME = "simple_queue"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); // 轻量级的 Connection，这是完成大部分API的地方。 Channel channel = connection.createChannel(); // 声明（创建）队列，必须声明队列才能够发送消息，我们可以把消息发送到队列中。 // 声明一个队列是幂等的 - 只有当它不存在时才会被创建 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 消息内容 String message = "Hello World!"; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); //关闭通道和连接 channel.close(); connection.close(); } }运行之后控制台会出现:这时候我们打开rabbitMQ的web页面（默认端口号：15672）,登陆上去查看队列中的信息:这就是我们刚才发的消息,在页面的控制台查看消息是不会进行消费的,所以消息还在消费者创建消费者来获取消息package com.test.demo.util.simple; import com.rabbitmq.client.*; import com.test.demo.util.ConnectionUtil; import java.io.IOException; /** * 简单模型---消费者 * 自动ACK：消息一旦被接收，消费者自动发送ACK * * @author jg * @since 2023/11/7 15:56 */ public class Recv { private final static String QUEUE_NAME = "simple_queue"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 创建通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); //添加异常，测试自动ACK //int i = 1 / 0; System.out.println(" [x] received : " + msg + "!"); } }; // 监听队列，第二个参数：是否自动进行消息确认。 channel.basicConsume(QUEUE_NAME, false, consumer); } }运行之后控制台会出现:这时,我们再点开rabbitmq的web页面就会发现刚才发的消息没有了因为我们并没有关闭消费者的连接,所以消费者是一直可以接收生产者发送的消息,每调用一次生产者Send发送消息,消费者Recv的控制台都会出现一条信息消息确认机制（ACK）通过刚才的案例可以看出，消息一旦被消费者接收，队列中的消息就会被删除。那么问题来了：RabbitMQ怎么知道消息被接收了呢？如果消费者领取消息后，还没执行操作就挂掉了呢？或者抛出了异常？消息消费失败，但是RabbitMQ无从得知，这样消息就丢失了！因此，RabbitMQ有一个ACK机制。当消费者获取消息后，会向RabbitMQ发送回执ACK，告知消息已经被接收。不过这种回执ACK分两种情况：自动ACK：消息一旦被接收，消费者自动发送ACK手动ACK：消息接收后，不会发送ACK，需要手动调用选择哪种ACK主要看消息的重要性：如果消息不太重要，丢失也没有影响，那么自动ACK会比较方便如果消息非常重要，不容丢失。那么最好在消费完成后手动ACK，否则接收消息后就自动ACK，RabbitMQ就会把消息从队列中删除。如果此时消费者宕机，那么消息就丢失了。测试自动ACK还是拿简单模式举例:在简单模式的消费者Recv中添加代码,引发异常:注意最后一行代码,第二个参数为true时,就会自动进行ACKchannel.basicConsume(QUEUE_NAME, true, consumer);修改完成之后我们重启消费者Recv,然后启动生产者发送一条信息可以看到控制台报错,但是rabbitMQ的队列中消息依旧被消费了测试手动ACK我们注释掉引发报错的代码,并且把basicConsume方法第二个参数改为false(手动ACK)先重启Recv然后用生产者发送一条信息可以看到这里有一条未进行ACK消息确认的信息.然后我们把Recv消费者停掉,控制台就会出现以下变化这是因为我们没有对消息进行ACK,所以消息没有被消费掉,在消费者宕机之后,消息就会再次进入ready状态现在我们修改代码,进行手动ACK /** * 手动进行ACK * deliveryTag：该消息的index * multiple：是否批量处理.true:将一次性ack所有消息。 */ channel.basicAck(envelope.getDeliveryTag(), false);改完代码重新启动消费者Recv,打开控制台就会发现消息已经没了:消息手动ACK成功!2.Work-工作模型工作队列或者竞争消费者模式工作队列，又称任务队列。主要思想就是避免执行资源密集型任务时，必须等待它执行完成。相反我们稍后完成任务，我们将任务封装为消息并将其发送到队列。 在后台运行的工作进程将获取任务并最终执行作业。当你运行许多消费者时，任务将在他们之间共享，但是一个消息只能被一个消费者获取。这个概念在Web应用程序中特别有用，因为在短的HTTP请求窗口中无法处理复杂的任务。假设现在有以下场景:P：生产者：任务的发布者C1：消费者，领取任务并且完成任务，假设完成速度较快C2：消费者2：领取任务并完成任务，假设完成速度慢生产者跟简单模型基本一样,不过这里是循环发送50条消息。package com.test.demo.util.work; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.test.demo.util.ConnectionUtil; /** * work模型---生产者 * 工作模型：一个生产者对应多个消费者，一个消息只能被一个消费者消费 * @author jg * @since 2023/11/7 16:30 */ public class Send { private final static String QUEUE_NAME = "work_queue"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 循环发布任务 for (int i = 0; i < 50; i++) { // 消息内容 String message = "task .. " + i; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); Thread.sleep(i * 2); } // 关闭通道和连接 channel.close(); connection.close(); } }消费者1package com.test.demo.util.work; import com.rabbitmq.client.*; import com.test.demo.util.ConnectionUtil; import java.io.IOException; /** * 工作模型消费者1 * @author jg * @since 2023/11/7 16:32 */ public class Recver1 { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); String QUEUE_NAME = "work_queue"; channel.queueDeclare(QUEUE_NAME,false,false,false, null); //保证一次只分发一个 channel.basicQos(1); //接收消息 DefaultConsumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { try { String msg = new String(body); System.out.println("Recver1：" + msg); channel.basicAck(envelope.getDeliveryTag(),false); Thread.sleep(2000); }catch (Exception e){ e.printStackTrace(); } } }; channel.basicConsume(QUEUE_NAME,false, consumer); } }消费者2package com.test.demo.util.work; import com.rabbitmq.client.*; import com.test.demo.util.ConnectionUtil; import java.io.IOException; /** * 工作模型消费者2 * * @author jg * @since 2023/11/7 16:33 */ public class Recver2 { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); String QUEUE_NAME = "work_queue"; channel.queueDeclare(QUEUE_NAME,false,false,false, null); //保证一次只分发一个 channel.basicQos(1); //接收消息 DefaultConsumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { try { String msg = new String(body); System.out.println("Recver2：" + msg); channel.basicAck(envelope.getDeliveryTag(),false); }catch (Exception e){ e.printStackTrace(); } } }; channel.basicConsume(QUEUE_NAME,false, consumer); } }消费者2与消费者1基本类似，就是没有设置消费耗时时间。这里是模拟有些消费者快，有些比较慢。接下来，两个消费者一同启动，然后发送50条消息：可以发现，两个消费者各自消费了25条消息，而且各不相同，这就实现了任务的分发。刚才的实现有问题吗？消费者1比消费者2的效率要低，一次任务的耗时较长然而两人最终消费的消息数量是一样的消费者2大量时间处于空闲状态，消费者1一直忙碌 现在的状态属于是把任务平均分配，正确的做法应该是消费越快的人，消费的越多。 怎么实现呢？ 我们可以使用basicQos方法和prefetchCount = 1设置。 这告诉RabbitMQ一次不要向工作人员发送多于一条消息。 或者换句话说，不要向工作人员发送新消息，直到它处理并确认了前一个消息。 相反，它会将其分派给不是仍然忙碌的下一个工作人员。 两个消费者对象中都需添加://设置每个消费者同时只能处理一条信息 channel.basicQos(1);改完之后再次测试就会发现消费者2处理了47条信息,消费者1实现了3条信息:这就实现能者多劳了.3.Fanout-广播模型在广播模式下，消息发送流程是这样的：1） 可以有多个消费者2） 每个消费者有自己的queue（队列）3） 每个队列都要绑定到Exchange（交换机）4） 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。5） 交换机把消息发送给绑定过的所有队列6） 队列的消费者都能拿到消息。实现一条消息被多个消费者消费生产者跟简单模型和工作模型相比有两个变化：1） 声明Exchange，不再声明Queue2） 发送消息到Exchange，不再发送到Queuepackage com.test.demo.util.fanout; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.test.demo.util.ConnectionUtil; /** * 广播模式--生产者 * 1） 声明Exchange，不再声明Queue * 2） 发送消息到Exchange，不再发送到Queue * * @author jg * @since 2023/11/8 10:35 */ public class Send { private final static String EXCHANGE_NAME = "fanout_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明exchange，指定类型为fanout channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); // 消息内容 String message = "Hello everyone"; // 发布消息到Exchange channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes()); System.out.println(" [生产者] Sent '" + message + "'"); channel.close(); connection.close(); } }消费者1package com.test.demo.util.fanout; import com.rabbitmq.client.*; import com.test.demo.util.ConnectionUtil; import java.io.IOException; /** * 广播模式--消费者1 * 代码中：队列需要和交换机绑定 * @author jg * @since 2023/11/8 10:36 */ public class Recv { private final static String QUEUE_NAME = "fanout_exchange_queue_1"; private final static String EXCHANGE_NAME = "fanout_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [消费者1] received : " + msg + "!"); } }; // 监听队列，自动返回完成 channel.basicConsume(QUEUE_NAME, true, consumer); } }消费者2package com.test.demo.util.fanout; import com.rabbitmq.client.*; import com.test.demo.util.ConnectionUtil; import java.io.IOException; /** * 广播模式--消费者2 * 代码中：队列需要和交换机绑定 * @author jg * @since 2023/11/8 10:36 */ public class Recv2 { private final static String QUEUE_NAME = "fanout_exchange_queue_2"; private final static String EXCHANGE_NAME = "fanout_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [消费者2] received : " + msg + "!"); } }; // 监听队列，自动返回完成 channel.basicConsume(QUEUE_NAME, true, consumer); } }因为我们这里是先声明交换机,然后再声明队列,所以我们要先启动生产者Send,然后再启动两个消费者,否则会报错测试结果启动完成后我们重新运行生产者Send发送一条信息,可以看到两个消费者的控制台出现:这就说明广播模式创建成功了广播模式是将消息发送给交换机下所有的队列,不用选择,所以速度也是最快的4.Direct-定向模型Direct-定向模型有选择性的接收消息在订阅模式中，生产者发布消息，所有消费者都可以获取所有消息。在路由模式中，我们将添加一个功能 - 我们将只能订阅一部分消息。 例如，我们只能将重要的错误消息引导到日志文件（以节省磁盘空间），同时仍然能够在控制台上打印所有日志消息。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。在Direct模型下，队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）消息的发送方在向Exchange发送消息时，也必须指定消息的routing key。P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列C1：消费者，其所在队列指定了需要routing key 为 error 的消息C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息生产者package com.test.demo.util.direct; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.test.demo.util.ConnectionUtil; /** * 路由（定向）模式--生产者 * * @author jg * @since 2023/11/8 13:51 */ public class Send { private final static String EXCHANGE_NAME = "direct_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明exchange，指定类型为direct channel.exchangeDeclare(EXCHANGE_NAME, "direct"); // 消息内容 String message = "商品新增了， id = 1001"; String message2 = "商品修改了， id = 1002"; String message3 = "商品删除了， id = 1003"; // 发送消息，并且指定routing key 为：insert ,代表新增商品 channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes()); channel.basicPublish(EXCHANGE_NAME, "update", null, message2.getBytes()); channel.basicPublish(EXCHANGE_NAME, "delete", null, message3.getBytes()); System.out.println(" [商品服务：] Sent '" + message + "'"); channel.close(); connection.close(); } }消费者1package com.test.demo.util.direct; import com.rabbitmq.client.*; import com.test.demo.util.ConnectionUtil; import java.io.IOException; /** * 定向模式--消费者1 * * @author jg * @since 2023/11/8 13:52 */ public class Recv { private final static String QUEUE_NAME = "direct_exchange_queue_1"; private final static String EXCHANGE_NAME = "direct_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机，同时指定需要订阅的routing key。假设此处需要update和delete消息 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete"); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [消费者1] received : " + msg + "!"); } }; // 监听队列，自动ACK channel.basicConsume(QUEUE_NAME, true, consumer); } }消费者2package com.test.demo.util.direct; import com.rabbitmq.client.*; import com.test.demo.util.ConnectionUtil; import java.io.IOException; /** * 定向模式--消费者2 * * @author jg * @since 2023/11/8 13:52 */ public class Recv2 { private final static String QUEUE_NAME = "direct_exchange_queue_2"; private final static String EXCHANGE_NAME = "direct_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机，同时指定需要订阅的routing key。订阅 insert、update、delete channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete"); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [消费者2] received : " + msg + "!"); } }; // 监听队列，自动ACK channel.basicConsume(QUEUE_NAME, true, consumer); } }还是先启动Send来创建交换机Exchange,然后再启动两个消费者测试结果我们分别发送增、删、改的RoutingKey，发现结果：定向模型的路由是全匹配的,不能带.或者#等符号,否则会获取不到消息5.Topic-主题模型Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert通配符规则：#：匹配一个或多个词*：匹配不多不少恰好1个词举例:audit.#：能够匹配:audit.irs.corporate` 或者 `audit.irs audit.*：只能匹配:audit.irs *.audit.*：只能匹配:test.audit.irs我们创建了两个绑定：Q1绑定了绑定键“*.orange.*”，Q2绑定了“*.*.rabbit”和“lazy.＃”。 Q1匹配所有的橙色动物。 Q2匹配关于兔子以及懒惰动物的消息。生产者发送如下消息，会进入哪个队列?quick.orange.rabbit--Q1lazy.orange.elephant--Q2quick.orange.fox--Q1lazy.pink.rabbit--Q2quick.brown.fox--无quick.orange.male.rabbit--无orange--无使用topic类型的Exchange，发送消息的routing key有3种： item.isnert、item.update、item.delete：生产者package com.test.demo.util.topic; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.MessageProperties; import com.test.demo.util.ConnectionUtil; import org.springframework.amqp.rabbit.core.RabbitTemplate; /** * Topic主题模型--生产者 * * @author jg * @since 2023/11/8 14:03 */ public class Send { private final static String EXCHANGE_NAME = "topic_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明exchange，指定类型为topic channel.exchangeDeclare(EXCHANGE_NAME, "topic"); // 消息内容 String message = "新增商品 : id = 1001"; // 发送消息，并且指定routing key 为：insert ,代表新增商品 channel.basicPublish(EXCHANGE_NAME, "item.insert", null, message.getBytes()); System.out.println(" [商品服务：] Sent '" + message + "'"); channel.close(); connection.close(); } }消费者1package com.test.demo.util.topic; import com.rabbitmq.client.*; import com.test.demo.util.ConnectionUtil; import java.io.IOException; /** * Topic主题模型--消费者 * * @author 蒋果 * @since 2023/11/8 14:04 */ public class Recv { private final static String QUEUE_NAME = "topic_exchange_queue_1"; private final static String EXCHANGE_NAME = "topic_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机，同时指定需要订阅的routing key。需要 update、delete channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete"); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [消费者1] received : " + msg + "!"); } }; // 监听队列，自动ACK channel.basicConsume(QUEUE_NAME, true, consumer); } }消费者2package com.test.demo.util.topic; import com.rabbitmq.client.*; import com.test.demo.util.ConnectionUtil; import java.io.IOException; import java.util.concurrent.TimeoutException; /** * Topic主题模型--消费者2 * * @author 蒋果 * @since 2023/11/8 14:04 */ /** * 消费者2 */ public class Recv2 { private final static String QUEUE_NAME = "topic_exchange_queue_2"; private final static String EXCHANGE_NAME = "topic_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, true, false, false, null); // 绑定队列到交换机，同时指定需要订阅的routing key。订阅 insert、update、delete channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.*.rabbit"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "lazy.#"); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [消费者2] received : " + msg + "!"); channel.basicAck(envelope.getDeliveryTag(), false); try { Thread.sleep(2000); } catch (InterruptedException e) { throw new RuntimeException(e); } } }; // 监听队列，自动ACK channel.basicConsume(QUEUE_NAME, false, consumer); } }还是一样,先启动Send生产者创建交换机,然后再启动消费者,最后再重启生产者发送信息测试结果只有消费者2接收到消息,说明主题模式搭建成功!持久化如何避免消息丢失？1） 消费者的手动ACK机制。可以防止业务处理失败。2） 但是，如果在消费者消费之前，MQ就宕机了，消息就没了。是可以将消息进行持久化呢？要将消息持久化，前提是：队列、Exchange都持久化我们再前面的例子中创建了很多的交换机和队列:交换机:队列:现在我们重启下rabbitMQ服务:重启后,我们之前创建的交换机和队列都消失了正是因为这样,所以我们才需要持久化.1.交换机持久化修改完成后启动,然后查看rabbitmq的交换机控制台出现这个标记,就表示这个交换机是持久化状态亲测,重启后,这个交换机还在2.队列持久化运行后查看队列控制台:队列持久化成功!3.消息持久化非持久化:在做非持久化测试时,消费者不能设置自动ACK,要手动ACK,因为设置了线程睡眠rabbitmq的队列控制台不会显示消息数量.生产者package com.test.demo.util.topic; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.MessageProperties; import com.test.demo.util.ConnectionUtil; import org.springframework.amqp.rabbit.core.RabbitTemplate; /** * Topic主题模型--生产者 * * @author jg * @since 2023/11/8 14:03 */ public class Send { private final static String EXCHANGE_NAME = "topic_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明exchange，指定类型为topic channel.exchangeDeclare(EXCHANGE_NAME, "topic",true); // 消息内容 String message = "消息 : 发送"; // 发送消息，并且指定routing key 为：insert ,代表新增商品 for (int i = 0; i < 50; i++) { channel.basicPublish(EXCHANGE_NAME, "quick.orange.rabbit", null, ("quick.orange.rabbit"+i).getBytes()); } System.out.println(" [商品服务：] Sent '" + message + "'"); channel.close(); connection.close(); } }消费者package com.test.demo.util.topic; import com.rabbitmq.client.*; import com.test.demo.util.ConnectionUtil; import java.io.IOException; /** * Topic主题模型--消费者 * * @author jg * @since 2023/11/8 14:04 */ public class Recv { private final static String QUEUE_NAME = "topic_exchange_queue_1"; private final static String EXCHANGE_NAME = "topic_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, true, false, false, null); // 绑定队列到交换机，同时指定需要订阅的routing key。需要 update、delete channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.orange.*"); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { try { // body 即消息体 String msg = new String(body); System.out.println(" [消费者1] received : " + msg + "!"); Thread.sleep(2000); } catch (InterruptedException e) { throw new RuntimeException(e); } } }; // 监听队列，自动ACK channel.basicConsume(QUEUE_NAME, true, consumer); } }可以看到消费者设置了,每接收一条信息就暂停2秒钟Thread.sleep(2000);发送50条信息后,在消费者还没接受完的时候,我们重启rabbitmq服务器,看看消息还在不在消费者接收控制台:重启服务器:可以看到,正在处理消息时,rabbitmq服务器重启,没有设置持久化的消息会全部丢失持久化:在发送消息时在basicPublish方法中配置这个参数:channel.basicPublish(EXCHANGE_NAME, "quick.orange.rabbit", MessageProperties.PERSISTENT_TEXT_PLAIN, ("quick.orange.rabbit"+i).getBytes());然后发送消息,在Recv消费者处理消息的时候,重启rabbitmq服务器重启时:重启后:可以看到消息恢复了,并且消费者还在继续处理消息消息持久化成功!使用RabbitTemplateRabbitAdmin类RabbitAdmin类是Spring AMQP提供的一个辅助类，用于简化与RabbitMQ交互的操作。它提供了一些便捷的方法，可以用于声明和管理队列、交换机和绑定等。RabbitAdmin类的主要作用有以下几点：声明和删除队列：RabbitAdmin类提供了declareQueue()和deleteQueue()等方法，可以用于声明和删除队列。你可以通过这些方法来创建和销毁队列，设置队列的属性和参数。声明和删除交换机：RabbitAdmin类提供了declareExchange()和deleteExchange()等方法，用于声明和删除交换机。你可以通过这些方法来创建不同类型的交换机，设置交换机的属性和参数。声明和绑定队列和交换机：RabbitAdmin类提供了declareBinding()和removeBinding()等方法，可以用于声明和删除队列和交换机之间的绑定。你可以使用这些方法来将队列绑定到交换机上，设置绑定的选项。管理RabbitMQ组件：RabbitAdmin类还提供了其他一些方法，如declareExchange()、declareBinding()等，用于管理RabbitMQ的其他组件，如连接、通道、消费者等。通过使用RabbitAdmin类，我们可以通过简单的方法调用来完成与RabbitMQ的操作，而不需要手动编写复杂的AMQP协议代码。它提供了一种方便和易于使用的方式来管理RabbitMQ的各种组件，简化了与RabbitMQ的交互过程。RabbitTemplate类RabbitTemplate类是Spring AMQP提供的一个核心类，用于在应用程序中与RabbitMQ进行交互。它封装了与RabbitMQ的连接、通道和基本操作，提供了一组方法来发送和接收消息，以及处理消息发送和接收的相关逻辑。RabbitTemplate类的主要作用有以下几点：消息发送：RabbitTemplate类提供了多个发送消息的方法，例如convertAndSend()、send()等。你可以使用这些方法将消息发送到指定的交换机和队列，并指定路由键、消息体和其他参数。同步和异步发送：RabbitTemplate类支持同步和异步的消息发送方式。你可以选择使用同步方式发送消息，等待发送结果返回；也可以选择使用异步方式发送消息，通过回调函数处理发送结果。消息接收：RabbitTemplate类提供了多个接收消息的方法，例如receiveAndConvert()、receive()等。你可以使用这些方法从指定的队列中接收消息，并指定超时时间、消息转换器和其他参数。消息转换：RabbitTemplate类支持消息的自动转换，可以将Java对象自动序列化为消息体，并在接收时将消息体自动反序列化为Java对象。你可以通过设置消息转换器（MessageConverter）来定制消息的序列化和反序列化过程。消息回复：RabbitTemplate类支持发送消息后等待并接收消息的模式，用于实现请求-响应模式。你可以在发送请求时指定回调队列，并在接收到响应消息时进行处理。通过使用RabbitTemplate类，我们可以方便地在应用程序中与RabbitMQ进行消息的发送和接收，而不需要手动处理底层的连接、通道和协议细节。它提供了一种高层次的抽象，简化了与RabbitMQ的交互过程，并提供了一些便捷的方法和功能来处理消息发送和接收的相关逻辑。使用示例1.配置部分server: port: 8080 spring: application: name: rabbitMQ-Demo #rabbitmq连接配置 rabbitmq: host: 你的IP地址 port: 5672 username: admin password: 123456 virtual-host: /2.代码部分package com.test.demo.util; import cn.hutool.json.JSONObject; import lombok.AllArgsConstructor; import org.springframework.amqp.core.*; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; /** * RabbitTemplate测试类 * * @author 蒋果 * @since 2023/11/8 15:38 */ @AllArgsConstructor @RestController @RequestMapping("/demo") public class rabbitMQController { private final RabbitTemplate rabbitTemplate; @GetMapping("/send") public String send(@RequestParam("message") String message) { //创建管理器 RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate.getConnectionFactory()); //声明交换机 rabbitAdmin.declareExchange(new DirectExchange("TopicExchange", true, false)); //声明队列 rabbitAdmin.declareQueue(new Queue("demo.Topic", true)); //声明绑定 rabbitAdmin.declareBinding( BindingBuilder .bind(QueueBuilder.durable("demo.Topic").build()) .to(new DirectExchange("TopicExchange", true, false)) .with("msg") ); //获取队列信息，名称，消息数量，消费者数量 rabbitAdmin.getQueueInfo("demo.Topic"); JSONObject msg = new JSONObject(); msg.put("msg", message); msg.put("msg2", "31212312312"); String s = msg.toString(); rabbitTemplate.convertAndSend("DirectExchange", "msg", s); return message; } }RabbitAdmin类主要用于创建交换机,队列还有绑定交换机RabbitTemplate类主要用来发送和接收队列中的信息,可以把信息转换成各种对象@RabbitListener注解开启注解使用这个注解需要在启动类加上@EnableRabbit注解://开启rabbit注解 @EnableRabbit代码示例:package com.test.demo.util; import cn.hutool.json.JSONObject; import lombok.AllArgsConstructor; import org.springframework.amqp.core.ExchangeTypes; import org.springframework.amqp.rabbit.annotation.*; import org.springframework.stereotype.Component; /** * rabbitmq监听器 * * @author 蒋果 * @since 2023/11/9 11:11 */ @Component @RabbitListener(bindings = @QueueBinding( value = @Queue(value = "demoDirect", durable = "true"), exchange = @Exchange(value = "DirectExchange", type = ExchangeTypes.DIRECT,declare = "true"), key = "msg" )) @AllArgsConstructor public class MyMessageListener { @RabbitHandler public void ttlListerer(String msg){ JSONObject msg1 = msg == null ? null : new JSONObject(msg); if (msg1 != null) { System.out.println("收到消息："+msg1.get("msg")); } } }一些报错的处理1.接收消息时反序列化出错Attempt to deserialize unauthorized class java.util.HashMap; add allowed class name patterns to the message converter or, if you trust the message orginiator, set environment variable 'SPRING_AMQP_DESERIALIZATION_TRUST_ALL' or system property 'spring.amqp.deserialization.trust.all' to true这个错误提示的是Spring AMQP正在尝试反序列化HashMap类，但是由于安全原因，这个类并未被允许。在Spring Boot中，可以通过以下方式来配置,在启动类中添加jsonMessageConverter和classMapper方法,并注册为bean：package com.test.demo; import org.springframework.amqp.rabbit.annotation.EnableRabbit; import org.springframework.amqp.support.converter.DefaultClassMapper; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; @SpringBootApplication @EnableRabbit public class DemoApplication { public static void main(String[] args) { SpringApplication.run(DemoApplication.class, args); } @Bean public MessageConverter jsonMessageConverter() { Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter(); converter.setClassMapper(classMapper()); return converter; } @Bean public DefaultClassMapper classMapper() { DefaultClassMapper classMapper = new DefaultClassMapper(); // 添加允许的类名模式 classMapper.setTrustedPackages("java.util.HashMap"); return classMapper; } }
1、RabbitMQ简介官网：https://www.rabbitmq.com/RabbitMQ是一个开源的遵循AMQP协议实现的基于Erlang语言编写，支持多种客户端（语言），用于在分布式系统中存储消息，转发消息，具有高可用高可扩性，易用性等特征。2.下载,安装,启动下载地址：https://www.rabbitmq.com/download.html选择对应的系统版本点击下载，下载后会得到.rpm文件下载ErlangRabbitMQ是采用 Erlang语言开发的，所以系统环境必须提供 Erlang环境，需要是安装 ErlangErlang和RabbitMQ版本对照：https://www.rabbitmq.com/which-erlang.html这里安装最新版本3.12.8的RabbitMQ，对应的Erlang版本推荐26.1.x，我们下载erlang-26.1.2-1.el9.x86_64.rpm下载地址：https://packagecloud.io/rabbitmq/erlang/packages/el/9/erlang-26.1.2-1.el9.x86_64.rpm?distro_version_id=240其中的el9表示Red Hat 9.x，即CentOS 9.x，因为我没找到7.x的版本，所以暂时用这个点击右上角下载即可得到.rpm文件安装Erlang首先将下载好的文件上传到服务器，创建一个文件夹用来存放文件mkdir -p /usr/rabbitmq再利用xftp工具将上述下载的两个.rpm文件上传到服务器的刚创建的文件夹中然后切换到/usr/rabbitmq目录，解压安装erlang# 解压 rpm -Uvh erlang-26.1.2-1.el9.x86_64.rpm # 安装 yum install -y erlang安装完成后输入如下指令查看版本号# 查看版本号 erl -v安装RabbitMQ在RabiitMQ安装过程中需要依赖socat插件，首先安装该插件# 安装socat插件 yum install -y socat然后解压安装RabbitMQ的安装包# 解压 rpm -Uvh rabbitmq-server-3.12.8-1.el8.noarch.rpm # 安装 yum install -y rabbitmq-server启动RabbitMQ服务# 启动rabbitmq systemctl start rabbitmq-server # 查看rabbitmq状态 systemctl status rabbitmq-server显示如下信息则启动成功：其他命令：# 设置rabbitmq服务开机自启动 systemctl enable rabbitmq-server # 关闭rabbitmq服务 systemctl stop rabbitmq-server # 重启rabbitmq服务 systemctl restart rabbitmq-server3、RabbitMQ Web管理界面及授权操作安装启动RabbitMQ Web管理界面默认情况下，rabbitMQ没有安装web端的客户端软件，需要安装才可以生效# 打开RabbitMQWeb管理界面插件 rabbitmq-plugins enable rabbitmq_management然后我们打开浏览器，访问服务器公网ip:15672（注意打开阿里云安全组以及防火墙的15672端口），就可以看到管理界面浏览器访问：rabbitmq有一个默认的账号密码guest，但该情况仅限于本机localhost进行访问，所以需要添加一个远程登录的用户添加远程用户# 添加用户 rabbitmqctl add_user 用户名 密码 # 设置用户角色,分配操作权限 rabbitmqctl set_user_tags 用户名 角色 # 为用户添加资源权限(授予访问虚拟机根节点的所有权限) rabbitmqctl set_permissions -p / 用户名 ".*" ".*" ".*"角色有四种：administrator：可以登录控制台、查看所有信息、并对rabbitMQ进行管理monToring：监控者；登录控制台，查看所有信息policymaker：策略制定者；登录控制台指定策略managment：普通管理员；登录控制这里创建用户admin，密码123456，设置administrator角色，赋予所有权限创建完成后，访问服务器ip:15672进行登录，然后便可进入到后台其他指令：# 修改密码 rabbitmqctl change_ password 用户名 新密码 # 删除用户 rabbitmqctl delete_user 用户名 # 查看用户清单 rabbitmqctl list_users
最近公司其他项目组出现了一次生产事故，导致那个项目的近几个月数据全部丢失，好在有纸质资料作为数据存根，为此我们几个组加班加点手动录入数据苦不堪言。那之后各项目都要求定时备份数据库，所以我整理了我在使用MySQL备份远程数据库(centOS)的相关步骤。系统环境：1.centOS72.MySQL 8.0.31创建脚本文件创建数据库备份存放的文件夹和脚本文件，如图：写入脚本打开刚创建好的backup_mysql.sh文件，将以下内容写入进去：#!/bin/bash # 定义备份文件保存路径 backup_property_dir="/home/database_Backup/property" backup_property_ac_dir="/home/database_Backup/property_ac" backup_property_config_dir="/home/database_Backup/property_config" backup_property_job_dir="/home/database_Backup/property_job" backup_property_manage_dir="/home/database_Backup/property_manage" # 转换日期 timestamp=$(date +%Y%m%d%H%M%S) year=${timestamp:0:4} month=${timestamp:4:2} day=${timestamp:6:2} hour=${timestamp:8:2} minute=${timestamp:10:2} second=${timestamp:12:2} formatted_date="${year}年${month}月${day}日_${hour}:${minute}:${second}" # 定义备份文件名 backup_property="$backup_property_dir/backup_property_$formatted_date.sql" backup_property_ac="$backup_property_ac_dir/backup_property_ac_$formatted_date.sql" backup_property_config="$backup_property_config_dir/backup_property_config_$formatted_date.sql" backup_property_job="$backup_property_job_dir/backup_property_job_$formatted_date.sql" backup_property_manage="$backup_property_manage_dir/backup_property_manage_$formatted_date.sql" # 定义 MySQL 连接参数 mysql_user="你的MySQL用户名" mysql_password="你的MySQL密码" # 定义要备份的数据库名称 property_database="property" property_ac_database="property_ac" property_config_database="property_config" property_job_database="property_job" property_manage_database="property_manage" # 删除七天前的备份文件 find /home/database_Backup/property* -type f -name "backup_property_*" -mtime +7 -exec rm {} \; find /home/database_Backup/property_ac* -type f -name "backup_property_ac_*" -mtime +7 -exec rm {} \; find /home/database_Backup/property_config* -type f -name "backup_property_config_*" -mtime +7 -exec rm {} \; find /home/database_Backup/property_job* -type f -name "backup_property_job_*" -mtime +7 -exec rm {} \; find /home/database_Backup/property_manage* -type f -name "backup_property_manage_*" -mtime +7 -exec rm {} \; # 执行备份操作 mysqldump -h 你的ip地址 -u $mysql_user -p$mysql_password $property_database > $backup_property mysqldump -h 你的ip地址 -u $mysql_user -p$mysql_password $property_ac_database > $backup_property_ac mysqldump -h 你的ip地址 -u $mysql_user -p$mysql_password $property_config_database > $backup_property_config mysqldump -h 你的ip地址 -u $mysql_user -p$mysql_password $property_job_database > $backup_property_job mysqldump -h 你的ip地址 -u $mysql_user -p$mysql_password $property_manage_database > $backup_property_manage 设置脚本权限设置定时任务手动打开文件设置我这里使用的是crontab命令，如果提示没找到命令就自行百度安装使用 service crond status 命令查看crontab是否运行：service crond status打开并编辑/etc/crontab文件：#每天凌晨1点执行备份数据库操作 00 1 * * * /home/database_Backup/backup_mysql.sh使用指令设置定时任务#打开vim编辑器： crontab -e #设置每天凌晨1点执行备份数据库 #打开vim编辑器后按s键，将以下命令输入到文件中然后按Esc键，再输入:wq保存。 00 1 * * * /home/database_Backup/backup_mysql.sh #查看命令是否写入成功，出现上述任务则表示成功 crontab -l以上两种方式都可以实现linux的定时任务，但是据我测试只有第二种使用vim编辑器输入命令才可以执行定时任务，暂时没找到第一种方案无效的原因。
异步执行对于开发者来说并不陌生，在实际的开发过程中，很多场景多会使用到异步，相比同步执行，异步可以大大缩短请求链路耗时时间，比如：发送短信、邮件、异步更新等，这些都是典型的可以通过异步实现的场景。异步的八种实现方式线程ThreadFuture异步框架CompletableFutureSpring注解@AsyncSpring ApplicationEvent事件消息队列第三方异步框架，比如Hutool的ThreadUtilGuava异步什么是异步？首先我们先看一个常见的用户下单的场景：在同步操作中，我们执行到 发送短信 的时候，我们必须等待这个方法彻底执行完才能执行 赠送积分 这个操作，如果 赠送积分 这个动作执行时间较长，发送短信需要等待，这就是典型的同步场景。实际上，发送短信和赠送积分没有任何的依赖关系，通过异步，我们可以实现赠送积分和发送短信这两个操作能够同时进行，比如：这就是所谓的异步，是不是非常简单，下面就说说异步的几种实现方式吧。1. 线程异步public class AsyncThread extends Thread { @Override public void run() { System.out.println("Current thread name:" + Thread.currentThread().getName() + " Send email success!"); } public static void main(String[] args) { AsyncThread asyncThread = new AsyncThread(); asyncThread.run(); } }当然如果每次都创建一个Thread线程，频繁的创建、销毁，浪费系统资源，我们可以采用线程池：private ExecutorService executorService = Executors.newCachedThreadPool(); public void fun() { executorService.submit(new Runnable() { @Override public void run() { log.info("执行业务逻辑..."); } }); }可以将业务逻辑封装到Runnable或Callable中，交由线程池来执行。2. Future异步@Slf4j public class FutureManager { public String execute() throws Exception { ExecutorService executor = Executors.newFixedThreadPool(1); Future<String> future = executor.submit(new Callable<String>() { @Override public String call() throws Exception { System.out.println(" --- task start --- "); Thread.sleep(3000); System.out.println(" --- task finish ---"); return "this is future execute final result!!!"; } }); //这里需要返回值时会阻塞主线程 String result = future.get(); log.info("Future get result: {}", result); return result; } @SneakyThrows public static void main(String[] args) { FutureManager manager = new FutureManager(); manager.execute(); } }输出结果： --- task start --- --- task finish --- Future get result: this is future execute final result!!!Future的不足之处Future的不足之处的包括以下几点：无法被动接收异步任务的计算结果：虽然我们可以主动将异步任务提交给线程池中的线程来执行，但是待异步任务执行结束之后，主线程无法得到任务完成与否的通知，它需要通过get方法主动获取任务执行的结果。Future件彼此孤立：有时某一个耗时很长的异步任务执行结束之后，你想利用它返回的结果再做进一步的运算，该运算也会是一个异步任务，两者之间的关系需要程序开发人员手动进行绑定赋予，Future并不能将其形成一个任务流（pipeline），每一个Future都是彼此之间都是孤立的，所以才有了后面的CompletableFuture，CompletableFuture就可以将多个Future串联起来形成任务流。Futrue没有很好的错误处理机制：截止目前，如果某个异步任务在执行发的过程中发生了异常，调用者无法被动感知，必须通过捕获get方法的异常才知晓异步任务执行是否出现了错误，从而在做进一步的判断处理。3. CompletableFuture实现异步public class CompletableFutureCompose { /** * thenAccept子任务和父任务公用同一个线程 */ @SneakyThrows public static void thenRunAsync() { CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> { System.out.println(Thread.currentThread() + " cf1 do something...."); return 1; }); CompletableFuture<Void> cf2 = cf1.thenRunAsync(() -> { System.out.println(Thread.currentThread() + " cf2 do something..."); }); //等待任务1执行完成 System.out.println("cf1结果->" + cf1.get()); //等待任务2执行完成 System.out.println("cf2结果->" + cf2.get()); } public static void main(String[] args) { thenRunAsync(); } }我们不需要显式使用ExecutorService，CompletableFuture 内部使用了ForkJoinPool来处理异步任务，如果在某些业务场景我们想自定义自己的异步线程池也是可以的。4. Spring的@Async异步自定义异步线程池：/** * 线程池参数配置，多个线程池实现线程池隔离，@Async注解，默认使用系统自定义线程池，可在项目中设置多个线程池，在异步调用的时候，指明需要调用的线程池名称，比如：@Async("taskName") **/ @EnableAsync @Configuration public class TaskPoolConfig { /** * 自定义线程池 * **/ @Bean("taskExecutor") public Executor taskExecutor() { //返回可用处理器的Java虚拟机的数量 12 int i = Runtime.getRuntime().availableProcessors(); System.out.println("系统最大线程数 ： " + i); ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); //核心线程池大小 executor.setCorePoolSize(16); //最大线程数 executor.setMaxPoolSize(20); //配置队列容量，默认值为Integer.MAX_VALUE executor.setQueueCapacity(99999); //活跃时间 executor.setKeepAliveSeconds(60); //线程名字前缀 executor.setThreadNamePrefix("asyncServiceExecutor -"); //设置此执行程序应该在关闭时阻止的最大秒数，以便在容器的其余部分继续关闭之前等待剩余的任务完成他们的执行 executor.setAwaitTerminationSeconds(60); //等待所有的任务结束后再关闭线程池 executor.setWaitForTasksToCompleteOnShutdown(true); return executor; } }AsyncService：public interface AsyncService { MessageResult sendSms(String callPrefix, String mobile, String actionType, String content); MessageResult sendEmail(String email, String subject, String content); } @Slf4j @Service public class AsyncServiceImpl implements AsyncService { @Autowired private IMessageHandler mesageHandler; @Override @Async("taskExecutor") public MessageResult sendSms(String callPrefix, String mobile, String actionType, String content) { try { Thread.sleep(1000); mesageHandler.sendSms(callPrefix, mobile, actionType, content); } catch (Exception e) { log.error("发送短信异常 -> ", e) } } @Override @Async("taskExecutor") public sendEmail(String email, String subject, String content) { try { Thread.sleep(1000); mesageHandler.sendsendEmail(email, subject, content); } catch (Exception e) { log.error("发送email异常 -> ", e) } } }在实际项目中， 使用@Async调用线程池，推荐等方式是是使用自定义线程池的模式，不推荐直接使用@Async直接实现异步。5. Spring ApplicationEvent事件实现异步定义事件：public class AsyncSendEmailEvent extends ApplicationEvent { /** * 邮箱 **/ private String email; /** * 主题 **/ private String subject; /** * 内容 **/ private String content; /** * 接收者 **/ private String targetUserId; }定义事件处理器：@Slf4j @Component public class AsyncSendEmailEventHandler implements ApplicationListener<AsyncSendEmailEvent> { @Autowired private IMessageHandler mesageHandler; @Async("taskExecutor") @Override public void onApplicationEvent(AsyncSendEmailEvent event) { if (event == null) { return; } String email = event.getEmail(); String subject = event.getSubject(); String content = event.getContent(); String targetUserId = event.getTargetUserId(); mesageHandler.sendsendEmailSms(email, subject, content, targerUserId); } }另外，可能有些时候采用ApplicationEvent实现异步的使用，当程序出现异常错误的时候，需要考虑补偿机制，那么这时候可以结合Spring Retry重试来帮助我们避免这种异常造成数据不一致问题。6. 消息队列回调事件消息生产者：@Slf4j @Component public class CallbackProducer { @Autowired AmqpTemplate amqpTemplate; public void sendCallbackMessage(CallbackDTO allbackDTO, final long delayTimes) { log.info("生产者发送消息，callbackDTO，{}", callbackDTO); amqpTemplate.convertAndSend(CallbackQueueEnum.QUEUE_GENSEE_CALLBACK.getExchange(), CallbackQueueEnum.QUEUE_GENSEE_CALLBACK.getRoutingKey(), JsonMapper.getInstance().toJson(genseeCallbackDTO), new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { //给消息设置延迟毫秒值，通过给消息设置x-delay头来设置消息从交换机发送到队列的延迟时间 message.getMessageProperties().setHeader("x-delay", delayTimes); message.getMessageProperties().setCorrelationId(callbackDTO.getSdkId()); return message; } }); } }回调事件消息消费者：@Slf4j @Component @RabbitListener(queues = "message.callback", containerFactory = "rabbitListenerContainerFactory") public class CallbackConsumer { @Autowired private IGlobalUserService globalUserService; @RabbitHandler public void handle(String json, Channel channel, @Headers Map<String, Object> map) throws Exception { if (map.get("error") != null) { //否认消息 channel.basicNack((Long) map.get(AmqpHeaders.DELIVERY_TAG), false, true); return; } try { CallbackDTO callbackDTO = JsonMapper.getInstance().fromJson(json, CallbackDTO.class); //执行业务逻辑 globalUserService.execute(callbackDTO); //消息消息成功手动确认，对应消息确认模式acknowledge-mode: manual channel.basicAck((Long) map.get(AmqpHeaders.DELIVERY_TAG), false); } catch (Exception e) { log.error("回调失败 -> {}", e); } } }7. ThreadUtil异步工具类@Slf4j public class ThreadUtils { public static void main(String[] args) { for (int i = 0; i < 3; i++) { ThreadUtil.execAsync(() -> { ThreadLocalRandom threadLocalRandom = ThreadLocalRandom.current(); int number = threadLocalRandom.nextInt(20) + 1; System.out.println(number); }); log.info("当前第：" + i + "个线程"); } log.info("task finish!"); } }8. Guava异步Guava的ListenableFuture顾名思义就是可以监听的Future，是对java原生Future的扩展增强。我们知道Future表示一个异步计算任务，当任务完成时可以得到计算结果。如果我们希望一旦计算完成就拿到结果展示给用户或者做另外的计算，就必须使用另一个线程不断的查询计算状态。这样做，代码复杂，而且效率低下。使用Guava ListenableFuture可以帮我们检测Future是否完成了，不需要再通过get()方法苦苦等待异步的计算结果，如果完成就自动调用回调函数，这样可以减少并发程序的复杂度。ListenableFuture是一个接口，它从jdk的Future接口继承，添加了void addListener(Runnable listener, Executor executor)方法。我们看下如何使用ListenableFuture。首先需要定义ListenableFuture的实例: ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool()); final ListenableFuture<Integer> listenableFuture = executorService.submit(new Callable<Integer>() { @Override public Integer call() throws Exception { log.info("callable execute...") TimeUnit.SECONDS.sleep(1); return 1; } });首先通过MoreExecutors类的静态方法listeningDecorator方法初始化一个ListeningExecutorService的方法，然后使用此实例的submit方法即可初始化ListenableFuture对象。ListenableFuture要做的工作，在Callable接口的实现类中定义，这里只是休眠了1秒钟然后返回一个数字1，有了ListenableFuture实例，可以执行此Future并执行Future完成之后的回调函数。 Futures.addCallback(listenableFuture, new FutureCallback<Integer>() { @Override public void onSuccess(Integer result) { //成功执行... System.out.println("Get listenable future's result with callback " + result); } @Override public void onFailure(Throwable t) { //异常情况处理... t.printStackTrace(); } });
本篇内容主要记录.NET中使用MinIO文件存储的常用方法操作类，有关MioIO的概念和优缺点就不写了，百度都有。MinIO文件服务器的搭建请查看：Java集成MinIO，项目中使用NuGet搜索MinIO进行安装即可。using System; using System.Collections.Generic; using System.IO; using System.Threading; using System.Threading.Tasks; using Minio; using Minio.DataModel; using Minio.Exceptions; namespace ConsoleApplication1 { public static class MinioHelper { static readonly string _endPoint = "127.0.0.1:9001"; static readonly string _accessKey = "minioadmin"; static readonly string _secretKey = "minioadmin"; static readonly bool _isSSL = false; private static readonly MinioClient _minioClient = new MinioClient() .WithEndpoint(_endPoint) .WithCredentials(_accessKey,_secretKey) .WithSSL(_isSSL) .Build(); /// /// 创建minio客户端连接 /// /// 127.0.0.1:9001 /// minioadmin /// minioadmin /// 是否是http /// <returns></returns> public static MinioClient Create(string endPoint, string accessKey, string secretKey, bool withSsl = false) { MinioClient client = new MinioClient() .WithEndpoint(endPoint) .WithCredentials(accessKey, secretKey); if (withSsl) { client = client.WithSSL(); } return client.Build(); } #region 操作存储桶 /// /// 创建存储桶 /// <example > /// /// MinioHelper.MakeBucket(minio, buckName).Wait(); /// /// </example> /// /// 存储桶名称 /// 可选参数 /// <returns ></returns > public static Task CreateBucket(string bucketName, string loc = "us-east-1") { try { bool found = BucketExists(bucketName); if (found) { throw new Exception($"存储桶[{bucketName}]已存在"); } MakeBucketArgs args = new MakeBucketArgs() .WithBucket(bucketName) .WithLocation(loc); return _minioClient.MakeBucketAsync(args); } catch (Exception e) { throw new Exception(e.Message); } } /// /// 校验桶是否存在,如果不存在则报错 /// <example> /// 调用示例 /// /// bool exists = MinioHelper.BucketExists(minio, buckName); /// /// </example> /// /// /// <exception cref="Exception"></exception> public static void CheckBucket(string bucketName) { bool exists = BucketExists(bucketName); if (!exists) { throw new BucketNotFoundException(bucketName, "存储桶不存在"); } } /// /// 列出所有的存储桶 /// /// <example> /// /// abc /// </example> /// <returns></returns> public static ListAllMyBucketsResult ListBuckets() { Task<ListAllMyBucketsResult> data = _minioClient.ListBucketsAsync(); data.Wait(); return data.Result; } /// /// 检查存储桶是否存在 /// /// <example> /// /// var data = MinioHelper.ListBuckets(minio); /// /// </example> /// 存储桶名称 /// 取消标记 /// <returns></returns> public static bool BucketExists(string bucketName, CancellationToken cancellationToken = default(CancellationToken)) { try { BucketExistsArgs args = new BucketExistsArgs() .WithBucket(bucketName); Task<bool> bucketExistTask = _minioClient.BucketExistsAsync(args, cancellationToken); Task.WaitAll(bucketExistTask); return bucketExistTask.Result; } catch (Exception e) { throw new Exception(e.Message); } } /// /// 删除一个存储桶 /// /// 存储桶名称 /// 默认不填 /// <returns></returns> public static Task RemoveBucket(string bucketName, CancellationToken cancellationToken = default(CancellationToken)) { try { CheckBucket(bucketName); RemoveBucketArgs args = new RemoveBucketArgs() .WithBucket(bucketName); return _minioClient.RemoveBucketAsync(args, cancellationToken); } catch (Exception e) { throw new Exception(e.Message); } } /// 列出存储桶里的对象 /// 列出存储桶里的对象 /// /// 存储桶名称 /// 对象的前缀 /// true代表递归查找，false代表类似文件夹查找，以'/'分隔，不查子文件夹 public static IObservable<Item> ListObjects(string bucketName, string prefix = null, bool recursive = true) { try { ListObjectsArgs args = new ListObjectsArgs() .WithBucket(bucketName) .WithPrefix(prefix) .WithRecursive(recursive); IObservable<Item> data = _minioClient.ListObjectsAsync(args); return data; } catch (Exception e) { throw new Exception(e.Message); } } #endregion #region 操作文件对象 public static bool FileExist(string bucketName, string objectName) { try { var obj = new StatObjectArgs().WithBucket(bucketName) .WithObject(objectName); var objStat = _minioClient.StatObjectAsync(obj); objStat.Wait(); } catch (AggregateException e) { foreach (var item in e.InnerExceptions) { if (item is ObjectNotFoundException) { return false; } } } return true; } /// /// 从桶下载文件到流 /// /// 存储桶名称 /// 存储桶里的对象名称 /// 默认不填 /// 默认不填 /// <returns></returns> public static Task<ObjectStat> FGetObject(string bucketName, string objectName, ServerSideEncryption sse = null, Action<Stream> cb = null) { CheckBucket(bucketName); try { GetObjectArgs args = new GetObjectArgs() .WithBucket(bucketName) .WithObject(objectName) .WithServerSideEncryption(sse) .WithCallbackStream(cb); return _minioClient.GetObjectAsync(args); } catch (MinioException e) { throw new Exception(e.Message); } } /// /// 从桶下载文件到本地 /// /// /// /// /// /// <returns></returns> public static Task<ObjectStat> FGetObject(string bucketName, string objectName, string fileName,ServerSideEncryption sse = null) { CheckBucket(bucketName); if (File.Exists(fileName)) { File.Delete(fileName); } return FGetObject(bucketName, objectName, sse, stream => { using (FileStream fileStream = new FileStream(fileName, FileMode.CreateNew, FileAccess.Write)) { stream.CopyTo(fileStream); } }); } /// /// 上传本地文件至存储桶 /// /// 存储桶名称 /// 存储桶里的对象名称 /// 本地路径 /// contentType：默认不填 /// metaData：默认不填 /// 默认不填 /// <returns></returns> public static Task FPutObject(string bucketName, string objectName, string fileName, string contentType = "application/octet-stream", Dictionary<string, string> metaData = null, ServerSideEncryption sse = null) { CheckBucket(bucketName); try { //var data= minio.PutObjectAsync(bucketName, objectName, fileName, contentType: "application/octet-stream"); PutObjectArgs args = new PutObjectArgs() .WithBucket(bucketName) .WithObject(objectName) .WithFileName(fileName) .WithContentType(contentType) .WithHeaders(metaData) .WithServerSideEncryption(sse); return _minioClient.PutObjectAsync(args); } catch (MinioException e) { throw new Exception(e.Message); } } #endregion #region Presigned操作 /// 生成一个给HTTP GET请求用的presigned URL。浏览器/移动端的客户端可以用这个URL进行下载，即使其所在的存储桶是私有的。这个presigned URL可以设置一个失效时间，默认值是7天。 /// 生成一个给HTTP GET请求用的presigned URL。浏览器/移动端的客户端可以用这个URL进行下载，即使其所在的存储桶是私有的。这个presigned URL可以设置一个失效时间，默认值是7天。 /// /// 存储桶名称 /// 存储桶里的对象名称 /// 失效时间（以秒为单位），默认是7天，不得大于七天 /// <returns></returns> public static Task<string> PresignedGetObject(string bucketName, string objectName, int expiresInt = 1000) { CheckBucket(bucketName); try { //Dictionary<string, string> reqParams = new Dictionary<string, string> { { "response-content-type", "application/json" } }; PresignedGetObjectArgs args = new PresignedGetObjectArgs() .WithBucket(bucketName) .WithObject(objectName) //.WithHeaders(reqParams) .WithExpiry(expiresInt); //.WithRequestDate(DateTime.Now.ToUniversalTime()); return _minioClient.PresignedGetObjectAsync(args); } catch (Exception e) { throw new Exception(e.Message); } } /// /// 生成一个给HTTP PUT请求用的presigned URL。浏览器/移动端的客户端可以用这个URL进行上传，即使其所在的存储桶是私有的。这个presigned URL可以设置一个失效时间，默认值是7天。 /// /// 存储桶名称 /// 存储桶里的对象名称 /// 失效时间（以秒为单位），默认是7天，不得大于七天 /// <returns></returns> public static Task<string> PresignedPutObject(string bucketName, string objectName, int expiresInt = 1000) { CheckBucket(bucketName); try { //string presignedUrl = await minio.PresignedPutObjectAsync(bucketName, objectName, expiresInt); //Ret = presignedUrl; //flag = true; PresignedPutObjectArgs args = new PresignedPutObjectArgs() .WithBucket(bucketName) .WithObject(objectName) .WithExpiry(expiresInt); return _minioClient.PresignedPutObjectAsync(args); } catch (Exception e) { throw new Exception(e.Message); } } /// /// 允许给POST请求的presigned URL设置策略，比如接收对象上传的存储桶名称的策略，key名称前缀，过期策略。 /// /// 对象的post策略 /// <returns></returns> public static async Task<(Uri, Dictionary<string, string>)> PresignedPostPolicy(PostPolicy form) { try { //例如 //PostPolicy form = new PostPolicy(); //DateTime expiration = DateTime.UtcNow; //form.SetExpires(expiration.AddDays(10)); //form.SetKey("my-objectname"); //form.SetBucket("my-bucketname"); (Uri, Dictionary<string, string>) data = await _minioClient.PresignedPostPolicyAsync(form); return data; } catch (Exception e) { throw new Exception(e.Message); } } #endregion #region 操作对象 /// /// 下载对象指定区域的字节数组做为流。offset和length都必须传 /// /// 存储桶名称 /// 存储桶里的对象名称 /// offset是起始字节的位置 /// length是要读取的长度 /// 处理流的回调函数 /// <returns></returns> public static async Task GetObjectAsync(string bucketName, string objectName, long offset, long length, Action<Stream> callback) { CheckBucket(bucketName); try { StatObjectArgs args = new StatObjectArgs() .WithBucket(bucketName) .WithObject(objectName) .WithServerSideEncryption(null); await _minioClient.StatObjectAsync(args); GetObjectArgs objArgs = new GetObjectArgs() .WithBucket(bucketName) .WithObject(objectName) .WithCallbackStream(callback) .WithOffsetAndLength(offset, length) .WithServerSideEncryption(null); await _minioClient.GetObjectAsync(objArgs); } catch (MinioException e) { throw new Exception(e.Message); } } /// /// 通过Stream上传对象 /// /// 存储桶名称 /// 存储桶里的对象名称 /// 要上传的Stream对象 /// 流的大小 /// 文件的Content type，默认是"application/octet-stream" /// 元数据头信息的Dictionary对象，默认是null /// <returns></returns> public static async Task PutObjectAsync(string bucketName, string objectName, Stream data, long size, string contentType = "application/octet-stream", Dictionary<string, string> metaData = null) { CheckBucket(bucketName); try { PutObjectArgs args = new PutObjectArgs() .WithBucket(bucketName) .WithObject(objectName) .WithStreamData(data) .WithObjectSize(size) .WithContentType(contentType) .WithHeaders(metaData) .WithServerSideEncryption(null); await _minioClient.PutObjectAsync(args); } catch (MinioException e) { throw new Exception(e.Message); } } /// /// 从objectName指定的对象中将数据拷贝到destObjectName指定的对象 /// /// 源存储桶名称 /// 源存储桶中的源对象名称 /// 目标存储桶名称 /// 要创建的目标对象名称,如果为空，默认为源对象名称 /// 拷贝操作的一些条件Map /// 默认不填 /// 默认不填 /// 默认不填 /// <returns></returns> public static async Task CopyObject(string fromBucketName, string fromObjectName, string destBucketName, string destObjectName, CopyConditions copyConditions = null, ServerSideEncryption sseSrc = null, ServerSideEncryption sseDest = null, Dictionary<string, string> metadata = null) { CheckBucket(fromBucketName); CheckBucket(destBucketName); try { CopySourceObjectArgs cpSrcArgs = new CopySourceObjectArgs() .WithBucket(fromBucketName) .WithObject(fromObjectName) .WithCopyConditions(copyConditions) .WithServerSideEncryption(sseSrc); CopyObjectArgs args = new CopyObjectArgs() .WithBucket(destBucketName) .WithObject(destObjectName) .WithCopyObjectSource(cpSrcArgs) .WithHeaders(metadata) .WithServerSideEncryption(sseDest); await _minioClient.CopyObjectAsync(args); } catch (MinioException e) { throw new Exception(e.Message); } } /// /// 删除一个对象 /// /// 存储桶名称 /// 存储桶里的对象名称 /// <returns></returns> public static Task RemoveObject(string bucketName, string objectName) { CheckBucket(bucketName); try { RemoveObjectArgs args = new RemoveObjectArgs() .WithBucket(bucketName) .WithObject(objectName); return _minioClient.RemoveObjectAsync(args); } catch (MinioException e) { throw new Exception(e.Message); } } /// /// 删除多个对象 /// /// 存储桶名称 /// 含有多个对象名称的IEnumerable /// <returns></returns> public static async Task<bool> RemoveObjects(string bucketName, List<string> objectsList) { CheckBucket(bucketName); bool flag = false; try { if (objectsList != null) { flag = true; RemoveObjectsArgs args = new RemoveObjectsArgs() .WithBucket(bucketName) .WithObjects(objectsList); await _minioClient.RemoveObjectsAsync(args); } } catch (MinioException e) { throw new Exception(e.Message); } return flag; } #endregion } }
在介绍Nginx的负载均衡实现之前，先简单的说下负载均衡的分类，主要分为硬件负载均衡和软件负载均衡，硬件负载均衡是使用专门的软件和硬件相结合的设备，设备商会提供完整成熟的解决方案，比如F5，在数据的稳定性以及安全性来说非常可靠，但是相比软件而言造价会更加昂贵；软件的负载均衡以Nginx这类软件为主，实现一种消息队列分发机制。简单来说所谓的负载均衡就是把很多请求进行分流，将他们分配到不同的服务器去处理。比如我有3个服务器，分别为A、B、C，然后使用Nginx进行负载均衡，使用轮询策略，此时如果收到了9个请求，那么会均匀的将这9个请求分发给A、B、Cf服务器，每一个服务器处理3个请求，这样的话我们可以利用多台机器集群的特性减少单个服务器的压力。Nginx实现负载均衡的示例图:负载均衡策略NGINX开源支持四种负载平衡方法，而NGINX Plus又增加了两种方法。1.Round Robin: 对所有的请求进行轮询发送请求，默认的分配方式。nginx.conf 配置示例:upstream jiangguo { server www.handsomejg.com; server www.handsomejg2.com; }注:上面的域名也可以用IP替代。2.Least Connections：以最少的活动连接数将请求发送到服务器，同样要考虑服务器权重。nginx.conf 配置示例:upstream jiangguo { least_conn; server www.handsomejg.com; server www.handsomejg2.com; }3.IP Hash : 发送请求的服务器由客户机IP地址决定。在这种情况下，使用IPv4地址的前三个字节或整个IPv6地址来计算散列值。该方法保证来自相同地址的请求到达相同的服务器，除非该服务器不可用。upstream jiangguo { ip_hash; server www.handsomejg.com; server www.handsomejg2.com; }4.Generic Hash: 请求发送到的服务器由用户定义的键决定，该键可以是文本字符串、变量或组合。upstream jiangguo { hash $request_uri consistent; server www.handsomejg.com; server www.handsomejg2.com; }5.Least Time (NGINX Plus only) – 对于每个请求，NGINX Plus选择具有最低平均延迟和最低活动连接数的服务器，其中最低平均延迟是根据包含least_time指令的下列参数计算的:header ： 从服务器接收第一个字节的时间。 last_byte： 从服务器接收完整响应的时间。 last_byte inflight：从服务器接收完整响应的时间。 upstream jiangguo { least_time header; server www.handsomejg.com; server www.handsomejg2.com; }6.Random：每个请求将被传递到随机选择的服务器。如果指定了两个参数，首先，NGINX根据服务器权重随机选择两个服务器，然后使用指定的方法选择其中一个。least_conn ：活动连接的最少数量 least_time=header (NGINX Plus)：从服务器接收响应标头的最短平均时间 ($upstream_header_time)。 least_time=last_byte (NGINX Plus) ：从服务器接收完整响应的最短平均时间（$upstream_response_time）。 upstream jiangguo { random two least_time=last_byte; server www.handsomejg.com; server www.handsomejg2.com; }环境准备依赖JDK1.8以上的版本；依赖Nginx环境；首先我们下载这个项目，输入:mvn clean package 将项目进行打包为jar文件,然后将application.properties和此jar项目放在一个文件夹中，然后复制该文件夹(这里为了清晰所以进行复制，实际不复制更改端口重启也行)，修改复制文件夹application.properties的端口，比如改为8086。Nginx 配置我们找到nginx的配置文件nginx.conf，该配置在nginx/conf/nginx.conf目录下，然后我们来修改该配置，新增如下配置:upstream pancm{ server 127.0.0.1:8085; server 127.0.0.1:8086; }upstream pancm：定义一个名称，随意就行；server + ip:端口 or 域名；如果不想使用Round Robin策略，也可以换成其他的。然后在server添加/修改如下配置: server { listen 80; server_name 127.0.0.1; location / { root html; proxy_pass http://pancm; proxy_connect_timeout 3s; proxy_read_timeout 5s; proxy_send_timeout 3s; index index.html index.htm; } error_page 500 502 503 504 /50x.html; location = /50x.html { root html; } }配置说明:server: 虚拟主机的名称，一个http中可以配置多个server；listen：Nginx默认的端口；server_name：Nginx服务的地址，可以使用域名，多个用空格分隔。proxy_pass：代理路径，一般配置upstream后面的名称用于实现负载均衡，可以直接配置ip进行跳转；nginx.conf 完整的配置:events { worker_connections 1024; } error_log nginx-error.log info; http { include mime.types; default_type application/octet-stream; sendfile on; keepalive_timeout 65; upstream pancm{ server 127.0.0.1:8085; server 127.0.0.1:8086; } server { listen 80; server_name 127.0.0.1; location / { root html; proxy_pass http://pancm; proxy_connect_timeout 3s; proxy_read_timeout 5s; proxy_send_timeout 3s; index index.html index.htm; } error_page 500 502 503 504 /50x.html; location = /50x.html { root html; } } }负载均衡测试在完成Nginx配置之后，启动Nginx。输入/usr/local/nginx/sbin/nginx -c /usr/local/nginx/conf/nginx.conf，如果已经启动可以使用/usr/local/nginx/sbin/nginx -s reload命令进行热加载配置文件，Windows直接点击Nginx目录下的nginx.exe或者 cmd运行start nginx进行启动，如果启动了依旧可以使用nginx -s reload进行热加载。Nginx启动完成之后，我们依次启动刚刚下载的springboot和复制更改端口的项目，输入:java -jar springboot-jsp-thymeleaf.jar启动。都启动成功之后，我们在浏览器输入服务的ip即可进行访问。示例图:注:这里我使用的是windows系统做测试，实际linux也是一样的。然后我们进行操作，并查看控制台日志！从上述示例图中我们进行4次界面刷新请求，最终平均分配到两个服务中去了，从上述的测试结果中我们实现了负载均衡。这里我再说一下使用Nginx的注意事项：在进行学习和测试的时候，使用nginx默认的端口实现负载均衡一般没有什么问题，但是当我们在项目中使用的时候，特别有登录界面的并且端口不是80的时候，会出现登录的界面无法跳转。进行调试的话会出现 net::ERR_NAME_NOT_RESOLVED这样的错误，出现这个原因的是因为nginx默认的端口是80，那么默认跳转的也是这个，所以出现这种情况的时候，需要在location 下添加proxy_set_header Host $host:port 这个配置，port 和listen 的端口保持一致就可以了。
1.对查询进行优化，应尽量避免全表扫描，首先应考虑在 where 及 order by 涉及的列上建立索引。2.应尽量避免在 where 子句中使用!=或<>操作符，否则将引擎放弃使用索引而进行全表扫描。3.应尽量避免在 where 子句中对字段进行 null 值判断，否则将导致引擎放弃使用索引而进行全表扫描，如：select id from t where num is null 可以在num上设置默认值0，确保表中num列没有null值，然后这样查询：select id from t where num=0 4.应尽量避免在 where 子句中使用 or 来连接条件，否则将导致引擎放弃使用索引而进行全表扫描，如：select id from t where num=10 or num=20 可以这样查询：select id from t where num=10 union all select id from t where num=20 5.下面的查询也将导致全表扫描：select id from t where name like '%abc%' 若要提高效率，可以考虑全文检索。6.in 和 not in 也要慎用，否则会导致全表扫描，如：select id from t where num in(1,2,3) 对于连续的数值，能用 between 就不要用 in 了：select id from t where num between 1 and 3 7.如果在 where 子句中使用参数，也会导致全表扫描。因为SQL只有在运行时才会解析局部变量，但优化程序不能将访问计划的选择推迟到运行时；它必须在编译时进行选择。然而，如果在编译时建立访问计划，变量的值还是未知的，因而无法作为索引选择的输入项。如下面语句将进行全表扫描：select id from t where num=@num 可以改为强制查询使用索引：select id from t with(index(索引名)) where num=@num 8.应尽量避免在 where 子句中对字段进行表达式操作，这将导致引擎放弃使用索引而进行全表扫描。如：select id from t where num/2=100 应改为:select id from t where num=100*2 9.应尽量避免在where子句中对字段进行函数操作，这将导致引擎放弃使用索引而进行全表扫描。如：select id from t where substring(name,1,3)='abc'--name以abc开头的id select id from t where datediff(day,createdate,'2005-11-30')=0--'2005-11-30'生成的id 应改为:select id from t where name like 'abc%' select id from t where createdate>='2005-11-30' and createdate<'2005-12-1' 10.不要在 where 子句中的“=”左边进行函数、算术运算或其他表达式运算，否则系统将可能无法正确使用索引。11.在使用索引字段作为条件时，如果该索引是复合索引，那么必须使用到该索引中的第一个字段作为条件时才能保证系统使用该索引，否则该索引将不会被使用，并且应尽可能的让字段顺序与索引顺序相一致。12.不要写一些没有意义的查询，如需要生成一个空表结构：select col1,col2 into #t from t where 1=0 这类代码不会返回任何结果集，但是会消耗系统资源的，应改成这样：create table #t(...) 13.很多时候用 exists 代替 in 是一个好的选择：select num from a where num in(select num from b) 用下面的语句替换：select num from a where exists(select 1 from b where num=a.num) 14.并不是所有索引对查询都有效，SQL是根据表中数据来进行查询优化的，当索引列有大量数据重复时，SQL查询可能不会去利用索引，如一表中有字段sex，male、female几乎各一半，那么即使在sex上建了索引也对查询效率起不了作用。15.索引并不是越多越好，索引固然可以提高相应的 select 的效率，但同时也降低了 insert 及 update 的效率，因为 insert 或 update 时有可能会重建索引，所以怎样建索引需要慎重考虑，视具体情况而定。一个表的索引数最好不要超过6个，若太多则应考虑一些不常使用到的列上建的索引是否有必要。16.应尽可能的避免更新 clustered 索引数据列，因为 clustered 索引数据列的顺序就是表记录的物理存储顺序，一旦该列值改变将导致整个表记录的顺序的调整，会耗费相当大的资源。若应用系统需要频繁更新 clustered 索引数据列，那么需要考虑是否应将该索引建为 clustered 索引。17.尽量使用数字型字段，若只含数值信息的字段尽量不要设计为字符型，这会降低查询和连接的性能，并会增加存储开销。这是因为引擎在处理查询和连接时会逐个比较字符串中每一个字符，而对于数字型而言只需要比较一次就够了。18.尽可能的使用 varchar/nvarchar 代替 char/nchar ，因为首先变长字段存储空间小，可以节省存储空间，其次对于查询来说，在一个相对较小的字段内搜索效率显然要高些。19.任何地方都不要使用 select from t ，用具体的字段列表代替“”，不要返回用不到的任何字段。20.尽量使用表变量来代替临时表。如果表变量包含大量数据，请注意索引非常有限（只有主键索引）。21.避免频繁创建和删除临时表，以减少系统表资源的消耗。22.临时表并不是不可使用，适当地使用它们可以使某些例程更有效，例如，当需要重复引用大型表或常用表中的某个数据集时。但是，对于一次性事件，最好使用导出表。23.在新建临时表时，如果一次性插入数据量很大，那么可以使用 select into 代替 create table，避免造成大量 log ，以提高速度；如果数据量不大，为了缓和系统表的资源，应先create table，然后insert。24.如果使用到了临时表，在存储过程的最后务必将所有的临时表显式删除，先 truncate table ，然后 drop table ，这样可以避免系统表的较长时间锁定。25.尽量避免使用游标，因为游标的效率较差，如果游标操作的数据超过1万行，那么就应该考虑改写。26.使用基于游标的方法或临时表方法之前，应先寻找基于集的解决方案来解决问题，基于集的方法通常更有效。27.与临时表一样，游标并不是不可使用。对小型数据集使用 FAST_FORWARD 游标通常要优于其他逐行处理方法，尤其是在必须引用几个表才能获得所需的数据时。在结果集中包括“合计”的例程通常要比使用游标执行的速度快。如果开发时间允许，基于游标的方法和基于集的方法都可以尝试一下，看哪一种方法的效果更好。28.在所有的存储过程和触发器的开始处设置 SET NOCOUNT ON ，在结束时设置 SET NOCOUNT OFF 。无需在执行存储过程和触发器的每个语句后向客户端发送 DONE_IN_PROC 消息。29.尽量避免向客户端返回大数据量，若数据量过大，应该考虑相应需求是否合理。30.尽量避免大事务操作，提高系统并发能力。