跳到主要内容
Jason Lee

Jason Lee

The owner of this blog site

I'm a programmer who loves to learn new technologies and build cool stuff. I'm currently working as a software engineer at a small company in Shanghai. Hope to find some connections with you!

View All Authors

pgsql常用且容易忘记的一些用法

· 阅读需 3 分钟
Jason Lee
Jason Lee
The owner of this blog site

总结一些工作中非常有用的 pgsql 用法。

日期类​

加入 time 列是 timestamp 类型的，那么

需要获得昨日的数据：select ... where time between current_date-1 and current_date;

过去 24 小时：select ... where time > current_timestamp - interval '1 day';

今日的数据：select ... where time > current_date；

过去七天的数据并按天列出：select ... where time > current_date-7 group by extract(day from parking_time)；

需要注意的是，在 where 子句中慎用强转符号::，因为这会让该列索引失效，从而搜索速度变得很慢。应该变通成上面这些形式，即 sql 会在 timestamp 类型和 date 类型比较时自动将 date 类型转为 timestamp 类型。

在 PostgreSQL 中，可以使用不同的函数和模式来将 timestamp 数据类型转换为特定的时间格式。以下是几种常见的方法：

  1. 使用 TO_CHAR 函数：TO_CHAR(timestamp, 'format') 函数将 timestamp 转换为指定的时间格式。例如，要将 timestamp 转换为年-月-日 小时:分钟:秒 的格式，可以使用以下代码：

    SELECT TO_CHAR(timestamp_column, 'YYYY-MM-DD HH24:MI:SS') FROM your_table;
    ```

  2. 使用 EXTRACT 函数：EXTRACT(field FROM timestamp) 函数允许提取 timestamp 中的特定时间部分（如年、月、日、小时等）。然后，可以将提取的时间部分按照需要的格式进行拼接。例如，以下代码将 timestamp 转换为年-月-日 格式：

    SELECT EXTRACT(YEAR FROM timestamp_column) || '-' || EXTRACT(MONTH FROM timestamp_column) || '-' || EXTRACT(DAY FROM timestamp_column) FROM your_table;
    ```

  3. 使用 TO_TIMESTAMP 函数和 TO_CHAR 函数的组合：如果要在转换过程中进行一些计算或调整，可以使用 TO_TIMESTAMP 函数将 timestamp 转换为特定格式的时间戳，然后再使用 TO_CHAR 函数将其格式化为字符串。例如，以下代码将 timestamp 转换为带有 AM/PM 标记的小时:分钟:秒 格式：

    SELECT TO_CHAR(TO_TIMESTAMP(timestamp_column), 'HH12:MI:SS AM') FROM your_table;
    ```

这些方法只是其中的几种，具体的选择取决于所需的时间格式和转换的要求。可以根据具体情况选择合适的方法。

having 和 where 的区别​

这里有另一种理解方法，WHERE 在数据分组前进行过滤，HAVING 在数据分组后进行过滤。这是一个重要的区别，WHERE 排除的行不包括在分组中。这可能会改变计算值，从而影响 HAVING 子句中基于这些值过滤掉的分组。

python机器学习-模型评估与参数调优

· 阅读需 3 分钟
Jason Lee
Jason Lee
The owner of this blog site

性能指标（二分类问题）​

混淆矩阵​

首先，为什么在有准确率（accuracy）的情况下，还要引入别的指标呢？因为受困于数据收集的客观限制，容易出现不平衡问题，比如正类的数量远远大于负类。这样，即使模型将所有样本预测为正类，损失函数也很低，达到了欺骗的效果。因此，我们需要将正类和负类分别的预测结果列出来，如混淆矩阵。

16981566899581698156689045.png

准确率和召回率​

基于混淆矩阵，得到了准确率（precision）和召回率（recall）：

precision = TP/(TP+FP)
recall = TP/(TP+FN)

ROC 曲线​

我们通过真正率（TPR）和假真率（FPR）来衡量分类器的性能。

通过 ROC 空间，我们明白了一条 ROC 曲线其实代表了无数个分类器。那么我们为什么常常用一条 ROC 曲线来描述一个分类器呢？仔细观察 ROC 曲线，发现其都是上升的曲线（斜率大于 0），且都通过点（0,0）和点（1,1）。其实，这些点代表着一个分类器在不同阈值下的分类效果，具体的，曲线从左往右可以认为是阈值从 0 到 1 的变化过程。当分类器阈值为 0，代表不加以识别全部判断为 0，此时 TP=FP=0，TPR=TP/P=0，FPR=FP/N=0；当分类器阈值为 1，代表不加以识别全部判断为 1，此时 FN=TN=0，P=TP+FN=TP, TPR=TP/P=1，N=FP+TN=FP, FPR=FP/N=1。所以，ROC 曲线描述的其实是分类器性能随着分类器阈值的变化而变化的过程。对于 ROC 曲线，一个重要的特征是它的面积，面积为 0.5 为随机分类，识别能力为 0，面积越接近于 1 识别能力越强，面积等于 1 为完全识别。

python机器学习-数据降维

· 阅读需 6 分钟
Jason Lee
Jason Lee
The owner of this blog site

PCA（主成分分析）​

example:

import numpy as np
from sklearn.decomposition import PCA

# 创建一个示例数据集
data = np.array([
    [1.5, 2.0, 3.0, 4.5],
    [2.2, 2.8, 3.9, 4.2],
    [1.9, 2.5, 3.2, 4.7],
    [2.7, 3.0, 3.6, 4.0],
    [2.3, 2.9, 3.7, 4.3]
])

# 初始化 PCA 模型，指定主成分的数量
pca = PCA(n_components=2)

# 对数据进行拟合
pca.fit(data)

# 获取主成分
principal_components = pca.components_

# 获取主成分的方差贡献比例
explained_variance_ratio = pca.explained_variance_ratio_

# 对数据进行 PCA 转换
transformed_data = pca.transform(data)

print("原始数据：\n", data)
print("主成分：\n", principal_components)
print("主成分的方差贡献比例：", explained_variance_ratio)
print("PCA 转换后的数据：\n", transformed_data)

输出：

原始数据：
 [[1.5 2.  3.  4.5]
 [2.2 2.8 3.9 4.2]
 [1.9 2.5 3.2 4.7]
 [2.7 3.  3.6 4. ]
 [2.3 2.9 3.7 4.3]]
主成分：
 [[-0.61305201 -0.55633419 -0.46392272  0.31533349]
 [-0.53989171 -0.03347232  0.83292811  0.11673604]]
主成分的方差贡献比例： [0.87919714 0.0703791 ]
PCA 转换后的数据：
 [[ 1.00928239 -0.02497259]
 [-0.37705186  0.28493986]
 [ 0.45617665 -0.0677326 ]
 [-0.71873459 -0.2649261 ]
 [-0.36967259  0.07269143]]

在深入之前，有必要解释一下 PCA 为什么在高维数据中找最大方差的方向而不是别的指标（比如最大值），来降维的？找到最大方差的方向意味着在这个方向上数据的变化最为显著，即信息越丰富。以我之前接触到的人脸图像为例，不同位置的像素有差异，才能让分类器有能力识别。

为了将我视频学习到的和这个例子对应起来，将输出中的几个部分进行解释：

首先，这个例子中的原始数据是有五个样本，每个样本有四个维度的特征（一开始我还真没搞明白这个）。然后，pca 算法找出四个主成分（找的过程是通过计算协方差矩阵和特征向量等步骤得来的，视频举的例子是几何平面便于理解，而书上的公式是一种通用的方法，适合任意数量纬度特征，二者本质相同），也就是输出中的主成分。主成分本质上是一个方向向量，且模为 1，所以本例中的主成分有两组，每组四个值，对应四种特征，哪个特征的绝对值最大，说明哪个特征在这个主成分中占主导。方差贡献比例就是，依次计算所有数据在某一主成分上的投影点到距离之和，然后全部求和，最后计算一个主成分所占的比例，比例越高，说明方差越大，数据在这个主成分上差异越明显。最后的转换后数据，则是将所有数据在某一主成分上的投影点到距离，作为以各主成分为新坐标轴的空间内的坐标。

LDA（线性判别分析）​

不同于 PCA，LDA 有类别标签参与训练，所以是有监督的。LDA 训练过程中的优化目标也变成了最大化不同类之间的类间差距，而不是 PCA 的找样本内方差最大的方向，可以说是完全不同的方法，但是都能实现降维。

LDA 在一条线上投影数据点，并且让投影点的不同类之间的均值差距最大，而同类的方差最小，这样就让不同类之间的类间差距最大化。

可以说，LDA 的适用范围更窄些，它要求数据有明确的标签，且只适用分类问题；而 PCA 则适用范围更广，因为它只要求提供特征，能适用于不限于分类的问题。

example:

# 导入必要的库
from sklearn import datasets
from sklearn.discriminant_analysis import LinearDiscriminantAnalysis
import matplotlib.pyplot as plt

# 加载示例数据集（这里以Iris数据集为例）
iris = datasets.load_iris()
X = iris.data
y = iris.target

# 创建LDA模型
lda = LinearDiscriminantAnalysis(n_components=2)  # 我们希望将数据降至二维

# 拟合模型并进行降维
X_lda = lda.fit_transform(X, y)

# 绘制降维后的数据
plt.figure(figsize=(8, 6))
colors = ['navy', 'turquoise', 'darkorange']
lw = 2

for color, i, target_name in zip(colors, [0, 1, 2], iris.target_names):
    plt.scatter(X_lda[y == i, 0], X_lda[y == i, 1], alpha=0.8, color=color,
                label=target_name)
plt.legend(loc='best', shadow=False, scatterpoints=1)
plt.title('LDA of IRIS dataset')
plt.show()

16978904104791697890409581.png

工作中的一些工具使用技巧和方法论

· 阅读需 2 分钟
Jason Lee
Jason Lee
The owner of this blog site

我的工作是后端开发，这几天的内容就是做一套增删改查接口，暂时还没涉及复杂的逻辑。之前在外包里面搞前端，工具就两个：vscode 和浏览器。现在包括但不限于：navicat（操纵数据库）、vscode（写代码）、winscp（把本地写好的代码传到测试服务器）、postman（测试接口）、xshell（查看服务报错日志来 debug）、金山文档（写接口文档）。项目是没有办法在本地运行的，都要上传到测试服务器，通过请求接口后才能知道有无 bug。

之前的一个痛点是，每次写完代码，要在 scp 里面改目录，拖进去；如果每次修改 model、service、controller，就要拖动三次，还容易看错目录，搞得很紧张。后来，自己想的办法是，把 winscp 的文本编辑器由默认的内置改为 vscode，这样，直接从远程服务器上下载一个文件到本地，用 vscode 打开，编辑完保存，改动就同步到服务器上了。

《python 机器学习》数据预处理

· 阅读需 3 分钟
Jason Lee
Jason Lee
The owner of this blog site

缺失数据值的处理​

在 scikit-learn 中处理缺失数据的方法：​

SimpleImputer: SimpleImputer 是 scikit-learn 中用于填充缺失数据的类。你可以使用不同的策略来填充缺失值，包括均值、中位数、众数等。例如：

from sklearn.impute import SimpleImputer

imputer = SimpleImputer(strategy='mean')  # 使用均值填充缺失值
X_imputed = imputer.fit_transform(X)

KNNImputer: KNNImputer 使用 K-最近邻算法来填充缺失值，根据相邻样本的值进行估计。

from sklearn.impute import KNNImputer

imputer = KNNImputer(n_neighbors=5)  # 使用5个最近邻的值来估计缺失值
X_imputed = imputer.fit_transform(X)

在 pandas 中处理缺失数据的方法：​

isna() 和 dropna(): isna() 方法用于检测缺失值，dropna() 方法用于删除包含缺失值的行或列。

import pandas as pd

df = pd.DataFrame({'A': [1, 2, None, 4], 'B': [None, 6, 7, 8]})
df.isna()  # 检测缺失值
df.dropna()  # 删除包含缺失值的行
df.dropna(axis=1)  # 删除包含缺失值的列

fillna(): fillna() 方法用于填充缺失值，你可以指定要使用的值或方法。

df.fillna(value=0)  # 用0填充缺失值
df.fillna(method='ffill')  # 使用前一个非缺失值进行前向填充
df.fillna(method='bfill')  # 使用后一个非缺失值进行后向填充

处理类别数据​

这里对类别数据做一个定义：一般以字符串形式出现，表示类别。类别数据分为两种，一种是内在有数值大小关系的，比如衣服尺寸：S，M，L；另一种是标称特征，不具备排序特征。既然是字符串，我们都要处理为数字方便训练。

sklearn​

这个函数是处理标签列的，本质上是通过枚举，将所有独立的标签从 0 开始编号。之所以可以这样处理，因为我们不关心类标被赋予哪个整数值。

from sklearn.preprocessing import LabelEncoder
class_le = LabelEncoder()
y = class_le.fir_transform(df['label'].values)

// 如果要将整数类标还原到字符串，可以用：
class_le.inverse_transform(y)

如果是属于特征的类别数据，特别是无序数据，用枚举则是不合适的。这是因为，这会使得模型假定不同类别之间有顺序关系。所以，独热编码更合适。

from sklearn.preprocessing import OneHotEncoder
ohe = OneHotEncoder()
ohe.fit_transform(df['color'].values)

pandas​

pandas 有 get_dummies 可以用于实现独热编码

pd.get_dummies(df) // 只对字符串列进行转化

《python机器学习》分类算法

· 阅读需 3 分钟
Jason Lee
Jason Lee
The owner of this blog site

前言​

我打算开一个全新的系列，记录我在学习《python 机器学习》这本书过程中的要点和感想。

决策边界图​

分类任务中的一种图，可以看出模型在整个网格内的预测结果。

16962158450461696215844398.png

代码如下：

import numpy as np
import matplotlib.pyplot as plt
from sklearn import datasets
from sklearn.svm import SVC

# 步骤2：加载鸢尾花数据集
iris = datasets.load_iris()
X = iris.data[:, :2]  # 使用前两个特征以便进行可视化
y = iris.target

# 步骤3：训练一个SVM分类器
clf = SVC(kernel='linear')
clf.fit(X, y)

# 步骤4：创建一个网格来覆盖特征空间
x_min, x_max = X[:, 0].min() - 1, X[:, 0].max() + 1
y_min, y_max = X[:, 1].min() - 1, X[:, 1].max() + 1
xx, yy = np.meshgrid(np.arange(x_min, x_max, 0.01), np.arange(y_min, y_max, 0.01))

# 步骤5：对网格上的每个点使用分类器进行预测，获取决策边界
Z = clf.predict(np.c_[xx.ravel(), yy.ravel()])
Z = Z.reshape(xx.shape)

# 步骤6：可视化数据点、决策边界和决策区域
plt.contourf(xx, yy, Z, cmap=plt.cm.Spectral, alpha=0.8)
plt.scatter(X[:, 0], X[:, 1], c=y, cmap=plt.cm.Spectral)
plt.xlabel('Sepal Length')
plt.ylabel('Sepal Width')
plt.title('Decision Boundary Visualization')
plt.show()

主要的函数有 meshgrid，功能是生成网格坐标数据。还有 contourf，绘制决策边界图。

逻辑回归​

线性计算方式，激活函数采用 sigmoid，从而得出属于正类别的概率。

决策树/随机森林​

决策树就是一种二叉树，通过计算一种特征能否尽可能的将类别区分来决定选择哪一种特征作为节点。随机森林则是决策树的集合，通过随机抽样得到训练样本来引入随机性，并且最后由多数投票来确定预测结果。

基于eggjs, sequelize和pgsql的小需求

· 阅读需 5 分钟
Jason Lee
Jason Lee
The owner of this blog site

前言​

这是我在部门遇到的第一个小需求。这一次，我写的是后端代码。刚开始，我还陷入在一种思维定式里，似乎一定要在现有项目的工程里添加代码。如果这样做，就需要在本地或者测试服务器启动项目，然而不熟悉服务器环境和配置的我，容易卡在一个环节上无法继续。我后面想明白了，完全可以将其视作一个独立的小项目。

pgsql 启动入口​

不同于 mysql 直接通过命令行工具启动，pgsql 的启动方式有两种。一种是自带的 pgadmin4 这款数据库可视化软件，只要第一次连接上，后续打开都会自动连接。另一种是 sql shell, 也是随 pgsql 一起安装的。打开后，前面几个参数都是默认，最后输入口令（密码）。

16959671454581695967145365.png

eggjs+sequelize 的开发流程​

初始化 egg 项目和安装 npm 包​

$ npm init egg --type=simple
$ npm i

# sequelize 和 nodejs的pgsql驱动
npm install --save egg-sequelize pg pg-hstore

在 config/plugin.js 中引入 egg-sequelize 插件​

'use strict';

exports.sequelize = {
  enable: true,
  package: 'egg-sequelize',
};

在 config/config.default.js 中编写 sequelize 配置​

/* eslint valid-jsdoc: "off" */

'use strict';

module.exports = appInfo => {
  /**
   * built-in config
   * @type {Egg.EggAppConfig}
   **/
  const config = exports = {};

  // use for cookie sign key, should change to your own and keep security
  config.keys = appInfo.name + '_1696034448361_9424';

  // add your middleware config here
  config.middleware = [];

  // mysql

  // config.sequelize = {
  //   dialect: 'mysql', // support: mysql, mariadb, postgres, mssql
  //   database: 'egg-sequelize-example-dev',
  //   host: '127.0.0.1',
  //   port: 3306,
  //   username: 'root',
  //   password: 'root',
  // };

  // pgsql

  config.sequelize = {
    database: 'egg-sequelize-example-dev',
    username: 'postgres',
    password: 'root',
    host: 'localhost', // 或者你的数据库主机地址
    port: 5432, // 或者你的数据库端口号
    dialect: 'postgres', // 指定数据库类型为PostgreSQL
  };

  // add your user config here
  const userConfig = {
    // myAppName: 'egg',
  };

  return {
    ...config,
    ...userConfig,
  };
};

创建数据库和数据表（在 pgadmin4 等工具中写 sql 语句）​

CREATE DATABASE IF NOT EXISTS 'egg-sequelize-doc-default';
create table [name];

编写 model​

注意字段要和数据表中定义的完全一致。这时候，model 可以通过 ctx.model 访问到，在编写 controller 或 service 层时可通过 ide 检查能否找到 model，若不能，检查 typings 文件夹下相应的 index.d.ts 文件，看看名称能否对应上。

# app/model/station.js

'use strict';

module.exports = (app) => {
	const { STRING, INTEGER, TEXT, DATE } = app.Sequelize;
	const { DataTypes } = require('sequelize');

    const Station = app.model.define('Station', {
      id: { type: INTEGER, primaryKey: true, autoIncrement: true },
      name: { type: STRING(30), allowNull: false },
      associated_yard: { type: STRING(30), allowNull: false },
      address: { type: STRING(50) },
      longitude: { type: DataTypes.NUMERIC(5), allowNull: false},
      latitude: { type: DataTypes.NUMERIC(5), allowNull: false },
      province: {type: STRING(32), allowNull: false},
      city: {type: STRING(32), allowNull: false},
      county: { type: STRING(32), allowNull: false },
      tag: { type: DataTypes.ARRAY(DataTypes.TEXT) },
      property_unit: { type: STRING(30) },
      property_contact: { type: STRING(30) },
      property_phone: { type: STRING(11) },
      status: { type: STRING(10), allowNull: false },
      map_induction: { type: DataTypes.BOOLEAN, allowNull: false },
      area: { type: STRING(30) },
      accounting_policy: { type: STRING(30), allowNull: false },
      operating_description: { type: TEXT, allowNull: false },
      comment: { type: TEXT },
      created_at: DATE,
      updated_at: DATE,
    });

    return Station;
};

上述代码中需要注意 sequelize 的表名推导问题。在const Station = app.model.define('Station', {...})中，第一个 Station 是模型名称，是 eggjs 中调用 app.model.[name]用到的名称。然后 define 函数中的'Station'和数据库中的表明存在对应关系，具体的对应规则是：

  1. 默认的是单数转复数：默认情况下，Sequelize 会将模型的名称转换为复数形式，并将其用作数据库表的名称。比如，station 转成 stations。所以创建表时，应该把表命名为复数形式。
  2. 自定义表名：可以在定义模型时明确指定要映射到的表的名称。
const User = sequelize.define('User', {
  // 模型属性定义
}, {
  tableName: 'custom_users_table' // 自定义表名
});

编写 controller 和 service 层，在 router.js 中添加路由。​

sklearn crash course

· 阅读需 4 分钟
Jason Lee
Jason Lee
The owner of this blog site

前言​

油管视频教程。想到哪儿写到哪儿。

如何让 jupyter notebook 切换到 venv 环境​

首先在虚拟环境下的 cmd 中运行：

pip install ipykernel

然后，在在虚环境中将当前的虚拟环境添加到 Jupyter Notebook 的 kernel 中：

python -m ipykernel install --name 虚环境名称 --display-name 虚环境名称 --user

框架​

16955688583221695568857524.png

上图是用 sklearn 完成机器学习任务的一般框架。首先数据类型分为特征（X）和标签（y）。pipeline（管道、流程）包含数据归一化（preprocessing 的一种）和模型（model）。pipeline 有两个重要 api：fit 和 predict，前者训练，后者测试。

如果要确定超参数的值，且训练数据有限的情况下，需要用交叉验证（Cross validation, CV）。用法是：

mod = GridSearchCV(estimator=pipe, param_grid={
    'model__n_neighbors': [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    },
    cv=3)

其中，estimator 是 pipeline 或者 model，param_grid 是需要确定的超参数，是一个字典，键是'model__[name]'的格式，cv 表示要将数据分为几折来进行交叉验证。

完整代码如下：

from sklearn.datasets import load_boston
from sklearn.neighbors import KNeighborsRegressor
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.model_selection import GridSearchCV
import pandas as pd

mod = GridSearchCV(estimator=pipe, param_grid={
    'model__n_neighbors': [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    },
    cv=3)
mod.fit(X, y);

## 查看交叉验证的结果
pd.DataFrame(mod.cv_results_)

预处理​

视频以一个存在离群点的场景为例，说明预处理措施的作用。离群点会导致模型预测的偏差。

16956472633051695647262551.png

最容易想到的预处理方式是正态分布化，但是离群点问题仍然存在。因此，采用一种均匀分布方法：QuantileTransformer。

16956474313051695647430838.png

可以看出，离群点不明显了。

另外，增加特征的维数（PolynomialFeatures）也在预处理范畴内。增加特征维数可以更好地捕获数据中的特征关系。

最后介绍了 One Hot Encoding，一种经典的将文本数据转为数值特征（标签）的预处理措施。

总结一下，sklearn 中，调用预处理措施的格式是：Transformer().fit_transform(data).

指标​

precision_score：分母是模型预测为正类的个数，分子是模型预测正类正确的个数。

recall_score: 分母是样本中所有正类的个数，分子是分母中模型预测正类正确的个数。

GridSearchCV （或者 model 和 pipeline）可以使用 scoring 来自定义目标函数。而自定义的损失函数需要 make_scorer 生成。

from sklearn.model_selection import GridSearchCV
from sklearn.metrics import precision_score, recall_score, make_scorer

def min_recall_precision(est, X, y_true, sample_weight=None):
    y_pred = est.predict(X)
    recall = recall_score(y_true, y_pred)
    precision = precision_score(y_true, y_pred)
    return min(recall, precision)

grid = GridSearchCV(
    estimator=LogisticRegression(max_iter=1000),
    param_grid={'class_weight': [{0: 1, 1: v} for v in np.linspace(1, 20, 30)]},
    scoring={'precision': make_scorer(precision_score),
             'recall': make_scorer(recall_score),
             'min_both': min_recall_precision},
    refit='min_both',
    return_train_score=True,
    cv=10,
    n_jobs=-1
)

kafka初体验

· 阅读需 4 分钟
Jason Lee
Jason Lee
The owner of this blog site

启动 zookeeper 和 kafka​

安装参考教程。本机版本是 2.12-3.5.1。

在启动 kafka 之前先启动 zookeeper：

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

然后

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

创建 topic​

.\bin\windows\kafka-topics.bat --create --bootstrap-server localhost:9092 --topic my-topic --partitions 3 --replication-factor 1

解释一下参数的含义：

  • kafka-topics.bat：用于创建、列出或删除 Kafka 主题的命令。
  • create：指示要创建主题。
  • bootstrap-server localhost:9092：指定 Kafka 服务器的地址和端口。确保这个地址和端口与你的 Kafka 服务器配置一致。
  • topic my-topic：指定要创建的主题名称，你可以将 my-topic 替换为你想要的主题名称。
  • partitions 3：指定主题的分区数。你可以根据需要调整分区数。
  • replication-factor 1：指定主题的复制因子。在本例中，复制因子为 1，这意味着每个分区只有一个副本。

列出 kafka 所有 topics，查看是否创建成功：

.\bin\windows\kafka-topics.bat --list --bootstrap-server localhost:9092

生产者发送消息​

.\bin\windows\kafka-console-producer.bat --bootstrap-server localhost:9092 --topic my-topic
  • kafka-console-producer.bat：这是 Kafka 的命令行生产者工具。
  • bootstrap-server localhost:9092：指定 Kafka 服务器的地址和端口。确保这个地址和端口与你的 Kafka 服务器配置一致。
  • topic your-topic-name：指定你要发送消息的主题名称，请将 your-topic-name 替换为你的实际主题名称。

运行上述命令后，命令行将进入消息输入模式。你可以在命令行中输入消息，然后按 Enter 键发送它们。当你完成发送消息后，你可以通过键入 Ctrl+C 来退出 Kafka 生产者。

消费者接受消息​

.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic my-topic --from-beginning
  • kafka-console-consumer.bat：这是 Kafka 的命令行消费者工具。
  • bootstrap-server localhost:9092：指定 Kafka 服务器的地址和端口。确保这个地址和端口与你的 Kafka 服务器配置一致。
  • topic your-topic-name：指定你要消费消息的主题名称，请将 your-topic-name 替换为你的实际主题名称。
  • from-beginning：这个选项表示你要从主题的起始位置开始消费消息。如果你希望从当前的消息位置开始消费，可以省略这个选项。

运行上述命令后，Kafka 消费者将开始消费指定主题中的消息。如果你之前已经在生产者中发送了消息到该主题，那么这些消息将会显示在命令行中。

你可以在命令行中看到消费的消息，直到你手动停止消费者（通过按 Ctrl+C）或者关闭命令行窗口。