springboot第80集：Seata，优化 Java 代码，物联网IOT

Seata 是一款开源的分布式事务解决方案，致力于提供高性能和简单易用的分布式事务服务。Seata 将为用户提供了 AT、TCC、SAGA 和

XA 事务模式，为用户打造一站式的分布式解决方案。AT模式是阿里首推的模式,阿里云上有商用版本的GTS（Global Transaction Service 全局事务服务）

Seata的三大角色

在 Seata 的架构中，一共有三个角色：

TC (Transaction Coordinator) - 事务协调者

维护全局和分支事务的状态，驱动全局事务提交或回滚。

TM (Transaction Manager) - 事务管理器

定义全局事务的范围：开始全局事务、提交或回滚全局事务。

RM (Resource Manager) - 资源管理器

管理分支事务处理的资源，与TC交谈以注册分支事务和报告分支事务的状态，并驱动分支事务提交或回滚。

其中，TC 为单独部署的 Server 服务端，TM 和 RM 为嵌入到应用中的 Client 客户端。

1. Sentinel 配置：用于流量控制、熔断降级等。

• Nacos 配置：包括服务发现和配置中心的设置。

• 服务器配置：使用 Undertow 作为服务器，并设置了线程数。

• 管理端点配置：暴露所有端点以便监控和管理。

• Feign 配置：启用了 Sentinel 支持。

内存使用率达到了 90%，可以通过以下步骤排查原因：

1. 查看内存使用情况：

• 使用监控工具（如 Prometheus + Grafana，或 APM 工具）查看具体哪些组件或服务占用了大量内存。
• 在 Node1 上使用命令如 top 或 htop 查看进程的内存使用情况。

• 生成堆转储文件，使用工具如 VisualVM、Eclipse MAT（Memory Analyzer Tool）分析堆转储，查看哪些对象占用的内存最多。
• 检查是否存在内存泄漏（如未释放的对象）。

• 确认是否有循环引用或长生命周期的对象。
• 检查缓存实现，确保不会无限制增加缓存的对象。

• 查看是否有大规模的数据处理或查询，导致内存暴涨。优化 SQL 查询，避免一次性加载大量数据。

• 根据需要调整 JVM 的内存设置（如 -Xms 和 -Xmx 参数），确保为应用分配合适的内存。

• 检查与 Node1 交互的外部服务，是否有导致内存使用增加的请求。

建议的排查步骤：

• 查看字符串和字符数组的生成：确认是否有循环或频繁的字符串拼接，考虑使用 StringBuilder。

• 分析数据传输：确认是否有大文件或数据流的传输，是否可以优化。

• 审查缓存实现：检查 LinkedHashMap 的使用，确保及时清理不再使用的条目。

• 优化数组使用：对于大量的数组实例，考虑是否可以减少使用或使用更有效的数据结构。

• 字符串和字符管理：

• 避免使用大量短字符串，使用 StringBuilder 进行拼接，考虑将常用的字符串放入常量池中。

• 集合优化：

• 检查 HashMap 和 LinkedHashMap 的使用，确保它们的容量和负载因子设置合理，避免不必要的扩展和占用。

• 减少数组创建：

• 复用数组对象而不是频繁创建新数组，特别是在高频操作中。

• 线程管理：

• 使用线程池来管理线程，避免频繁创建和销毁线程。

• 性能监测：

• 使用 Java Profiler 工具（如 VisualVM、YourKit）进一步分析内存使用，查找内存泄漏或不必要的对象持有。

报告对最多包含一个实参的 Arrays.asList() 的调用。 在 JDK 9 及更高版本中，此类调用可以替换为 Collections.singletonList()、Collections.emptyList() 或 List.of()，从而节省内存。 特别是，无实参的 Collections.emptyList() 和 List.of() 总是返回共享实例，而无实参的 Arrays.asList() 每次调用时都会创建一个新对象。 注意：Collections.singletonList() 和 List.of() 返回的列表不可变，而列表返回的 Arrays.asList() 允许调用 set() 方法。 这在极少数情况下可能会破坏代码。

报告 equals() 被调用以将 String 与空字符串进行比较的情况。 在这种情况下，使用 .isEmpty() 更好，因为它可以准确显示您正在检查的内容。

报告可以优化并且以 count() 运算结束的 Stream API 调用链。 以下调用链替换为此检查： Collection.stream().count() → Collection.size(). 在 Java 8 中，Collection.stream().count() 实际上是通过迭代集合元素来进行计数，而 Collection.size() 对于大多数集合来说速度要快得多。 Stream.flatMap(Collection::stream).count() → Stream.mapToLong(Collection::size).sum(). 同样，不需要遍历所有嵌套集合。 相反，它们的大小可以相加。 Stream.filter(o -> ...).count() > 0 → Stream.anyMatch(o -> ...). 与初始调用不同，一旦找到匹配元素后 anyMatch() 就可以立即停止计算。 Stream.filter(o -> ...).count() == 0 → Stream.noneMatch(o -> ...). 与以上相似。 请注意，如果替换涉及 anyMatch() 等短路操作，那么中间流操作产生副作用时，可能会出现明显的行为变化。 在 Stream API 调用中通常应避免副作用。

在使用无参构造函数实例化集合后，立即报告 Collection.addAll() 和 Map.putAll() 调用。 此类结构可被替换为对形参化构造函数的单次调用，从而简化代码。 此外，对于某些集合，替换可能会更高效。

报告主要用作基元类型的包装器类型的局部变量。 在某些情况下，装箱可能会导致严重的性能损失，尤其是在循环中。 采用启发估计装箱操作次数。 例如，循环内的转换数量视为更多。

报告循环内部可以替换为批量方法的单一运算。 批量方法不仅更短，而且有时性能也更好。

报告用作 StringBuffer.append()、StringBuilder.append() 或 Appendable.append() 的实参的 String 串联。 此类调用可以有利地转变为现有 StringBuffer/Builder/Appendable 中的链式追加调用，从而节省额外的 StringBuffer/Builder 分配成本。 此检查将忽略编译时求值的 String 串联，在这种情况下，转换只会降低性能。

报告循环中的 String 串联。 由于每个 String 串联都会复制整个字符串，因此通常最好将其替换为对 StringBuilder.append() 或 StringBuffer.append() 的显式调用。

报告可被替换为 System.arraycopy() 调用的数组内容的手动复制。

报告可以使之为 static 的实例初始值设定项。 如果实例初始值设定项不引用其类的任何非 static 成员，则可以为 static。 static 初始值设定项在类解析后执行，而实例初始值设定项对此类的每个实例化执行。 此检查不报告匿名类中的实例空初始值设定项和初始值设定项。

Object的equals方法容易抛空指针异常，应使用常量或确定有值的对象来调用equals。

public void f(String str){
    String inner = "hi";
    if (inner.equals(str)) {
        System.out.println("hello world");
    }
}
        

List<Map<String, Object>>

字符串是否为空白 空白的定义如下： 1、为null 2、为不可见字符（如空格） 3、""

• Parameters:

• str 被检测的字符串

• Returns:

• 是否为空

所有的包装类对象之间值的比较，全部使用equals方法比较。 说明：对于Integer var=?在-128至127之间的赋值，Integer对象是在IntegerCache.cache产生，会复用已有对象，这个区间内的Integer值可以直接使用==进行判断，但是这个区间之外的所有数据，都会在堆上产生，并不会复用已有对象，这是一个大坑，推荐使用equals方法进行判断。

报告迭代集合或数组并且可以自动替换为增强型 for 循环（foreach 迭代语法）的 for 循环。

示例：

  for (Iterator<String> iterator = list.iterator(); iterator.hasNext(); ) {
    String item = iterator.next();
    System.out.println(item);
  }

在应用快速修复后：

  for (String item : list) {
    System.out.println(item);
  }

for (int i = 0; i < value.length; i++) {
    String item=value[i];
    boolean defend=doDefend(requestWrapper,servletResponse,item);
    if(defend){
        return;
    }
}

报告与 String.indexOf() 调用（可以替换为对 String.contains() 方法的调用）的比较。

报告声明为 StringBuffer 的变量，并建议将它们替换为 StringBuilder。 StringBuilder 是 StringBuffer 的非线程安全替换。

报告在循环内发生的对 java.lang.Thread.sleep() 的调用。 此类调用表示“忙等待”。 忙等待通常效率低下，并且可能导致意外死锁，因为忙等待线程不会释放锁定的资源。

示例：

  class X {
     volatile int x;
     public void waitX() throws Exception {
      while (x > 0) {
        Thread.sleep(10);// 警告：在循环中调用 'Thread.sleep()'，可能处于忙等待
      }
    }
  }

报告具有以下问题的 Javadoc 注释和标记： 标记名称无效 标记描述不完整 标记重复 缺少 Javadoc 描述

报告对目标和实参的类型不兼容的 equals() 的调用。 虽然此类调用理论上可能有用，但很可能是错误。 示例：

  new HashSet<String>().equals(new TreeSet<Integer>());

根据数据流分析，报告始终违反为 null 性约定、可能抛出异常或只是冗余的代码结构。 示例：

  if (array.length < index) {
    System.out.println(array[index]);
  } // 数组索引始终超出界限

  if (str == null) System.out.println("str is null");
  System.out.println(str.trim());
  // 最后一个语句可能会抛出 NPE

  @NotNull
  Integer square(@Nullable Integer input) {
      // 违反方法约定
      return input == null ? null : input * input;
  }

报告对最多包含一个实参的 Arrays.asList() 的调用。 在 JDK 9 及更高版本中，此类调用可以替换为 Collections.singletonList()、Collections.emptyList() 或 List.of()，从而节省内存。 特别是，无实参的 Collections.emptyList() 和 List.of() 总是返回共享实例，而无实参的 Arrays.asList() 每次调用时都会创建一个新对象。 注意：Collections.singletonList() 和 List.of() 返回的列表不可变，而列表返回的 Arrays.asList() 允许调用 set() 方法。 这在极少数情况下可能会破坏代码。 示例：

  List<String> empty = Arrays.asList();
  List<String> one = Arrays.asList("one");

在应用快速修复后：

  List<String> empty = Collections.emptyList();
  List<String> one = Collections.singletonList("one");

这是冗余的操作，因为任何装箱的值都会先自动拆箱，然后再次对该值装箱。 如果在内部循环中进行，此类代码可能会导致性能问题。

BigDecimal fee = new BigDecimal(0);
BigDecimal old;
if (OrderConstant.FEE_RULE_TIME.equals(feeRule.getRType())){
    old = new BigDecimal(bExchSvcOrderBO.getCycleM());
}else{
    old = new BigDecimal(bExchSvcOrder.getCycleNum());
    fee = new BigDecimal(feeRule.getCycleNum());
    old = old.add(fee);
    bExchSvcOrder.setCycleNum(old.toString());
    old = new BigDecimal(bExchSvcOrderBO.getCycleM());
}
fee = new BigDecimal(feeRule.getCycleM());
old = old.add(fee);
bExchSvcOrder.setCycleM(old.intValue());

报告对没有舍入模式实参的 divide() 或 setScale() 的调用。 在结果中不能表示精确值时（例如由于具有非终止十进制扩展），此类调用可能导致 ArithmeticException。 指定舍入模式可防止 ArithmeticException。 示例：

  BigDecimal.valueOf(1).divide(BigDecimal.valueOf(3));

报告对无实参的 Throwable.printStackTrace() 的调用。 此类语句通常用于临时调试，应当从生产代码中移除，或者替换为更稳健的日志记录工具。

报告 equals() 被调用以将 String 与空字符串进行比较的情况。 在这种情况下，使用 .isEmpty() 更好，因为它可以准确显示您正在检查的内容。 示例：

  void checkString(String s){
    if ("".equals(s)) throw new IllegalArgumentException();
  }

在应用快速修复后：

  void checkString(String s){
    if (s != null && s.isEmpty()) throw new IllegalArgumentException();
  }

报告可被分别替换为 Files.newInputStream() 或 Files.newOutputStream() 调用的 new FileInputStream() 或 new FileOutputStream() 表达式。 使用 Files 方法创建的流通常比使用流构造函数创建的流更有效。

使用 @Transactional： 在代理模式（默认）下，只有通过代理进入的外部方法调用会被拦截。 这意味着，自调用（实际上是目标对象内的方法调用目标对象的另一个方法）在运行时不会导致实际的事务，即使被调用的方法被标记为 @Transactional。

• Mysql
• Redis
• RabbitMQ
• MongoDB
• Nacos
• getway 网关服务
• auth 认证服务
• manager 管理服务
• data 数据服务
• virtual 虚拟驱动

使用的RabbitMQ的mqtt插件

数据库（MySQL、Redis、MongoDB）和消息组件（RabbitMQ）均采用容器启动

.
├──  资源文件,如sh,sql等
├── -api gRpc定义的接口结构
├── -center 平台中心模块
├── -common 平台公共模块
├── -driver 平台驱动模块
├── -driver-sdk 平台驱动SDK模块
└── -gateway 平台网关模块

.
├── -center-auth 授权模块,主要负责接口权限
├── -center-data 数据模块,主要负责驱动数据处理
└── -center-manager 管理模块

.
├──  git脚本
├── -common-api api
├── -common-auth 授权相关
├── -common-constant 常量相关
├── -common-exception 异常相关
├── -common-influxdata influxDataDB相关
├── -common-log 日志相关
├── -common-model 模型相关
├── -common-mongo mongoDB相关
├── -common-mqtt mqtt相关
├── -common-mysql 数据库相关
├── -common-public 公共配置相关
├── -common-quartz 定时任务
├── -common-rabbitmq 消息队列相关
├── -common-redis 缓存相关
├── -common-thread 线程相关
└── -common-web web服务配置

├── -driver-dtu-yeecom Dtu驱动相关
├── -driver-edge-gateway 边缘网关相关
├── -driver-listening-virtual 虚拟网关相关
├── -driver-lwm2m Lwm2m&Coap相关
├── -driver-modbus-tcp modbusTcp相关
├── -driver-mqtt mqtt相关
├── -driver-opc-da opc-da相关
├── -driver-opc-ua opc-ua相关
├── -driver-plcs7 plcs7相关
├── -driver-virtual 测试驱动相关
└── -driver-weather-amap 高德地图天气相关

存储策略

• mogodb
• influxdb
• redis
• opentsdb
• elasticsearch
• TDengine

/**
 * 根据枚举索引获取枚举
 *
 * @param index 索引
 * @return {@link ApiTypeFlagEnum}
 */
public static ApiTypeFlagEnum ofIndex(Byte index) {
    // 使用流和过滤器查找匹配的枚举
    return Arrays.stream(ApiTypeFlagEnum.values())
                 .filter(type -> type.getIndex().equals(index))
                 .findFirst()
                 .orElse(null); // 如果没有找到匹配的枚举，返回null
}

/**
 * 根据枚举编码获取枚举
 *
 * @param code 编码
 * @return {@link ApiTypeFlagEnum}
 */
public static ApiTypeFlagEnum ofCode(String code) {
    // 使用流和过滤器查找匹配的枚举
    return Arrays.stream(ApiTypeFlagEnum.values())
                 .filter(type -> type.getCode().equals(code))
                 .findFirst()
                 .orElse(null); // 如果没有找到匹配的枚举，返回null
}

/**
 * 根据枚举内容获取枚举
 *
 * @param name 枚举内容
 * @return {@link ApiTypeFlagEnum}
 */
public static ApiTypeFlagEnum ofName(String name) {
    try {
        // 使用valueOf方法获取枚举
        return valueOf(name);
    } catch (IllegalArgumentException e) {
        // 如果name不匹配任何枚举，捕获异常并返回null
        return null;
    }
}


@Mapping(target = "apiExt", ignore = true) // 忽略映射到目标对象的apiExt字段
@Mapping(target = "enableFlag", ignore = true) // 忽略映射到目标对象的enableFlag字段
@Mapping(target = "apiTypeFlag", ignore = true) // 忽略映射到目标对象的apiTypeFlag字段
@Mapping(target = "deleted", ignore = true) // 忽略映射到目标对象的deleted字段


@Getter
@Setter
@Builder
@NoArgsConstructor
@AllArgsConstructor

1. EMQX集群部署，单节点部署，由单节点迁移到集群部署模式
2. Mongo集群部署，单节点部署，由单节点迁移到集群部署模式
3. RabbitMQ集群部署，单节点部署，由单节点迁移到集群部署模式
4. Redis集群部署，单节点部署，由单节点迁移到集群部署模式

1. EMQX 单节点部署

• 下载和安装：

• 从 EMQX 官方网站 下载适合的版本，并根据操作系统进行安装。

• 启动节点：

  cd /path/to/emqx
  ./bin/emqx start
  

• 配置：

• 修改 etc/emqx.conf 根据需要配置 MQTT 相关设置。

• 验证运行：

• 使用 ./bin/emqx_ctl status 检查节点状态。

2. EMQX 集群部署

• 准备集群节点：

• 在每个节点上安装 EMQX。

• 配置集群：

• 在 etc/emqx.conf 中设置集群节点信息，例如：

  cluster.1 = node@hostname1
  cluster.2 = node@hostname2
  

• 启动节点：

• 启动每个节点：

  ./bin/emqx start
  

• 加入集群：

• 在主节点上运行：

  ./bin/emqx_ctl cluster join node@hostname1
  

• 检查集群状态：

• 使用命令：

  ./bin/emqx_ctl cluster status
  

3. 从单节点迁移到集群部署

• 备份配置：

• 备份当前单节点的 emqx.conf 配置。

• 安装集群节点：

• 在新的服务器上安装 EMQX。

• 配置主节点：

• 在主节点的配置文件中，设置要加入的集群节点。

• 启动并加入集群：

• 启动主节点后，使用 emqx_ctl cluster join 命令将新节点加入集群。

• 数据迁移（如果需要）：

• 如果需要迁移数据，可以根据需要使用备份和恢复功能。

• 验证集群状态：

• 使用 emqx_ctl cluster status 检查所有节点是否正常工作。

EMQX 单节点部署

下载和安装

# 下载 EMQX（假设下载最新的 Linux 版本）
wget https://www.emqx.io/downloads/broker/v5.0.0/emqx-ubuntu20.04-v5.0.0_amd64.deb

# 安装 EMQX
sudo dpkg -i emqx-ubuntu20.04-v5.0.0_amd64.deb

启动节点

# 启动 EMQX
sudo /usr/lib/emqx/bin/emqx start

验证运行

# 检查节点状态
sudo /usr/lib/emqx/bin/emqx_ctl status

2. EMQX 集群部署

准备集群节点

在每个节点上执行以下命令：

# 在节点 1 上
sudo /usr/lib/emqx/bin/emqx start

# 在节点 2 上
sudo /usr/lib/emqx/bin/emqx start

配置集群

在每个节点的配置文件中（如 etc/emqx.conf）进行如下配置：

# 在节点 1 的 emqx.conf 中
cluster.1 = node@node1-hostname

# 在节点 2 的 emqx.conf 中
cluster.2 = node@node2-hostname

加入集群

在节点 1 上（主节点）运行：

# 从节点 2 加入集群
sudo /usr/lib/emqx/bin/emqx_ctl cluster join node@node1-hostname

检查集群状态

在节点 1 上：

# 查看集群状态
sudo /usr/lib/emqx/bin/emqx_ctl cluster status

3. 从单节点迁移到集群部署

备份配置

# 备份当前配置文件
cp /usr/lib/emqx/etc/emqx.conf /usr/lib/emqx/etc/emqx.conf.bak

安装新节点

在新服务器上执行：

# 下载并安装 EMQX（类似于单节点部署）
wget https://www.emqx.io/downloads/broker/v5.0.0/emqx-ubuntu20.04-v5.0.0_amd64.deb
sudo dpkg -i emqx-ubuntu20.04-v5.0.0_amd64.deb

配置主节点

在主节点的 emqx.conf 文件中添加其他节点的配置。

启动并加入集群

在主节点上启动并记录其节点名称：

sudo /usr/lib/emqx/bin/emqx start

在新节点上启动并加入集群：

sudo /usr/lib/emqx/bin/emqx start
sudo /usr/lib/emqx/bin/emqx_ctl cluster join node@主节点的hostname

验证集群状态

在主节点上：

sudo /usr/lib/emqx/bin/emqx_ctl cluster status

1. MongoDB 单节点部署

下载和安装

# 更新包索引
sudo apt update

# 安装 MongoDB（以 Ubuntu 为例）
sudo apt install -y mongodb

# 启动 MongoDB 服务
sudo systemctl start mongodb

# 设置开机自启
sudo systemctl enable mongodb

验证运行

# 检查 MongoDB 服务状态
sudo systemctl status mongodb

# 进入 MongoDB shell
mongo

2. MongoDB 集群部署（副本集）

准备集群节点

在每个节点上安装 MongoDB（与单节点相同）。

配置副本集

• 在每个节点的配置文件（如 /etc/mongod.conf）中进行如下配置：

  replication:
    replSetName: "rs0"
  

启动节点

在每个节点上启动 MongoDB：

sudo systemctl start mongodb

初始化副本集

在主节点上（假设 IP 地址为 192.168.1.1）：

# 进入 MongoDB shell
mongo

# 初始化副本集
rs.initiate({
  _id: "rs0",
  members: [
    { _id: 0, host: "192.168.1.1:27017" },
    { _id: 1, host: "192.168.1.2:27017" },
    { _id: 2, host: "192.168.1.3:27017" }
  ]
})

验证集群状态

# 查看副本集状态
rs.status()

3. 从单节点迁移到集群部署

备份数据

# 使用 mongodump 备份数据
mongodump --out /path/to/backup

安装新节点

在新服务器上安装 MongoDB（与单节点相同）。

配置副本集

在新节点的配置文件中，添加相同的副本集配置：

replication:
  replSetName: "rs0"

启动节点

在每个新节点上启动 MongoDB：

sudo systemctl start mongodb

添加新节点到副本集

在原主节点上：

# 进入 MongoDB shell
mongo

# 添加新节点
rs.add("新节点的IP:27017")

验证集群状态

# 查看副本集状态
rs.status()

注意事项

• 确保所有节点的 MongoDB 版本相同。
• 在集群中，确保各节点之间的网络连接正常。
• 如果数据量较大，考虑使用 mongorestore 从备份恢复数据。

1. Redis 单节点部署

下载和安装

# 更新包索引
sudo apt update

# 安装 Redis（以 Ubuntu 为例）
sudo apt install -y redis-server

启动 Redis

# 启动 Redis 服务
sudo systemctl start redis-server

# 设置开机自启
sudo systemctl enable redis-server

验证运行

# 检查 Redis 服务状态
sudo systemctl status redis-server

# 进入 Redis CLI
redis-cli ping
# 应返回 "PONG"

2. Redis 集群部署

准备集群节点

在每个节点上安装 Redis（与单节点相同）。

配置集群

• 在每个节点的配置文件（如 /etc/redis/redis.conf）中进行如下配置：

  cluster-enabled yes
  cluster-config-file nodes.conf
  cluster-node-timeout 5000
  

启动节点

在每个节点上启动 Redis：

sudo systemctl start redis-server

创建集群

使用 redis-cli 创建集群（假设有三个节点，IP 地址分别为 192.168.1.1, 192.168.1.2, 192.168.1.3）：

# 创建集群
redis-cli --cluster create \
  192.168.1.1:6379 \
  192.168.1.2:6379 \
  192.168.1.3:6379 \
  --cluster-replicas 1

验证集群状态

# 检查集群状态
redis-cli -c -h 192.168.1.1 -p 6379 cluster info

3. 从单节点迁移到集群部署

备份数据

# 使用 redis-cli 备份数据
redis-cli save
# 备份生成在 /var/lib/redis/dump.rdb

安装新节点

在新服务器上安装 Redis（与单节点相同）。

配置集群

在新节点的配置文件中，添加集群相关配置（与之前相同）。

启动新节点

在新节点上启动 Redis：

sudo systemctl start redis-server

创建集群

在主节点上，将新节点添加到集群：

# 假设新节点的 IP 为 192.168.1.4
redis-cli --cluster add-node 192.168.1.4:6379 192.168.1.1:6379

验证集群状态

# 检查集群状态
redis-cli -c -h 192.168.1.1 -p 6379 cluster info

注意事项

• 确保每个节点的 Redis 版本一致。
• 所有节点之间必须可以互相访问。
• 进行迁移时，考虑使用 redis-shake 或其他工具进行数据同步。

1. EMQX集群部署，单节点部署，由单节点迁移到集群部署模式
2. Mongo集群部署，单节点部署，由单节点迁移到集群部署模式
3. RabbitMQ集群部署，单节点部署，由单节点迁移到集群部署模式
4. Redis集群部署，单节点部署，由单节点迁移到集群部署模式
5. kafka集群部署，单节点部署，由单节点迁移到集群部署模式
6. elasticsearch集群部署，单节点部署，由单节点迁移到集群部署模式
7. cassandra集群部署，单节点部署，由单节点迁移到集群部署模式
8. zookeeper集群部署，单节点部署，由单节点迁移到集群部署模式
9. mysql集群部署，单节点部署，由单节点迁移到集群部署模式

1. RabbitMQ 单节点部署

下载和安装

# 更新包索引
sudo apt update

# 安装 RabbitMQ（以 Ubuntu 为例）
sudo apt install -y rabbitmq-server

启动 RabbitMQ

# 启动 RabbitMQ 服务
sudo systemctl start rabbitmq-server

# 设置开机自启
sudo systemctl enable rabbitmq-server

验证运行

# 检查 RabbitMQ 服务状态
sudo systemctl status rabbitmq-server

# 进入 RabbitMQ 管理界面（默认在 http://localhost:15672）
# 默认用户名和密码都是 'guest'

2. RabbitMQ 集群部署

准备集群节点

在每个节点上安装 RabbitMQ（与单节点相同）。

配置集群

• 确保 Erlang 版本一致，并在每个节点上设置相同的 cookie：

  sudo nano /var/lib/rabbitmq/.erlang.cookie
  

  确保所有节点的 .erlang.cookie 文件内容相同（通常为随机字符串）。

启动节点

在每个节点上启动 RabbitMQ：

sudo systemctl start rabbitmq-server

初始化集群

选择一个节点作为主节点，假设节点名称为 rabbit1。在其他节点上运行：

# 在 rabbit2 和 rabbit3 上，加入主节点的集群
sudo rabbitmqctl stop_app
sudo rabbitmqctl join_cluster rabbit@rabbit1
sudo rabbitmqctl start_app

验证集群状态

在主节点上：

# 查看集群状态
sudo rabbitmqctl cluster_status

3. 从单节点迁移到集群部署

备份数据

在单节点上，使用以下命令备份队列数据：

# 备份所有队列数据
sudo rabbitmqctl export_definitions /path/to/backup.json

安装新节点

在新服务器上安装 RabbitMQ（与单节点相同）。

配置集群

在新节点上设置相同的 Erlang cookie，并确保网络互通。

启动新节点

在新节点上启动 RabbitMQ：

sudo systemctl start rabbitmq-server

加入新节点到集群

在单节点（主节点）上，使用以下命令将新节点加入集群：

# 假设新节点的名称为 rabbit2
sudo rabbitmqctl stop_app
sudo rabbitmqctl join_cluster rabbit@rabbit2
sudo rabbitmqctl start_app

恢复数据（可选）

如果需要，可以将备份的数据导入新集群：

# 从备份文件恢复队列数据
sudo rabbitmqctl import_definitions /path/to/backup.json

验证集群状态

在主节点上：

# 查看集群状态
sudo rabbitmqctl cluster_status

注意事项

• 确保所有节点的 RabbitMQ 和 Erlang 版本一致。
• 运行 RabbitMQ 集群时，确保所有节点的网络连接正常。
• 如果需要跨数据中心部署，考虑使用网络分区策略。

1. Kafka 单节点部署

下载和安装

# 下载 Kafka（以 Kafka 2.8.0 为例）
wget https://downloads.apache.org/kafka/2.8.0/kafka_2.12-2.8.0.tgz

# 解压文件
tar -xzf kafka_2.12-2.8.0.tgz
cd kafka_2.12-2.8.0

启动 Zookeeper

# 启动 Zookeeper（Kafka 需要 Zookeeper）
bin/zookeeper-server-start.sh config/zookeeper.properties

启动 Kafka

# 在另一个终端中启动 Kafka
bin/kafka-server-start.sh config/server.properties

验证运行

# 创建一个主题以进行测试
bin/kafka-topics.sh --create --topic test --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

# 列出主题
bin/kafka-topics.sh --list --bootstrap-server localhost:9092

2. Kafka 集群部署

准备集群节点

在每个节点上安装 Kafka（与单节点相同）。

配置集群

• 在每个节点的 server.properties 文件中，修改以下配置：

  # 每个节点的唯一 ID（在每个节点上设置不同的 id）
  broker.id=0  # 对于第一个节点
  # 对于第二个节点
  broker.id=1
  # 对于第三个节点
  broker.id=2
  
  # Zookeeper 地址
  zookeeper.connect=localhost:2181,localhost:2182,localhost:2183  # 根据节点调整
  
  # 数据存储路径
  log.dirs=/var/lib/kafka-logs
  

启动 Zookeeper

在每个节点上启动 Zookeeper（可以使用一个节点的 Zookeeper）。

启动 Kafka

在每个节点上启动 Kafka：

# 启动 Kafka
bin/kafka-server-start.sh config/server.properties

验证集群状态

在任一节点上：

# 列出主题并确认所有 broker
bin/kafka-topics.sh --describe --topic test --bootstrap-server localhost:9092

3. 从单节点迁移到集群部署

备份数据

在单节点上，使用以下命令备份数据（如果需要）：

# 可以使用工具如 kafka-dump-log 进行数据备份
# 或直接复制 log.dirs 指定的目录

安装新节点

在新服务器上安装 Kafka（与单节点相同）。

配置集群

在新节点的 server.properties 文件中，设置唯一的 broker.id 和 Zookeeper 地址。

启动 Zookeeper

在每个节点上启动 Zookeeper（可以使用一个节点的 Zookeeper）。

启动新节点的 Kafka

在新节点上启动 Kafka：

bin/kafka-server-start.sh config/server.properties

加入集群

在主节点上确认新节点已加入：

# 列出所有 brokers
bin/kafka-topics.sh --describe --topic test --bootstrap-server localhost:9092

注意事项

• 确保所有节点的 Kafka 版本一致。
• Zookeeper 配置要正确，确保所有节点都可以访问。
• 数据迁移时，可以考虑使用 Kafka 的复制功能，或通过消费者将数据重新写入到新的集群。

1. Elasticsearch 单节点部署

下载和安装

# 下载 Elasticsearch（以 8.0.0 为例）
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.0.0-linux-x86_64.tar.gz

# 解压文件
tar -xzf elasticsearch-8.0.0-linux-x86_64.tar.gz
cd elasticsearch-8.0.0

启动 Elasticsearch

# 启动 Elasticsearch
bin/elasticsearch

验证运行

• 打开一个新的终端，使用 curl 验证 Elasticsearch 是否正常运行：

curl -X GET "localhost:9200/"

• 应返回包含版本信息的 JSON 格式响应。

2. Elasticsearch 集群部署

准备集群节点

在每个节点上安装 Elasticsearch（与单节点相同）。

配置集群

• 在每个节点的配置文件（如 config/elasticsearch.yml）中进行如下配置：

cluster.name: my-cluster   # 集群名称
node.name: node-1           # 节点名称，节点 2 和 3 需要相应更改
network.host: 0.0.0.0       # 允许所有网络接口
http.port: 9200              # HTTP 端口

# 集群中的其他节点
discovery.seed_hosts: ["192.168.1.2:9200", "192.168.1.3:9200"]  # 其他节点的 IP 地址
cluster.initial_master_nodes: ["node-1", "node-2", "node-3"]  # 初始主节点

启动节点

在每个节点上启动 Elasticsearch：

bin/elasticsearch

验证集群状态

在任一节点上使用 curl 验证集群状态：

curl -X GET "localhost:9200/_cluster/health?pretty"

3. 从单节点迁移到集群部署

备份数据

在单节点上，可以使用快照功能备份数据：

# 创建快照仓库（以文件系统为例）
curl -X PUT "localhost:9200/_snapshot/my_backup" -H 'Content-Type: application/json' -d '{
  "type": "fs",
  "settings": {
    "location": "/path/to/backup"
  }
}'
# 创建快照
curl -X PUT "localhost:9200/_snapshot/my_backup/snapshot_1?wait_for_completion=true"

安装新节点

在新服务器上安装 Elasticsearch（与单节点相同）。

配置集群

在新节点的配置文件中，设置相应的 cluster.name 和 discovery.seed_hosts，确保每个节点的 node.name 唯一。

启动新节点

在新节点上启动 Elasticsearch：

bin/elasticsearch

加入集群

在主节点上，确认新节点已加入集群：

curl -X GET "localhost:9200/_cat/nodes?v"

恢复数据（可选）

如果需要将数据恢复到新集群，可以从快照中恢复：

# 恢复快照
curl -X POST "localhost:9200/_snapshot/my_backup/snapshot_1/_restore"

注意事项

• 确保所有节点的 Elasticsearch 版本一致。
• 各节点的网络设置要正确，确保互相可达。
• 在生产环境中，考虑使用专用的配置管理工具（如 Ansible、Chef 等）来管理节点配置。

1. Cassandra 单节点部署

下载和安装

# 下载 Cassandra（以 4.0.0 为例）
wget https://downloads.apache.org/cassandra/4.0.0/apache-cassandra-4.0.0-bin.tar.gz

# 解压文件
tar -xzf apache-cassandra-4.0.0-bin.tar.gz
cd apache-cassandra-4.0.0

启动 Cassandra

# 启动 Cassandra
bin/cassandra -f

验证运行

• 打开另一个终端，使用 cqlsh 验证 Cassandra 是否正常运行：

# 进入 CQL shell
bin/cqlsh

2. Cassandra 集群部署

准备集群节点

在每个节点上安装 Cassandra（与单节点相同）。

配置集群

• 在每个节点的配置文件（如 conf/cassandra.yaml）中进行如下配置：

# 集群名称
cluster_name: 'MyCluster'

# 节点的唯一 ID
seeds: '192.168.1.1'  # 主节点的 IP 地址

# 监听地址（每个节点设置其自身 IP）
listen_address: 192.168.1.1  # 节点 1 的 IP
# 节点 2 和 3 需要相应更改

# RPC 地址（客户端连接的地址）
rpc_address: 0.0.0.0

# 数据存储路径
data_file_directories:
    - /var/lib/cassandra/data

启动节点

在每个节点上启动 Cassandra：

bin/cassandra -f

验证集群状态

在任一节点上，使用 cqlsh 验证集群状态：

# 查看节点信息
bin/nodetool status

3. 从单节点迁移到集群部署

备份数据

在单节点上，可以使用 nodetool snapshot 命令进行数据备份：

# 创建快照
bin/nodetool snapshot

安装新节点

在新服务器上安装 Cassandra（与单节点相同）。

配置集群

在新节点的 cassandra.yaml 文件中，设置相应的 seeds 和 listen_address。

启动新节点

在新节点上启动 Cassandra：

bin/cassandra -f

加入集群

在主节点上，确认新节点已加入集群：

# 查看节点信息
bin/nodetool status

恢复数据（可选）

如果需要将数据恢复到新集群，可以从快照中恢复。将快照数据复制到新节点的相应目录（如 /var/lib/cassandra/data），并启动 Cassandra。

注意事项

• 确保所有节点的 Cassandra 版本一致。
• 各节点的网络设置要正确，确保互相可达。
• 在生产环境中，建议使用工具如 Ansible 或 Puppet 来管理集群配置和部署。

加群联系作者vx：xiaoda0423

仓库地址：https://github.com/webVueBlog/JavaGuideInterview