# kafka-map文件夹
cd /usr/local/kafka-map
# 根据需求自行修改配置
vi application.yml
# 启动
java -jar kafka-map.jar

byte minByte = -128; byte maxByte = 127;

用于表示一个 8 位（1 字节）有符号整数。它的值范围是 -128（-2^7）到 127（2^7 - 1）。

由于 byte 类型占用的空间较小，它通常用于处理大量的数据，如文件读写、网络传输等场景，以节省内存空间。

用于声明一个字符类型的变量。char 类型的变量可以存储任意的 Unicode 字符，可以使用单引号将字符括起来来表示。

char c = 'A';

用于继续下一个循环，可以在指定条件下跳过其余代码。

for (int i = 1; i <= 10; i++) {if (i % 2 == 0) {continue;}System.out.println(i);
}

do 后紧跟循环体。

int i = 1;
do {System.out.println(i);i++;
} while (i <= 10);

int i = 1;
do {System.out.println(i);i++;
} while (i <= 10);

do-while 循环与 while 循环类似，不同之处在于 do-while 循环会先执行循环体中的代码，然后再检查循环条件。因此，do-while 循环至少会执行一次循环体中的代码。

用于声明一个双精度浮点类型的变量。

double a = 3.14;
double b = 2.0;
double c = a + b;

public enum PlayerType { TENNIS, FOOTBALL, BASKETBALL }

class Animal {public void eat() {System.out.println("动物正在吃东西");}
}class Dog extends Animal {public void bark() {System.out.println("狗在汪汪叫");}
}public class ExtendsDemo {public static void main(String[] args) {Dog dog = new Dog();dog.eat();dog.bark();}
}

Animal 类中有一个 eat() 方法，输出字符串 "动物正在吃东西"。Dog 类继承自 Animal 类，并定义了一个 bark() 方法，输出字符串 "狗在汪汪叫"。

final 方法表示一个不能被子类重写的方法。这在设计类时，确保某个方法的实现不会被子类修改时非常有用。

class Animal {final void makeSound() {System.out.println("动物发出声音.");}
}class Dog extends Animal {// 错误: 无法覆盖来自 Animal 的 final 方法// void makeSound() {//     System.out.println("狗吠叫.");// }
}public class Main {public static void main(String[] args) {Dog dog = new Dog();dog.makeSound();}
}

final 类表示一个不能被继承的类。这在设计类时，确保其不会被其他类继承时非常有用。[String 类就是 final 的]

final class Animal {void makeSound() {System.out.println("动物发出声音.");}
}// 错误: 类型 Dog 无法继承 final 类 Animal
// class Dog extends Animal {
//     void makeSound() {
//         System.out.println("狗吠叫.");
//     }
// }public class Main {public static void main(String[] args) {Animal animal = new Animal();animal.makeSound();}
}

用于声明一个 for 循环，如果循环次数是固定的，建议使用 for 循环。

[用于判断对象是否属于某个类型（class）]

例如，假设有一个 Person 类和一个 Student 类，Student 类继承自 Person 类，可以使用 instanceof 运算符来判断一个对象是否为 Person 类或其子类的实例：
Person p = new Student();
if (p instanceof Person) {System.out.println("p is an instance of Person");
}
if (p instanceof Student) {System.out.println("p is an instance of Student");
}

用于表示长整数值。

long x;           // 声明一个 long 类型的变量 x
x = 10000000000L; // 将长整数值 10000000000 赋给变量 x，需要在数字后面加上 L 或 l 表示这是一个 long 类型的值
long y = 20000000000L; // 声明并初始化一个 long 类型的变量 y，赋值为长整数值 20000000000

本地方法是指在 Java 代码中声明但在本地代码（通常是 C 或 C++ 代码）中实现的方法，它通常用于与操作系统或其他本地库进行交互。

public native void nativeMethod();

如果一个变量是空的（什么引用也没有指向），就可以将它赋值为 null，和[空指针]异常息息相关。

String str = null; // 声明一个字符串引用，初始化为 null
MyClass obj = null; // 声明一个 MyClass 类型的引用，初始化为 null

表示方法或变量只对当前类可见。

public class MyClass {private int x; // 私有属性 x，只能在当前类的内部访问private void foo() {// 私有方法 foo，只能在当前类的内部调用}
}

一个访问权限修饰符，表示方法或变量对同一包内的类和所有子类可见。

package com.example.mypackage;public class MyBaseClass {protected int x; // 受保护的属性 x，可以被子类和同一包中的其他类访问protected void foo() {// 受保护的方法 foo，可以被子类和同一包中的其他类调用}
}package com.example.mypackage;public class MySubClass extends MyBaseClass {public void bar() {x = 10; // 可以访问 MyBaseClass 中的受保护属性 xfoo(); // 可以调用 MyBaseClass 中的受保护方法 foo}
}

一个访问权限修饰符，除了可以声明方法和变量（所有类可见），还可以声明类。main() 方法必须声明为 public。

public class MyClass {public int x; // 公有属性 x，可以被任何类访问public void foo() {// 公有方法 foo，可以被任何类调用}
}

用于表示短整数，占用 2 个字节（16 位）的内存空间。

表示该变量或方法是[静态变量或静态方法]

public class MyClass {public static int x; // 静态变量 x，属于类的成员public static void foo() {// 静态方法 foo，属于类的成员}
}

在这个示例中，MyClass 类有一个静态变量 x 和一个静态方法 foo()。这些成员属于类的成员，可以通过类名直接访问，不需要创建对象。

可用于调用父类的方法或者字段。

class Animal {protected String name;public Animal(String name) {this.name = name;}public void eat() {System.out.println(name + " is eating.");}
}public class Dog extends Animal {public Dog(String name) {super(name); // 调用父类的构造方法}public void bark() {System.out.println(name + " is barking.");}public void eat() {super.eat(); // 调用父类的方法System.out.println(name + " is eating bones.");}
}

[用于指定多线程代码中的同步方法、变量或者代码块]

public class MyClass {private int count;public synchronized void increment() {count++; // 同步方法}public void doSomething() {synchronized(this) { // 同步代码块// 执行一些需要同步的操作}}
}

[可用于在方法或构造方法中引用当前对象]

public class MyClass {private int num;public MyClass(int num) {this.num = num; // 使用 this 关键字引用当前对象的成员变量}public void doSomething() {System.out.println("Doing something with " + this.num); // 使用 this 关键字引用当前对象的成员变量}public MyClass getThis() {return this; // 返回当前对象本身}
}

[修饰的字段不会被序列化]

public class MyClass implements Serializable {private int id;private String name;private transient String password;public MyClass(int id, String name, String password) {this.id = id;this.name = name;this.password = password;}// 省略 getter 和 setter 方法@Overridepublic String toString() {return "MyClass{" +"id=" + id +", name='" + name + ''' +", password='" + password + ''' +'}';}
}

保证不同线程对它修饰的变量进行操作时的[可见性]，即一个线程修改了某个变量的值，新值对其他线程来说是立即可见的。

public class MyThread extends Thread {private volatile boolean running = true;@Overridepublic void run() {while (running) {// 线程执行的代码}}public void stopThread() {running = false;}
}

技术栈

• Spring Cloud: 服务注册与发现 (Eureka/Nacos)

• Spring Boot: 微服务基础

• Spring Data JPA: ORM 框架

• MySQL: 数据库

• Spring Security: 用户认证与授权

• OpenFeign: 服务间调用

• Spring Cloud Gateway: API 网关

• Spring Boot Admin: 微服务监控

---

服务拆分

我们将服务拆分为以下几个微服务：

1. 服务注册中心：eureka-server

2. 网关服务：gateway-service

3. 用户服务：user-service（管理用户相关）

4. 房屋服务：house-service（管理房屋相关）

5. 帖子服务：post-service（管理帖子、评论等）

6. 消息服务：message-service（处理消息）

修复 Kafka 的文件锁问题

如果文件锁定问题持续出现，可以尝试以下解决方案：

1. 切换日志目录： 修改 server.properties 中的 log.dirs 为新的目录：

   log.dirs=D:/software/kafka/logs-new

2. 配置日志清理： 避免日志文件过大或过多：

   log.retention.hours=168  # 保留7天的日志
   log.segment.bytes=1073741824  # 每个日志段1GB
   log.cleanup.policy=delete  # 自动清理旧日志

Linux环境下安装Elasticsearch

1 前言2 Linux下安装Elasticsearch2.1 下载和解压安装包2.2 解决es强依赖jdk问题2.3 解决内存不足问题2.4 创建专用用户启动ES2.5 修改ES核心配置信息2.6 vm.max_map_count [65530] is too low问题2.7 可能遇到的max file descriptors [4096]问题2.8 ES服务的启动与停止2.9 为Elasticsearch设置登录密码3 安装elasticsearch-head插件3.1 安装node环境3.2 安装elasticsearch-head3.2.1 下载elasticsearch-head安装包3.2.2 解压zip包3.2.3 npm install3.2.4 启动elasticsearch-head服务3.2.5 访问elasticsearch-head服务3.2.6 使用Head插件访问elasticsearch

Kafka的bug解决方法 -ERROR Shutdown broker because all log dirs in xxx have failed(kafka.log.LogManager)

删除kafka 中一个 topic 之后 kafka 停掉了，再次启动失败报了一下错误

ERROR Shutdown broker because all log dirs in C:\temp\kafka-logs have failed (kafka.log.LogManager)

解决方案 ：

• 删除指定文件夹kafka-logs 下日志；

• 删除zookeeper 日志

• 再次重新启动

• 注意： 清空全部日志是危险的操作，建议先进行备份。

1）基本数据类型。

基本数据类型是 Java 语言操作数据的基础，包括 boolean、char、byte、short、int、long、float 和 double，共 8 种。

2）引用数据类型。

除了基本数据类型以外的类型，都是所谓的引用类型。常见的有[数组]、class（也就是[类]），以及[接口]（指向的是实现接口的类的对象）。

[变量]可以分为局部变量、成员变量、静态变量。

当变量是局部变量的时候，必须得先初始化，否则编译器不允许你使用它

当变量是成员变量或者静态变量时，可以不进行初始化，它们会有一个默认值

public class LocalVar {private int a;static int b;public static void main(String[] args) {LocalVar lv = new LocalVar();System.out.println(lv.a);System.out.println(b);}
}0
0

int 作为成员变量时或者静态变量时的默认值是 0。那不同的基本数据类型，是有不同的默认值和占用大小的

数据类型    默认值 大小
boolean false   不确定
char    '\u0000'    2 字节
byte    0   1 字节
short   0   2 字节
int 0   4 字节
long    0L  8 字节
float   0.0f    4 字节
double  0.0 8 字节

1）bit（比特）

比特作为信息技术的最基本存储单位，非常小

计算机是以二进制存储数据的，二进制的一位，就是 1 比特，也就是说，比特要么为 0 要么为 1。

2）Byte（字节）

通常来说，一个英文字符是一个字节，一个中文字符是两个字节。字节与比特的换算关系是：1 字节 = 8 比特。

在往上的单位就是 KB，并不是 1000 字节，因为计算机只认识二进制，因此是 2 的 10 次方，也就是 1024 个字节。

1）布尔

布尔（boolean）仅用于存储两个值：true 和 false，也就是真和假，通常用于条件的判断。

但在语言层面，Java 没有明确规定 boolean 类型的大小。

对于单独使用的 boolean 类型，JVM 并没有提供专用的字节码指令，而是使用 int 相关的指令 istore 来处理，那么 int 明确是 4 个字节，所以此时的 boolean 也占用 4 个字节。

对于作为数组来使用的 boolean 类型，JVM 会按照 byte 的指令来处理（bastore），那么已知 byte 类型占用 1 个字节，所以此时的 boolean 也占用 1 个字节。

5）long

long 的取值范围在 -9,223,372,036,854,775,808(-2^63) 和 9,223,372,036,854,775,807(2^63 -1)（含）之间。如果 int 存储不下，就用 long。代码示例：

long l; // 声明一个 long 类型变量l = 100000000000L; // 将值 100000000000L 赋给变量 l（注意要加上 L 后缀）long m = -20000000000L; // 声明并初始化一个 long 类型变量 m，赋值为 -20000000000L

为了和 int 作区分，long 型变量在声明的时候，末尾要带上大写的“L”。不用小写的“l”，是因为小写的“l”容易和数字“1”混淆。

6）float

float 是单精度的浮点数，32 位（4 字节）

float 不适合用于精确的数值，比如说金额

7）double

double 是双精度浮点数，占 64 位（8 字节）

double 同样不适合用于精确的数值，比如说金额。

在进行金融计算或需要精确小数计算的场景中，可以使用 BigDecimal 类来避免浮点数舍入误差。BigDecimal 可以表示一个任意大小且精度完全准确的浮点数。

在实际开发中，如果不是特别大的金额（精确到 0.01 元，也就是一分钱），一般建议乘以 100 转成整型进行处理。

8）char

char 用于表示 Unicode 字符，占 16 位（2 字节）的存储空间，取值范围为 0 到 65,535。

注意，字符字面量应该用单引号（''）包围，而不是双引号（""），因为双引号表示字符串字面量。

03、单精度和双精度

单精度（single-precision）和双精度（double-precision）是指两种不同精度的浮点数表示方法。

单精度是这样的格式，1 位符号，8 位指数，23 位小数。

单精度浮点数通常占用 32 位（4 字节）存储空间。

双精度是这样的格式，1 位符号，11 位指数，52 为小数。

双精度浮点数通常占用 64 位（8 字节）存储空间

Java中比较流行的搜索引擎是Elasticsearch

Linux下安装Elasticsearch 在Linux命令行，直接执行以下命令进行下载（下载比较慢）： wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.13.2-linux-x86_64.tar.gz

执行解压缩命令： tar -zxvf elasticsearch-7.13.2-linux-x86_64.tar.gz -C /usr/local

解决办法：

进入bin目录 cd /usr/local/elasticsearch-7.13.2/bin

修改elasticsearch配置 vim ./elasticsearch

解决es强依赖jdk问题 进入bin目录 cd /usr/local/elasticsearch-7.13.2/bin

修改elasticsearch配置 vim ./elasticsearch

############## 添加配置解决jdk版本问题 ##############
# 将jdk修改为es中自带jdk的配置目录
export JAVA_HOME=/usr/local/elasticsearch-7.13.2/jdk
export PATH=$JAVA_HOME/bin:$PATHif [ -x "$JAVA_HOME/bin/java" ]; thenJAVA="/usr/local/elasticsearch-7.13.2/jdk/bin/java"
elseJAVA=`which java`
fi

解决内存不足问题

由于 elasticsearch 默认分配 jvm空间大小为2g，修改 jvm空间，如果Linux服务器本来配置就很高，可以不用修改。

进入config文件夹开始配置，编辑jvm.options： vim /usr/local/elasticsearch-7.13.2/config/jvm.options

默认配置如下：
-Xms2g
-Xmx2g
默认的配置占用内存太多了，调小一些：
-Xms256m
-Xmx256m

创建专用用户启动ES

root用户不能直接启动Elasticsearch，所以需要创建一个专用用户，来启动ES

创建用户 useradd user-es

创建所属组： chown user-es:user-es -R /usr/local/elasticsearch-7.13.2

切换到user-es用户 su user-es

进入bin目录 cd /usr/local/elasticsearch-7.13.2/bin

启动elasticsearch ./elasticsearch

修改ES核心配置信息 执行命令修改elasticsearch.yml文件内容 vim /usr/local/elasticsearch-7.13.2/config/elasticsearch.yml

修改数据和日志目录 这里可以不用修改，如果不修改，默认放在elasticsearch根目录下

数据目录位置

path.data: /home/新用户名称/elasticsearch/data

日志目录位置

path.logs: /home/新用户名称/elasticsearch/logs

修改绑定的ip允许远程访问

#默认只允许本机访问，修改为0.0.0.0后则可以远程访问

绑定到0.0.0.0，允许任何ip来访问

network.host: 0.0.0.0

初始化节点名称

cluster.name: elasticsearch node.name: es-node0cluster.initial_master_nodes: ["es-node0"]

修改端口号（非必须）

http.port: 19200

vm.max_map_count [65530] is too low问题

elasticsearch用户拥有的内存权限太小，至少需要262144，解决办法： 在 /etc/sysctl.conf 文件最后添加如下内容，即可永久修改

切换到root用户 执行命令：su root

执行命令 vim /etc/sysctl.conf

添加如下内容 vm.max_map_count=262144

保存退出，刷新配置文件 sysctl -p

切换user-es用户，继续启动 su user-es

启动es服务 /usr/local/elasticsearch-7.13.2/bin/elasticsearch

启动成功后，可以通过http://127.0.0.1:19200/访问

可能遇到的max file descriptors [4096]问题
[1]: max file descriptors [4096] for elasticsearch process is too low, increase to at least [65535]

切换到root用户，执行命令： vi /etc/security/limits.conf

添加如下内容：

• soft nofile 65536

• hard nofile 131072

• soft nproc 2048

• hard nproc 4096

ES服务的启动与停止 前台运行，Ctrl + C 则程序终止 /usr/local/elasticsearch-7.13.2/bin/elasticsearch

后台运行 /usr/local/elasticsearch-7.13.2/bin/elasticsearch -d 出现started时启动完成

关闭ES服务 kill pid

Elasticsearch端口9300、9200，其中：
9300是tcp通讯端口，集群ES节点之间通讯使用，9200是http协议的RESTful接口

ES7.x以后的版本将安全认证功能免费开放了，并将X-pack插件集成了到了[开源]的ElasticSearch版本中。

编辑配置文件 vim /usr/local/elasticsearch-7.13.2/config/elasticsearch.yml

在 elasticsearch.yml 末尾，加入以下内容：

xpack.security.enabled: true xpack.security.transport.ssl.enabled: true

设置用户名和密码
/usr/local/elasticsearch-7.13.2/bin/elasticsearch-setup-passwords interactive
这里依次设置elastic、kibana、logstash等的访问密码，test123

elasticsearch-head是一款专门针对于elasticsearch的客户端工具，用来展示数据。

elasticsearch-head是基于JavaScript语言编写的，可以使用npm部署
npm是Nodejs下的包管理器。

安装node环境

下载elasticsearch-head安装包 下载地址： https://github.com/mobz/elasticsearch-head

行命令：
unzip elasticsearch-head-5.0.0.zip

把安装包移动到/usr/local下：
mv elasticsearch-head-5.0.0 /usr/local/

启动elasticsearch-head服务

• 启动服务
  npm run start 或者 npm run-script start

• 可以后台启动
  nohup npm run-script start &

Apache Kafka是一个开源的流式平台，用于处理实时数据流。它可以用于各种用途，包括日志聚合、事件处理、监控等。

配置Kafka服务器

在Kafka目录中，找到config[文件夹]，然后编辑**server.properties**文件。确保以下配置正确：

Kafka监听的端口号

listeners=PLAINTEXT://localhost:9092

日志存储目录

log.dirs=D:\data\logs\kafkalogs

Kafka依赖于Zookeeper来管理集群的状态。在Kafka目录中，找到config文件夹，然后编辑**zookeeper.properties**文件。确保以下配置正确：

dataDir=D:\data\logs\zookeeperclientPort=2181

以启动Zookeeper服务器：

.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

运行Nacos建议的最低硬件配置为2C 4G 60G。

linux安装nacos

首先，确认您要安装的Nacos版本。根据推荐，稳定版本为2.3.2。

预备环境准备

• 确保操作系统兼容：使用64位的Linux/Unix/Mac/Windows系统，推荐Linux/Unix/Mac。

• 安装Java环境：需要64位的JDK 1.8或以上版本。

配置Maven（仅构建时需要）：安装Maven 3.2.x或更高版本，并正确配置。

解决步骤：

1. 解决 $'\r': command not found 问题

   这个问题通常是因为脚本文件中包含 Windows 风格的换行符（\r\n）。你可以使用 dos2unix 工具来转换脚本中的换行符。

   然后，再次运行脚本：

   bash mysql57-yum-install.sh

• 安装 dos2unix（如果没有安装）：

  sudo yum install dos2unix

• 使用 dos2unix 转换脚本：

  dos2unix mysql57-yum-install.sh

[Linux安装ElasticSearch、Kibana和Elasticsearch-head ]

• [1、ES对应JDK版本]

• [2、安装JDK]

• [3、安装ElasticSearch]

• [4、安装Kibana]

• [5、安装Elasticsearch-head]

• [6、启用Elasticsearch X-Pack安全功能]

  • [6.1、X-Pack介绍]

  • [6.2、ElasticSearch 启用 xpack]

  • [6.3、Kibana 配置 xpack]

1、下载安装包

JDK8下载地址：https://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html

由于JDK下载现在要登录账户了，这里提供一下：账号：liwei@xiaostudy.com 密码：OracleTest1234 或者这个链接：http://bugmenot.com/view/oracle.com

2、解压压缩包

将压缩包上传到/usr/local下

tar -zxvf jdk-8u333-linux-x64.tar.gz -C /usr/local/

3、设置环境变量，编辑/etc/profile文件

vi /etc/profile

在profile文件最下面添加如下内容(这里根据你自己的路径和名称添加，不一定和我一样)：

JAVA_HOME=/usr/local/jdk1.8.0_333
PATH=$JAVA_HOME/bin:$PATH
CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
export JAVA_HOME
export PATH
export CLASSPATH

4、使环境变量生效

source /etc/profile

5、检查

java -version

java

配置环境变量

vim /etc/profile

export JAVA_HOME=/sh/jdk1.8.0_341
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib:$CLASSPATH
export JAVA_PATH=${JAVA_HOME}/bin:${JRE_HOME}/bin
export PATH=$PATH:${JAVA_PATH}

刷新

source /etc/profile

查看是否生效

java -version

检查 Cassandra 配置文件

Cassandra 的默认监听地址可能是 127.0.0.1 或 localhost，这意味着它只能接受本地连接。如果你想从其他机器（例如 172.16.0.14）进行连接，你需要调整 Cassandra 的配置文件：

• 打开 cassandra.yaml 配置文件，通常位于 /etc/cassandra/cassandra.yaml 或 /etc/cassandra/conf/cassandra.yaml。

  编辑配置文件，找到以下配置项并修改：

  listen_address: 0.0.0.0
  broadcast_address: <服务器的IP地址>
  broadcast_rpc_address: <服务器的IP地址>
  rpc_address: 0.0.0.0

  其中：

  修改后，保存并重新启动 Cassandra：

  sudo systemctl restart cassandra

  • listen_address 设置为 0.0.0.0，表示 Cassandra 可以在所有的网络接口上监听连接。

  • broadcast_address 和 broadcast_rpc_address 设置为服务器的 IP 地址，确保其他机器可以通过该地址访问 Cassandra。

  • rpc_address 设置为 0.0.0.0，允许接受来自任何地址的连接。

采取查看进程方式

ps -aux | grep redis

采取端口监听查看方式

netstat -lanp | grep 6379

./bin/cassandra

推荐的解决方案：使用非 root 用户运行 Cassandra

为了避免安全风险，建议为 Cassandra 创建一个专用的非 root 用户。以下是实现的方法：

第一步：创建 Cassandra 用户

1. 创建一个新的用户和组： 可以为 Cassandra 创建一个新的用户和组，以便限制其权限。

sudo useradd cassandra sudo groupadd cassandra

第二步：设置 Cassandra 目录的权限

确保 Cassandra 数据目录属于 cassandra 用户，通常 Cassandra 需要对以下目录拥有写入权限：

• /var/lib/cassandra

• /var/log/cassandra

• /etc/cassandra

最安全的做法是为 Cassandra 创建一个专用的非 root 用户（例如 cassandra），并为必要的目录设置适当的权限，然后以该用户身份运行 Cassandra。这是官方推荐的做法。除非非常必要，否则不建议以 root 用户运行 Cassandra。

sudo useradd cassandra

sudo groupadd cassandra

Cassandra需要python3才能正常运行

验证以下配置。 默认情况下，这些值将设置为指定的目录。

• data_file_directories“/ var / lib / cassandra / data”

• commitlog_directory  “/var/lib/cassandra/commitlog”

• saved_caches_directory  “/var/lib/cassandra/saved_caches”

pip3 -V pip 9.0.3 from /usr/lib/python3.6/site-packages (python 3.6)

sudo chown -R cassandra:cassandra /data1/cassandra

sudo chmod 755 -R /data1/cassandra

#配置环境变量

vi /etc/profile

#set cassandra environment

export CASSANDRA_HOME=/data/server/cassandra

export PATH=CASSANDRA_HOME/bin

:wq! #保存退出

source /etc/profile #使配置立即生效

#设置目录权限

sudo chown -R cassandra:cassandra /data/server/cassandra

sudo chmod 755 -R /data/server/cassandra

#切换到cassandra用户

su - cassandra

编辑配置文件

cp /data/server/cassandra/conf/cassandra.yaml /data/server/cassandra/conf/cassandra.yaml.bak

vi /data/server/cassandra/conf/cassandra.yaml

authenticator: PasswordAuthenticator #启用用户名和密码验证

authorizer: CassandraAuthorizer

listen_address: 192.168.21.100 #用于监听客户端请求的地址

rpc_address: 0.0.0.0 #用于接受远程过程调用（RPC）请求的地址

broadcast_rpc_address: 192.168.21.100 #定节点之间进行远程过程调用（RPC）通信时使用的广播地址

• seeds: "192.168.21.100,127.0.0.1" #种子节点的地址

native_transport_max_concurrent_connections: -1   #最大连接数，默认是-1没有限制

:wq! #保存退出

共享充电宝管理系统

1.注册用户

账号注册登录，查看充电宝推荐，充电宝租赁，充电宝归还，充电宝保修，发布评论信息，查看充电账单，评论信息管理，收藏信息管理。

2.商家用户

个人中心管理，充电宝管理，充电宝租赁，充电账单管理，故障保修管理，充电评价管理，查看通知公告。

3.管理员

用户信息管理，商家信息管理，品牌分类管理，充电宝管理，租赁信息管理，充电账单管理，投诉建议管理，故障保修管理，充电评价管理，查看通知公告。

ps -ef | grep cassandra

kill -9 pid

• /sh/apache-cassandra-3.11.4/bin/cassandra：指定 Cassandra 的启动脚本路径。

• -R：确保 Cassandra 在启动时移除控制台日志并运行为守护进程。

• >cassandra.log：将标准输出（stdout）重定向到 cassandra.log。

• 2>&1：将错误输出（stderr）也重定向到 cassandra.log。

• &：让该命令在后台运行，释放终端。

tail -f cassandra.log

解决方案

1. 创建一个普通用户并运行 Cassandra

建议创建一个专用用户运行 Cassandra。操作步骤如下：

1. 创建用户 cassandra：

   useradd cassandra

2. 赋予 Cassandra 安装目录的权限给 cassandra 用户：

   chown -R cassandra:cassandra /sh/apache-cassandra-3.11.4

• 切换到 cassandra 用户：

  su - cassandra

• 启动 Cassandra：

  /sh/apache-cassandra-3.11.4/bin/cassandra

0.0.0.0

Cassandra 不允许 listen_address 设置为 0.0.0.0，因为它表示绑定所有网络接口，可能会导致一些安全问题或者网络配置问题。

修改 jvm.options 配置

如果你依然遇到 IPv6 相关的问题，可以尝试禁用 Cassandra 使用 IPv6。打开 jvm.options 文件（位于 Cassandra 配置目录下），并添加以下配置来禁用 IPv6：

-sh/apache-cassandra-3.11.4/conf/jvm.options

在该文件中，确保添加如下内容（如果没有的话）：

-Djava.net.preferIPv4Stack=true

此选项强制 JVM 优先使用 IPv4，而不是尝试使用 IPv6。

lsof -i :7199

public final class ZKCacheListener implements TreeCacheListener {// 监听的路径private final String watchPath;// 监听器private final ServiceListener listener;// 构造函数，传入监听的路径和监听器public ZKCacheListener(String watchPath, ServiceListener listener) {this.watchPath = watchPath;this.listener = listener;}// 监听事件@Overridepublic void childEvent(CuratorFramework curator, TreeCacheEvent event) throws Exception {// 获取事件的数据ChildData data = event.getData();if (data == null) {return;}// 获取数据路径String dataPath = data.getPath();if (Strings.isNullOrEmpty(dataPath)) {return;}// 判断数据路径是否以监听路径开头if (dataPath.startsWith(watchPath)) {// 根据事件类型调用监听器的相应方法switch (event.getType()) {// 当节点被添加时case NODE_ADDED:// 调用监听器的onServiceAdded方法，传入节点路径和解析后的CommonServiceNode对象listener.onServiceAdded(dataPath, Jsons.fromJson(data.getData(), CommonServiceNode.class));break;
// 当节点被移除时case NODE_REMOVED:// 调用监听器的onServiceRemoved方法，传入节点路径和解析后的CommonServiceNode对象listener.onServiceRemoved(dataPath, Jsons.fromJson(data.getData(), CommonServiceNode.class));break;
// 当节点被更新时case NODE_UPDATED:// 调用监听器的onServiceUpdated方法，传入节点路径和解析后的CommonServiceNode对象listener.onServiceUpdated(dataPath, Jsons.fromJson(data.getData(), CommonServiceNode.class));break;default:break;}// 打印日志Logs.CONN.debug("ZK node data change={}, nodePath={}, watchPath={}, ns={}");}}
}

// 定义一个抽象类NettyTCPClient，继承自BaseService，实现Client接口
public abstract class NettyTCPClient extends BaseService implements Client {// 定义一个EventLoopGroup类型的workerGroup变量private EventLoopGroup workerGroup;// 定义一个Bootstrap类型的bootstrap变量protected Bootstrap bootstrap;// 创建客户端private void createClient(Listener listener, EventLoopGroup workerGroup, ChannelFactory<? extends Channel> channelFactory) {// 设置工作线程组this.workerGroup = workerGroup;// 创建Bootstrap对象this.bootstrap = new Bootstrap();// 设置工作线程组bootstrap.group(workerGroup)//// 设置SO_REUSEADDR选项，允许重复使用本地地址.option(ChannelOption.SO_REUSEADDR, true)//// 设置内存分配器.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)//// 设置通道工厂.channelFactory(channelFactory);// 设置通道初始化器bootstrap.handler(new ChannelInitializer<Channel>() { // (4)@Overridepublic void initChannel(Channel ch) throws Exception {// 初始化通道initPipeline(ch.pipeline());}});// 初始化选项initOptions(bootstrap);// 回调成功listener.onSuccess();}public ChannelFuture connect(String host, int port) {// 连接到指定的主机和端口return bootstrap.connect(new InetSocketAddress(host, port));}public ChannelFuture connect(String host, int port, Listener listener) {// 连接到指定的主机和端口，并添加监听器return bootstrap.connect(new InetSocketAddress(host, port)).addListener(f -> {if (f.isSuccess()) {// 连接成功if (listener != null) {// 调用监听器的onSuccess方法listener.onSuccess(port);}Logs.CONN.debug("start netty client success, host={}, port={}", host, port);} else {// 连接失败if (listener != null) {// 调用监听器的onFailure方法listener.onFailure(f.cause());}Logs.CONN.error("start netty client failure, host={}, port={}", host, port, f.cause());}});}// 创建Nio客户端private void createNioClient(Listener listener) {// 创建NioEventLoopGroup，用于处理I/O操作NioEventLoopGroup workerGroup = new NioEventLoopGroup(// 获取工作线程数getWorkThreadNum(),// 创建默认的线程工厂new DefaultThreadFactory(ThreadNames.T_TCP_CLIENT),// 获取选择器提供者getSelectorProvider());// 设置I/O操作的比率workerGroup.setIoRatio(getIoRate());// 创建客户端createClient(listener, workerGroup, getChannelFactory());}// 创建Epoll客户端private void createEpollClient(Listener listener) {// 创建Epoll事件循环组，设置工作线程数和线程工厂EpollEventLoopGroup workerGroup = new EpollEventLoopGroup(getWorkThreadNum(), new DefaultThreadFactory(ThreadNames.T_TCP_CLIENT));// 设置IO比例workerGroup.setIoRatio(getIoRate());// 创建客户端createClient(listener, workerGroup, EpollSocketChannel::new);}// 初始化管道protected void initPipeline(ChannelPipeline pipeline) {// 添加解码器pipeline.addLast("decoder", getDecoder());// 添加编码器pipeline.addLast("encoder", getEncoder());// 添加处理器pipeline.addLast("handler", getChannelHandler());}// 获取解码器protected ChannelHandler getDecoder() {return new PacketDecoder();}// 获取编码器protected ChannelHandler getEncoder() {return PacketEncoder.INSTANCE;}// 获取IO速率protected int getIoRate() {return 50;}// 获取ChannelFactorypublic ChannelFactory<? extends Channel> getChannelFactory() {// 返回NioSocketChannel的构造函数return NioSocketChannel::new;}// 获取SelectorProviderpublic SelectorProvider getSelectorProvider() {// 返回系统默认的SelectorProviderreturn SelectorProvider.provider();}@Overridepublic String toString() {// 返回NettyClient的字符串表示return "NettyClient{" +", name=" + this.getClass().getSimpleName() +'}';}// 获取工作线程数protected int getWorkThreadNum() {return 1;}// 获取ChannelHandlerpublic abstract ChannelHandler getChannelHandler();// 重写doStart方法@Overrideprotected void doStart(Listener listener) throws Throwable {// 如果使用NettyEpollif (useNettyEpoll()) {// 创建EpollClientcreateEpollClient(listener);} else {// 否则创建NioClientcreateNioClient(listener);}}// 判断是否使用NettyEpollprivate boolean useNettyEpoll() {// 如果配置文件中设置了使用NettyEpollif (CC.mp.core.useNettyEpoll()) {try {// 调用Native类中的offsetofEpollData方法Native.offsetofEpollData();// 打印日志，表示客户端已经使用了NettyEpollLogs.CONN.debug("client has used netty epoll");// 返回truereturn true;} catch (UnsatisfiedLinkError error) {// 如果加载NettyEpoll失败，打印错误日志，并切换到nio模式Logs.CONN.error("can not load netty epoll, switch nio model.");}}// 返回falsereturn false;}@Overrideprotected void doStop(Listener listener) throws Throwable {// 关闭workerGroupif (workerGroup != null) {workerGroup.shutdownGracefully();}// 打印日志Logs.CONN.debug("netty client [{}] stopped.", this.getClass().getSimpleName());// 执行listener的onSuccess方法listener.onSuccess();}protected void initOptions(Bootstrap b) {// 设置连接超时时间为4000毫秒b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 4000);// 设置TCP_NODELAY为true，禁用Nagle算法b.option(ChannelOption.TCP_NODELAY, true);}}

加群联系作者vx：xiaoda0423

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