实验内容实验环境语言：PythonIDE：Pycharm实验思路步骤一这一步其实就是筛选节点，得到图的点集，并把每个点都连边形成一个无向完全图。 每组数据行为基因，列为细胞。每个基因的表达方差其实就是这个基因的这一行的表达值的方差，然后从所有基因中筛选出方差最大的前 1000 个来作为图的节点集合，节点标识符定义为 ”基因名/CancerCell“ 或 ”基因名/TCell“。每种癌症有两组数据，所以每种癌症会有两个无向完全图。步骤二这一步其实就是把上一步得到的六个无向完全图之间连边。根据提供的 LigandReceptor_Human.txt 提供的受体配体来连边。最终每种癌症的两个无向完全图会连在一起，形成一个新的无向完全图。步骤三这一步是为步骤一连的边加上边权。根据要求计算边权即可。spearman 相关系数可以直接用 pandas 库里的方法，效率要比 spicy 库的 spearmanr 函数快很多。步骤四这一步是为步骤二连的边加上边权。根据要求计算边权即可。步骤五把每种癌症得到的无向图求最小生成树即可。由于是稠密图，所以考虑使用 Prim 算法。直接用库里的 get_mininum_spanning_tree 函数即可，效率要比自己手写的 MST 要高。步骤六把步骤五中得到的最小生成树中的 Intercellular Edges 提取出来，并把其两端延伸的五阶邻居也提取出来。可使用库里的 single_source_shortest_path 函数来处理，比自己手写的 BFS 效率要高。步骤七对于每个最小生成树，求出其所有节点的度中心性，筛选出前 50 名。每个最小生成树的前 50 名比较异同，用韦恩图展示结果。步骤八对于每个最小生成树，求出其所有节点的介数中心性，筛选出前 50 名。每个最小生成树的前 50 名比较异同，用韦恩图展示结果。代码根据上述思路可得以下代码。import os import networkx as nx import pandas as pd import numpy as np import matplotlib.pyplot as plt from matplotlib_venn import venn3 from sklearn.preprocessing import minmax_scale os.makedirs("Results", exist_ok=True) def get_data(cancer, cell, base_path="InputData"): file_path = os.path.join(base_path, f"{cancer}_{cell}_GeneExpression.csv") df = pd.read_csv(file_path, index_col=0) return df def select_data(df): var = df.var(axis=1) selected = var.sort_values(ascending=False).head(1000).index return df.loc[selected] def build_intracellular_graph(df, cell): graph = nx.Graph() genes = df.index.tolist() for gene in genes: graph.add_node(f"{gene}/{cell}") corr_matrix = df.T.corr(method='spearman').abs() corr_matrix = corr_matrix.fillna(0) upper_triangle = np.triu_indices_from(corr_matrix, k=1) edges = [ (f"{genes[i]}/{cell}", f"{genes[j]}/{cell}") for i, j in zip(upper_triangle[0], upper_triangle[1]) ] weights = 1 - minmax_scale(corr_matrix.values[upper_triangle]) for (u, v), w in zip(edges, weights): graph.add_edge(u, v, weight=w) return graph def add_intercellular_edges(graph, cancer_df, tcell_df, ligand_receptor_file="LigandReceptor_Human.txt"): ligand_receptor_df = pd.read_csv(ligand_receptor_file, sep='\t', header=None, names=['Ligand', 'Receptor']) edges = [] res = [] for _, row in ligand_receptor_df.iterrows(): ligand, receptor = row['Ligand'], row['Receptor'] if ligand in cancer_df.index and receptor in tcell_df.index: mean_ligand = cancer_df.loc[ligand].mean() mean_receptor = tcell_df.loc[receptor].mean() edges.append((f"{ligand}/CancerCell", f"{receptor}/TCell")) res.append(mean_ligand * mean_receptor) if ligand in tcell_df.index and receptor in cancer_df.index: mean_ligand = tcell_df.loc[ligand].mean() mean_receptor = cancer_df.loc[receptor].mean() edges.append((f"{ligand}/TCell", f"{receptor}/CancerCell")) res.append(mean_ligand * mean_receptor) if res: weights = 1 - minmax_scale(res) for (u, v), w in zip(edges, weights): graph.add_edge(u, v, weight=w) def get_minimum_spanning_tree(graph): return nx.minimum_spanning_tree(graph, weight='weight', algorithm='prim') def get_intercellular_edges_and_neighbors(mst): intercellular_edges = [(u, v) for u, v in mst.edges if u.split('/')[-1] != v.split('/')[-1]] nodes = set() for u, v in intercellular_edges: nodes.update(nx.single_source_shortest_path(mst, u, cutoff=5).keys()) nodes.update(nx.single_source_shortest_path(mst, v, cutoff=5).keys()) return mst.subgraph(nodes) def visualize_signal_pathway(graph, cancer): plt.figure(figsize=(40, 24)) pos = nx.circular_layout(graph) node_colors = [] for node in graph.nodes: if node.endswith("/CancerCell"): node_colors.append("red") else: node_colors.append("blue") inter_edges = [(u, v) for u, v in graph.edges if u.split('/')[-1] != v.split('/')[-1]] intra_edges = [(u, v) for u, v in graph.edges if u.split('/')[-1] == v.split('/')[-1]] nx.draw_networkx_nodes(graph, pos, node_color=node_colors, node_size=30, alpha=0.8) nx.draw_networkx_edges( graph, pos, edgelist=intra_edges, edge_color="black", width=0.5, alpha=0.5 ) nx.draw_networkx_edges( graph, pos, edgelist=inter_edges, edge_color="limegreen", width=1.2, alpha=0.8, style="solid", ) nx.draw_networkx_labels( graph, pos, font_size=6, font_family='Arial', font_weight='normal', alpha=0.9 ) plt.title(f"{cancer} signal-pathway") plt.axis('off') plt.tight_layout() plt.savefig(os.path.join("Results", f"{cancer}_signal_pathway.png"), dpi=300) plt.close() print(f"The signal pathway of {cancer} is done.") def calculate_top_centrality_genes(graphs, centrality_type, top_n=50): top_genes_list = [] for graph in graphs: if centrality_type == "degree": centrality = nx.degree_centrality(graph) elif centrality_type == "betweenness": centrality = nx.betweenness_centrality(graph, weight='weight') sorted_genes = sorted(centrality.items(), key=lambda item: item[1], reverse=True)[:top_n] top_genes = {gene for gene, score in sorted_genes} top_genes_list.append(top_genes) return top_genes_list def plot_venn_diagram(sets, labels, title, filename): plt.figure(figsize=(6, 6)) venn3(sets, set_labels=labels) plt.title(title) plt.tight_layout() plt.savefig(os.path.join("Results", f"{filename}.png"), dpi=300) plt.close() def analyze_and_visualize_centrality(cancer_types, mst_list): top_degree_genes = calculate_top_centrality_genes(mst_list, "degree") plot_venn_diagram( sets=top_degree_genes, labels=cancer_types, title="Degree Centrality Top 50 Genes Comparision", filename="degree_centrality_venn" ) print("The degree centrality venn is done.") top_betweenness_genes = calculate_top_centrality_genes(mst_list, "betweenness") plot_venn_diagram( sets=top_betweenness_genes, labels=cancer_types, title="Betweenness Centrality Top 50 Genes Comparision", filename="betweenness_centrality_venn" ) print("The betweenness centrality venn is done.") def main(): cancer_types = ["BreastTumor", "ColonTumor", "LungTumor"] mst_list = [] for cancer in cancer_types: print(f"Processing {cancer}...") cancer_df = select_data(get_data(cancer, "CancerCell")) tcell_df = select_data(get_data(cancer, "TCell")) cancer_graph = build_intracellular_graph(cancer_df, "CancerCell") tcell_graph = build_intracellular_graph(tcell_df, "TCell") graph = nx.compose(cancer_graph, tcell_graph) add_intercellular_edges(graph, cancer_df, tcell_df) mst = get_minimum_spanning_tree(graph) mst_list.append(mst) subgraph = get_intercellular_edges_and_neighbors(mst) visualize_signal_pathway(subgraph, cancer) analyze_and_visualize_centrality(cancer_types, mst_list) if __name__ == "__main__": main()生成结果生成的图像结果如下。（图太大了直接传传不上来，所以压缩了一下，特别糊）
实验目的学习基于MPI分布式并行计算程序设计方法。学习基于MapReduce框架的分布式并行计算程序设计方法。学习基于Spark框架的分布式并行计算程序设计方法。实验题目题目 1编写基于 MPI 的并行计算程序，实现对连续函数 $f(x)=\sin(x^2+𝑥)(x\in R)$ 在区 间 $[a, b]$ 上的定积分求解。要求使用“梯形法（Trapezoidal Rule）”进行近似计算。题目 2输入文件为学生成绩信息，包含了必修课与选修课成绩，格式如下： 班级1, 姓名1, 科目1, 必修, 成绩1 （注： 为换行符） 班级2, 姓名2, 科目1, 必修, 成绩2 班级1, 姓名1, 科目2, 选修, 成绩3 ………., ………, ………, ……… 编写 Hadoop 平台上的 MapReduce 程序，分别实现如下功能：计算每个学生必修课的平均成绩。统计每个班级中所有课程（必修+选修）平均成绩排名前五的学生姓名和成绩题目 3输入文件的每一行为具有父子/父女/母子/母女/关系的一对人名，例如： Tim, Andy Harry, Alice Mark, Louis Andy, Joseph ……….., ………… 假定不会出现重名现象。 编写 Hadoop 平台上的 MapReduce 程序，找出所有具有 grandchild-grandparent 关系的人名对。题目 4输入文件为学生成绩信息，包含了必修课与选修课成绩，格式如下： 班级1, 姓名1, 科目1, 必修, 成绩1 （注： 为换行符） 班级2, 姓名2, 科目1, 必修, 成绩2 班级1, 姓名1, 科目2, 选修, 成绩3 ………., ………, ………, ……… 编写Spark程序，实现如下功能： 按班级统计必修课平均成绩在：90~100、80~89、70~79、60~69 和 60 分以下这 5 个分数段的学生人数。准备工作安装 Microsoft MPI首先需要安装 Microsoft MPI。参考了 windows安装MPI及python安装mpi4py_下载mpi4py代码在vmware-CSDN博客。直接照着下载安装配置环境变量即可。安装 Docker安装 WSL 2Docker Desktop for Windows 需要 WSL 2 支持，所以需要先安装 WSL 2。参考了 # Windows10安装WSL的Ubuntu20.04系统及踩坑记录安装 Docker参考了 在国内 Windows 平台上安装 Docker 的详细教程_docker windows intel-CSDN博客部署 Hadoop 和 Spark 实验环境git clone 一下 hadoop-sandbox/hadoop-sandbox: A fully-functional Hadoop Yarn cluster as docker-compose deployment.然后切换到解压后的目录下，使用 powershell 执行 docker-compose up -d，会根据该目录下 .yml 文件来下载镜像并启动容器。接着执行 docker-compose down 来关闭启动的容器。下载老师提供的 spark-install.zip 压缩包，并解压。然后用 powershell 切换到解压后的目录，并执行 docker build -t packet23/hadoop-client:latest . ，此命令将根据 Dockerfile 的指示在名为 hadoop-client 的 Docker 镜像中下载并安装 Python3.8，Spark-3.2.1 和 apache-maven-3.8.5。上述操作结束后就成功生成了实验所需的全部 Docker 镜像。为 clientnode 安装 mrjob 库由于我使用的语言为 Python，所以需要为容器安装 mrjob 库。由于容器内没有 pip，所以先要对其安装 pip。先 docker ps 查看 clientnode 容器 ID，然后 docker exec -it --user root clientnode bash （其中 clientnode 换成容器 ID），进入后再执行apt update apt install -y python3-pip pip3 install mrjob如此，便安装好了 mrjob 库。为 nodemanager 安装 python3再测试老师给的代码时，发现 nodemanager 没有安装 python3 ，无法运行。所以需要对其安装 python3。先 docker ps 查看 nodemaanager 容器 ID，然后 docker exec -it --user root nodemanager bash （其中 nodemanager 换成容器 ID），进入后再执行apt-get update apt-get install -y python3 python3-pip如此，便成功为 nodemanager 安装了 python3 。实验思路题目 1梯形法计算定积分近似值。考虑将定积分区间分成 n 个子区间，然后再均分给进程（开的进程数是 n 的因数，假设为 size ），每个进程计算 n // size 个 子区间用梯形法计算出的值的和，然后再用 Reduce 函数来把各个进程的结果用 MPI.SUM 累加起来即可。题目 2对于第一个要求，计算每个学生必修课的平均成绩，在 map 阶段只需将每一行信息转换成键值对 {(班级, 姓名), (必修课分数, 1)}，键加入班级是为了避免重名问题。然后在 reduce 阶段对同一个人的成绩进行聚合，计算每个人的必修课平均成绩即可。在做第一个要求的时候遇到了一个问题，花了很长时间才解决，那就是输出的文件的中文总是会变成 unicode 编码。问了很久大模型，提供的方法都无法解决。最后无果去 google 了一下，在 mrjob 的 github 仓库找到了一个 issue，也是跟我一样的问题，参考了一下成功解决了。看来还是不能太依赖大模型（当然也有可能是我提问的姿势不太对）。对于第二个要求，在 map 阶段把每一行信息转换为键值对 {班级, (姓名, 分数, 1)}。在 reduce 阶段对同一个班级的学生成绩，开一个字典统计其总分与科目数，用于计算平均成绩，接着根据平均成绩进行从高到低排序，取前五即可。题目 3根据给出的亲子关系可以画出一个图，要找 grandchild-grandparent 关系的话，只需找一个中介点，其所有子代与父代相互之间构成 grandchild-grandparent 关系。所以，在 map 阶段，将每行的信息 A,B，输出两个键值对 {A, ("child", B)}, {B, ("parent", A)}，分别代表 A 的孩子是 B，B 的父母是 A。在 reduce 阶段，对同一个人的信息进行聚合，开两个列表分别存储这个人的孩子与父母，接着把这两个列表相互之间输出 grandchild-grandparent 关系即可。题目 4显然，这个题目需要用到题目 2 的要求 1 生成的结果。这个结果的格式为 班级 姓名 必修课平均成绩。所以读入之后，先对其进行两次 map，第一次将信息按照空格分割成一个列表，第二次将列表的第一个字段和第三个字段（班级和分数，因为不需要关注姓名）形成键值对，形成了一个键值对 rdd。得到这个键值对 rdd 后，可以写一个函数来把分数映射到对应的区间，然后生成形如 {班级, (分数区间, 1)} 的键值对，这个只需写好这个分数映射函数之后使用 mapValues 算子即可。然后，是对同一个班级的同一个分数区间进行计数统计，所以可以把原键值对 {班级, (分数区间, 1)} 变成一个新的键值对 {(班级, 分数区间), 1)} ，这样方便统计结果。这个也是用 map 算子即可。然后统计结果就很简单了，直接用 reduceByKey 即可。为了使结果更清晰，我使用 sortByKey 函数把结果按照区间从小到大的顺序来输出了。代码题目 1from mpi4py import MPI from math import sin def f(x): return sin(x * x + x) comm = MPI.COMM_WORLD rank = comm.Get_rank() size = comm.Get_size() if rank == 0: a = float(input("请输入定积分下限：")) b = float(input("请输入定积分上限：")) else : a = None b = None a = comm.bcast(a, root=0) b = comm.bcast(b, root=0) n = 10000000 chunk_size = n // size length = (b - a) / n partial_res = 0 for i in range(0, chunk_size): l = a + (rank * chunk_size + i) * length partial_res += length * (f(l) + f(l + length)) * 0.5 total_res = comm.reduce(partial_res, MPI.SUM, root=0) if rank == 0: print(f"近似值为：{total_res}")题目 21.# aveScore.py from mrjob.job import MRJob from mrjob.protocol import RawProtocol class MRAveScore(MRJob): OUTPUT_PROTOCOL = RawProtocol # Mapper函数：把每一行信息分解出来，然后为每一行信息生成一个键值对{(班级, 姓名), (必修课分数, 1)} def mapper(self, _, line): cls, name, subject, subject_type, score = line.strip().split(',') score = float(score) if subject_type == "必修": yield (cls, name), (score, 1) # Reducer函数：对同一个人的成绩进行聚合，计算每个人的必修课平均成绩。 def reducer(self, key, values): total_score = 0 total_num = 0 for score, num in values: total_score += score total_num += num average_score = total_score / total_num if total_num else 0 yield f"{key[0]} {key[1]}", f"{average_score:.2f}" if __name__ == '__main__': MRAveScore().run()2.# top5.py from mrjob.job import MRJob from mrjob.protocol import RawProtocol class MRTop5(MRJob): OUTPUT_PROTOCOL = RawProtocol # Mapper函数：把每一行的信息分解出来，生成键值对 {班级, (姓名, 分数, 1)} def mapper(self, _, line): cls, name, subject, subject_type, score = line.strip().split(',') score = float(score) yield cls, (name, score, 1) # Reducer函数：对同一个班级的学生信息进行聚合，开一个 scores 字典用来统计该班级内的学生总分与科目数，用于计算平均成绩 def reducer(self, key, values): scores = {} for name, score, num in values: if name not in scores: scores[name] = [0.0, 0] scores[name][0] += score scores[name][1] += num avg_scores = [(name, score / num) for name, (score, num) in scores.items()] top5 = sorted(avg_scores, key=lambda x: x[1], reverse=True)[:5] for name, score in top5: yield key, f"{name}\t{score:.2f}" if __name__ == '__main__': MRTop5.run()题目 3from mrjob.job import MRJob class MRFindGrandparents(MRJob): # Mapper函数：输出两个键值对 `{A, ("child", B)}, {B, ("parent", A)}`，分别代表 A 的孩子是 B，B 的父母是 A。 def mapper(self, _, line): parent, child = line.strip().split(',') yield parent, ("child", child) yield child, ("parent", parent) # Reducer函数：对同一个人的信息进行聚合，生成 grandchild-grandparent 关系 def reducer(self, key, values): parents = [] children = [] for relation, name in values: if relation == "child": children.append(name) else: parents.append(name) for child in children: for parent in parents: yield child, parent if __name__ == '__main__': MRFindGrandparents.run()题目 4from pyspark import SparkContext # 分数映射到区间函数 def score_to_range(score): if score < 60: return "0~59" elif score < 70: return "60~69" elif score < 80: return "70~79" elif score < 90: return "80~89" else: return "90~100" def score_statistics(input_file): sc = SparkContext("local", "Score Statistics") data_rdd = sc.textFile(input_file) class_score = data_rdd.map(lambda line: line.split()).map(lambda fields: (fields[0], float(fields[2]))) # 分数映射到区间 class_score_ranges = class_score.mapValues(lambda score: (score_to_range(score), 1)) # 将键变为 {班级, 分数区间}，方便统计数量 class_score_counts = class_score_ranges.map(lambda x: ((x[0], x[1][0]), x[1][1])) class_score_result = class_score_counts.reduceByKey(lambda x, y: x + y) result = class_score_result.sortByKey(ascending=True, numPartitions=1).map(lambda x: (x[0][0], x[0][1], x[1])) result.saveAsTextFile("result") sc.stop() if __name__ == "__main__": input_file = "input.txt" score_statistics(input_file)运行示例题目 1题目 21.2.题目 3题目 4
实验内容准备工作环境配置根据老师发的教程安装 activemq 即可。实验环境系统：Windows 11语言：JavaIDE：Intellij IDEA使用 maven 管理项目。pom.xml 文件如下：<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>HoshiuZ</groupId> <artifactId>Log</artifactId> <version>1.0-SNAPSHOT</version> <properties> <maven.compiler.source>19</maven.compiler.source> <maven.compiler.target>19</maven.compiler.target> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> </properties> <dependencies> <dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-all</artifactId> <version>6.1.6</version> </dependency> <dependency> <groupId>jakarta.jms</groupId> <artifactId>jakarta.jms-api</artifactId> <version>3.1.0</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-api</artifactId> <version>2.17.2</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-core</artifactId> <version>2.17.2</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.13.0</version> </dependency> <dependency> <groupId>org.jfree</groupId> <artifactId>jfreechart</artifactId> <version>1.5.3</version> </dependency> </dependencies> </project>jackson 用于 JSON 与对象的转化，jfreechart 用于画折线图，activemq 为使用的消息队列。思路分析先对实验题目进行分析。对于第一个日志采集节点，显然其是一个生产者；第二个异常检测与统计分析微服务，其显然是一个消费者，但是对于其功能二三，其也要生成消息发送到消息队列，所以其既是一个消费者也是一个生产者；第三个实时日志可视化监控微服务，其显然是一个消费者。对于发送 json 数据，可以考虑使用 java 中的 Jackson 库，可以比较方便的将数据打包成 json 发送到消息队列。对于日志采集节点的“每个节点用唯一 ID 标识”这一要求，可以考虑对该程序每次传一个参数进去来作为 ID 标识。需要三个消息队列，一个日志队列 logs，一个分析结果队列 analysisResults ，一个严重告警队列 criticalAlerts 。多个任务采集节点只需一个 producer 。每隔 100 毫秒向日志队列 logs 发送一个日志，其 log_level 可以随机生成。异常检测与统计分析微服务需要一个 consumer 和两个 producer 。一直从日志队列 logs 接收日志，对每个设备维护最近的 N 条日志，可以用队列实现，当队列大小大于 N 时就弹出元素。每隔 T 秒将分析结果打包成新消息，这个可以用 java 中的 Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate() 方法来实现，只需再写一个方法来计算结果并发送。可以动态地维护 N 条日志内错误与警告的数量，然后结果直接与队列大小相除即可。S 秒内的 ERROR 占比，可以对每个设备再维护一个队列，里面存放 S 秒内的日志，当队头日志与当前日志的时间相差超过了 S 秒，就弹出队头。同理动态维护错误的数量，计算结果大于百分之五十的时候就发送严重告警。由于用到了多线程，所以 map 和 queue 要用线程安全的 ConcurrentHashMap 和 ConcurrentLinkedDeque。实时日志可视化监控微服务一个纯消费者，从异常检测与统计分析微服务生产的两个队列接收消息，所以是两个 consumer。对于显示 WARN/ERROR 占比、最近一次 ERROR 事件、严重告警状态次数等信息，直接输出消息队列中获取的信息与监听器中统计的信息即可。对于绘制折线图，可以使用 Java 的 JFreeChart ，在 maven 的 pom.xml 中添加依赖项后即可使用。项目结构日志采集节点为类 Publisher 。三个消息队列的类似于 JSON 的消息为类 LogEntry, AnalysisResult, CriticalAlert 。异常检测与统计分析微服务为类 LogStatsThread，其监听器为类 LogStatsListener。实时日志可视化监控微服务的消费者类为类 LogVisualConsumerThread，其两个监听器分别为类 AnalysisResultListener 与 CriticalAlertListener。显示数据为类 ShowDeviceData，画图为类 LogVisualDraw。多类共用的数据用了类 Storage 来处理。主类 Main 用于多线程启用功能。结构图如下：classDiagram class Main { +main(String[] args) } class Publisher { +send(LogEntry) } class LogEntry { -device_id: String -timestamp: String -log_level: String -message: String } class AnalysisResult { -device_id: String -errorRatio: double -warnRatio: double -lastErrorTime: String } class CriticalAlert { -device_id: String -timestamp: String -msg: String } class Storage { +addAnalysisResult(device_id, AnalysisResult) +updateCriticalAlert(device_id) +updateLastErrorTimestamp(device_id, timestamp) +getLastErrorTimestamp(device_id): String } class LogStatsThread { +run() } class LogStatsListener { +onMessage(Message) +analyzeAndSend() } class LogVisualConsumerThread { +run() } class AnalysisResultListener { +onMessage(Message) } class CriticalAlertListener { +onMessage(Message) } class ShowDeviceData class LogVisualDraw %% 关联关系 Main --> LogStatsThread Main --> LogVisualConsumerThread Publisher --> LogEntry LogStatsThread --> LogStatsListener LogStatsListener --> Storage LogStatsListener --> LogEntry LogStatsListener --> AnalysisResult LogStatsListener --> CriticalAlert LogVisualConsumerThread --> AnalysisResultListener LogVisualConsumerThread --> CriticalAlertListener AnalysisResultListener --> AnalysisResult CriticalAlertListener --> CriticalAlert AnalysisResultListener --> Storage CriticalAlertListener --> Storage ShowDeviceData --> AnalysisResult ShowDeviceData --> CriticalAlert LogVisualDraw --> ShowDeviceData Storage --> AnalysisResult Storage --> CriticalAlert Storage --> LogEntry 代码LogEntry 类public class LogEntry { public String device_id; public String timestamp; public String log_level; public String message; public LogEntry() {} public LogEntry(String device_id, String timestamp, String log_level, String message) { this.device_id = device_id; this.timestamp = timestamp; this.log_level = log_level; this.message = message; } }Publisher 类public class Publisher { private static String brokerURL = "tcp://localhost:61616"; private static ConnectionFactory factory; private static Connection connection; private Session session; private MessageProducer producer; private String deviceid; public Publisher() throws JMSException { factory = new ActiveMQConnectionFactory(brokerURL); connection = factory.createConnection(); connection.start(); session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); producer = session.createProducer(null); deviceid = null; } public void close() throws JMSException { if (connection != null) { connection.close(); } } public void sendMessage() throws JMSException { Destination destination = session.createQueue("logs"); ObjectMapper mapper = new ObjectMapper(); Random random = new Random(); String[] levels = {"INFO", "WARN", "ERROR"}; while(true) { String timestamp = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")); String level = levels[random.nextInt(levels.length)]; String message = switch (level) { case "WARN" -> "警告"; case "ERROR" -> "错误"; default -> "正常"; }; LogEntry logentry = new LogEntry(deviceid, timestamp, level, message); String json = ""; try { json = mapper.writeValueAsString(logentry); } catch (Exception e) { e.printStackTrace(); } Message msg = session.createTextMessage(json); producer.send(destination, msg); System.out.println("Sent message: " + json); try{ Thread.sleep(100); } catch(InterruptedException e) { Thread.currentThread().interrupt(); } } } public static void main(String[] args) throws JMSException { if(args.length == 0) { System.out.println("Please provide the device_id as a program parameter."); return; } Publisher publisher = new Publisher(); publisher.deviceid = args[0]; publisher.sendMessage(); publisher.close(); } }AnalysisResult 类public class AnalysisResult { public String device_id; public double errorRatio; public double warnRatio; public String lastErrorTime; public AnalysisResult() {} public AnalysisResult(String device_id, double errorRatio, double warnRatio, String lastErrorTime) { this.device_id = device_id; this.errorRatio = errorRatio; this.warnRatio = warnRatio; this.lastErrorTime = lastErrorTime; } }CriticalAlert 类public class CriticalAlert { public String device_id; public String timestamp; public String msg; public CriticalAlert() {} public CriticalAlert(String device_id, String timestamp, String msg) { this.device_id = device_id; this.timestamp = timestamp; this.msg = msg; } }Storage 类public class Storage { private final Map<String, Deque<AnalysisResult>> analysisResultsQueue = new ConcurrentHashMap<>(); private final Map<String, String> lastErrorTimestamp = new ConcurrentHashMap<>(); private final Map<String, Boolean> deviceIdFlag = new ConcurrentHashMap<>(); private int criticalAlertCount = 0; private static final int MAX_HISTORY_SIZE = 20; public void addAnalysisResult(String deviceId, AnalysisResult analysisResult) { deviceIdFlag.put(deviceId, true); analysisResultsQueue.compute(deviceId, (k, deque) -> { if(deque == null) { deque = new ConcurrentLinkedDeque<>(); } deque.addLast(analysisResult); if(deque.size() > MAX_HISTORY_SIZE) { deque.removeFirst(); } return deque; }); } public boolean checkDeviceId(String deviceId) { return deviceIdFlag.containsKey(deviceId); } public void updateCriticalAlert(String deviceId, CriticalAlert criticalAlert) { criticalAlertCount++; } public int getCriticalAlertCount() { return criticalAlertCount; } public Deque<AnalysisResult> getAnalysisResults(String deviceId) { return analysisResultsQueue.getOrDefault(deviceId, new ConcurrentLinkedDeque<>()); } public void updateLastErrorTimestamp(String deviceId, String timestamp) { lastErrorTimestamp.put(deviceId, timestamp); } public String getLastErrorTimestamp(String deviceId) { return lastErrorTimestamp.getOrDefault(deviceId, "No error."); } }LogStatsListener 类public class LogStatsListener implements MessageListener { private static final int N = 200; private static final int T = 5; private static final int S = 1; private final Storage storage; private Map<String, Deque<LogEntry>> deviceLogs = new ConcurrentHashMap<>(); private Map<String, Integer> errorCount = new ConcurrentHashMap<>(); private Map<String, Integer> warnCount = new ConcurrentHashMap<>(); private Map<String, Deque<LogEntry>> deviceLogsWithinSSeconds = new ConcurrentHashMap<>(); private Map<String, Integer> errorCountWithinSSeconds = new ConcurrentHashMap<>(); private Session session; private MessageProducer producer1, producer2; private static class TimedLogEntry{ long timestamp; String logLevel; TimedLogEntry(long timestamp, String logLevel){ this.timestamp = timestamp; this.logLevel = logLevel; } } private static class TimeUtil{ private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); public static long toTimestampMillis(String timeStr){ LocalDateTime localDateTime = LocalDateTime.parse(timeStr, formatter); return localDateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); } } public LogStatsListener(Session session, MessageProducer producer1, MessageProducer producer2, Storage storage) throws JMSException { this.session = session; this.producer1 = session.createProducer(session.createQueue("analysisResults")); this.producer2 = session.createProducer(session.createQueue("criticalAlerts")); this.storage = storage; Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(this::analyzeAndSend, T, T, TimeUnit.SECONDS); } @Override public void onMessage(Message message) { try{ if(!(message instanceof TextMessage textMessage)) return; String msg = textMessage.getText(); ObjectMapper mapper = new ObjectMapper(); LogEntry log = null; try { log = mapper.readValue(msg, LogEntry.class); } catch (JsonProcessingException e) { e.printStackTrace(); return; } String deviceId = log.device_id; deviceLogs.putIfAbsent(deviceId, new ConcurrentLinkedDeque<>()); Deque<LogEntry> logs = deviceLogs.get(deviceId); if(logs.size() >= N) { LogEntry removedLog = logs.pollFirst(); if("ERROR".equalsIgnoreCase(removedLog.log_level)) { errorCount.put(deviceId, errorCount.get(deviceId) - 1); } else if("WARN".equalsIgnoreCase(removedLog.log_level)) { warnCount.put(deviceId, warnCount.get(deviceId) - 1); } } logs.addLast(log); if("ERROR".equalsIgnoreCase(log.log_level)) { errorCount.put(deviceId, errorCount.getOrDefault(deviceId, 0) + 1); storage.updateLastErrorTimestamp(deviceId, log.timestamp); } else if("WARN".equalsIgnoreCase(log.log_level)) { warnCount.put(deviceId, warnCount.getOrDefault(deviceId, 0) + 1); } deviceLogsWithinSSeconds.putIfAbsent(deviceId, new ConcurrentLinkedDeque<>()); logs = deviceLogsWithinSSeconds.get(deviceId); DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); LocalDateTime now = LocalDateTime.now(); while(!logs.isEmpty()) { LogEntry firstLog = logs.peekFirst(); LocalDateTime firstTime = LocalDateTime.parse(firstLog.timestamp, formatter); if(java.time.Duration.between(firstTime, now).getSeconds() > S) { LogEntry removedLog = logs.pollFirst(); if("ERROR".equalsIgnoreCase(removedLog.log_level)) { errorCountWithinSSeconds.put(deviceId, errorCountWithinSSeconds.get(deviceId) - 1); } } else { break; } } logs.addLast(log); if("ERROR".equalsIgnoreCase(log.log_level)) { errorCountWithinSSeconds.put(deviceId, errorCountWithinSSeconds.getOrDefault(deviceId, 0) + 1); } } catch (JMSException e) { e.printStackTrace(); } } private void analyzeAndSend() { long currentTime = System.currentTimeMillis(); ObjectMapper mapper = new ObjectMapper(); for(String deviceId : deviceLogs.keySet()) { Deque<LogEntry> logs = deviceLogs.get(deviceId); int total = logs.size(); int errors = errorCount.getOrDefault(deviceId, 0); int warns = warnCount.getOrDefault(deviceId, 0); double errorRatio = total > 0 ? (double) errors / total : 0.0; double warnRatio = total > 0 ? (double) warns / total : 0.0; String lastErrorTime = storage.getLastErrorTimestamp(deviceId); AnalysisResult analysisResult = new AnalysisResult(deviceId, errorRatio, warnRatio, lastErrorTime); try{ String analysisJson = mapper.writeValueAsString(analysisResult); Message analysisMsg = session.createTextMessage(analysisJson); producer1.send(analysisMsg); logs = deviceLogsWithinSSeconds.get(deviceId); int totalWithinSSeconds = logs.size(); int errorsWithinSseconds = errorCountWithinSSeconds.get(deviceId); double errorRatioWithinSseconds = totalWithinSSeconds > 0 ? (double) errorsWithinSseconds / totalWithinSSeconds : 0.0; if(errorRatioWithinSseconds > 0.5) { String timestamp = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")); String msg = String.format("The ERROR rate over the past %d seconds has exceeded 50%%, with a ratio of %d/%d", S, errorsWithinSseconds, totalWithinSSeconds); CriticalAlert criticalAlert = new CriticalAlert(deviceId, timestamp, msg); String alertJson = mapper.writeValueAsString(criticalAlert); Message alertMsg = session.createTextMessage(alertJson); producer2.send(alertMsg); } } catch (JsonProcessingException | JMSException e) { e.printStackTrace(); } } } }LogStatsThread 类public class LogStatsThread implements Runnable { private static String brokerURL = "tcp://localhost:61616"; private static ConnectionFactory factory; private static Connection connection; private Session session; private MessageProducer producer1, producer2; private MessageConsumer consumer; private LogStatsListener logStatsListener; private final Storage storage; public LogStatsThread(Storage storage) { this.storage = storage; } public void close() throws JMSException { if (connection != null) { connection.close(); } } @Override public void run() { try { factory = new ActiveMQConnectionFactory(brokerURL); connection = factory.createConnection(); connection.start(); session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); producer1 = session.createProducer(session.createQueue("analysisResults")); producer2 = session.createProducer(session.createQueue("criticalAlerts")); consumer = session.createConsumer(session.createQueue("logs")); logStatsListener = new LogStatsListener(session, producer1, producer2, storage); consumer.setMessageListener(logStatsListener); System.in.read(); } catch (Exception e) { e.printStackTrace(); } } }AnalysisResultListener 类public class AnalysisResultListener implements MessageListener { private final Storage storage; public AnalysisResultListener(Storage storage) { this.storage = storage; } @Override public void onMessage(Message message) { try{ if(!(message instanceof TextMessage textMessage)) return; String msg = textMessage.getText(); ObjectMapper mapper = new ObjectMapper(); AnalysisResult analysisResult = mapper.readValue(msg, AnalysisResult.class); storage.addAnalysisResult(analysisResult.device_id, analysisResult); } catch (JMSException | JsonProcessingException e) { e.printStackTrace(); } } }CriticalAlertListener 类public class CriticalAlertListener implements MessageListener { private final Storage storage; public CriticalAlertListener(Storage storage) { this.storage = storage; } @Override public void onMessage(Message message) { try{ if(!(message instanceof TextMessage textMessage)) return; String msg = textMessage.getText(); ObjectMapper mapper = new ObjectMapper(); CriticalAlert criticalAlert = mapper.readValue(msg, CriticalAlert.class); storage.updateCriticalAlert(criticalAlert.device_id, criticalAlert); } catch (JMSException | JsonProcessingException e) { e.printStackTrace(); } } }ShowDeviceData 类public class ShowDeviceData { private final Storage storage; public ShowDeviceData(Storage storage) { this.storage = storage; } public void showData(String deviceId) { Deque<AnalysisResult> analysisResults = storage.getAnalysisResults(deviceId); if(!analysisResults.isEmpty()) { AnalysisResult latestResult = analysisResults.peekLast(); System.out.printf("Now the ratio of error is %.2f%%.%n", latestResult.errorRatio * 100); System.out.printf("Now the ratio of warn is %.2f%%.%n", latestResult.warnRatio * 100); System.out.printf("The latest error timestamp is %s.%n", storage.getLastErrorTimestamp(deviceId)); } } }LogVisualDraw 类public class LogVisualDraw { private static final int T = 5; public static void draw(String deviceId, Deque<AnalysisResult> results) throws Exception { DefaultCategoryDataset dataset = new DefaultCategoryDataset(); int size = results.size(); for(AnalysisResult result : results) { dataset.addValue(result.errorRatio * 100, "ERROR Ratio", String.valueOf(-T * size)); dataset.addValue(result.warnRatio * 100, "WARN Ratio", String.valueOf(-T * size)); size--; } JFreeChart chart = ChartFactory.createLineChart( "LogVisualization of device " + deviceId, "Time(s)", "Ratio(%)", dataset, PlotOrientation.VERTICAL, true, true, false ); File dir = new File("pic"); if(!dir.exists()) { dir.mkdirs(); } String timestamp = String.valueOf(System.currentTimeMillis() / 1000); String filePath = "pic/" + deviceId + "_" + timestamp + ".png"; File outputFile = new File(filePath); ChartUtils.saveChartAsPNG(outputFile, chart, 800, 600); System.out.println("The chart is saved to :" + outputFile.getAbsolutePath()); } }LogVisualConsumerThread 类public class LogVisualConsumerThread implements Runnable{ private final Storage storage; private volatile boolean running = true; public LogVisualConsumerThread(Storage storage) { this.storage = storage; } @Override public void run() { try{ String brokerURL = "tcp://localhost:61616"; ConnectionFactory factory = new ActiveMQConnectionFactory(brokerURL); Connection connection = factory.createConnection(); connection.start(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); MessageConsumer consumer1 = session.createConsumer(session.createQueue("analysisResults")); MessageConsumer consumer2 = session.createConsumer(session.createQueue("criticalAlerts")); AnalysisResultListener analysisResultListener = new AnalysisResultListener(storage); CriticalAlertListener criticalAlertListener = new CriticalAlertListener(storage); consumer1.setMessageListener(analysisResultListener); consumer2.setMessageListener(criticalAlertListener); System.in.read(); } catch (Exception e) { e.printStackTrace(); } } public void stop(){ running = false; } }Main 类public class Main { public static void main(String[] args) throws Exception { Storage storage = new Storage(); Thread logStatsThread = new Thread(new LogStatsThread(storage)); Thread logVisualConsumerThread = new Thread(new LogVisualConsumerThread(storage)); ShowDeviceData showDeviceData = new ShowDeviceData(storage); logStatsThread.start(); logVisualConsumerThread.start(); String choice = "1"; while(!"exit".equalsIgnoreCase(choice)) { Scanner sc = new Scanner(System.in); System.out.println("--------------------------------------------------"); System.out.println("Enter device id: "); String deviceId = sc.nextLine(); if(storage.checkDeviceId(deviceId)) { System.out.println("The log information of this device is as follows: "); showDeviceData.showData(deviceId); System.out.println("The number of severe alarms is " + storage.getCriticalAlertCount() + "."); System.out.println("The line chart is as follows: "); Deque<AnalysisResult> result = storage.getAnalysisResults(deviceId); LogVisualDraw.draw(deviceId, result); System.out.println("Enter anything to continue and enter exit to quit"); choice = sc.nextLine(); if ("exit".equalsIgnoreCase(choice)) { break; } System.out.println("--------------------------------------------------"); } else { System.out.println("The device id does not exist."); } } } }运行示例开了三个 publisher，deviceid 分别为 1, 2, 3.activemq 控制台界面如下：得到的部分折线图如下：
实验内容准备工作我选择的语言是 Python，选用的中间件是 gRPC。环境的配置参考了 快速入门 | Python | gRPC 框架写 .proto文件.proto 文件是 gRPC 的核心部分，它定义了通信使用的消息结构与RPC的接口方法。根据实验内容，可以写出以下 .proto 文件。我新加了一个叫做 getStorageSize 的接口，因为我认为在实际应用中，预约ID应当是服务端生成给客户端的，而最简单的ID生成方法就是从 1 开始挨个排，所以可以在服务端中定义一个全局的字典用来存储预约信息，每次给的 ID 就是字典内元素数量加 1。所以我们需要实现一个给客户端的接口用来获取这个服务端的全局的字典内的元素个数。syntax = 'proto3'; message Appointment{ int32 appointmentId = 1; string patientName = 2; string doctorName = 3; string department = 4; string date = 5; string timeSlot = 6; } message BoolResult{ bool res = 1; } message AppointmentId{ int32 appointmentId = 1; } message PatientName{ string patientName = 1; } message AppointmentList{ repeated Appointment appointments = 1; } message Empty{} message StorageSize{ int32 size = 1; } service AppointmentService{ rpc bookAppointment(Appointment) returns (BoolResult); rpc queryById(AppointmentId) returns (Appointment); rpc queryByPatient(PatientName) returns (AppointmentList); rpc cancelAppointment(AppointmentId) returns (BoolResult); rpc getStorageSize(Empty) returns (StorageSize); }使用 protoc 生成 Python 代码使用 cmd，切换到对应目录下，由于我的文件名为 HospitalAppointment.proto，所以在 cmd 中输入以下命令来生成 Python 代码。python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. HospitalAppointment.proto这会生成两个文件：HospitalAppointment_pb2.py：包含消息类定义HospitalAppointment_pb2_grpc.py：包含服务的接口定义编写服务端代码创建 Python 文件 server.py，实现 .proto 文件中定义的接口，并启用 gRPC 服务监听端口。这几个接口其实并不难写，每个接口都有 request, context 两个参数，而这个 request 就相当于客户端向服务端请求的数据，对于每个方法，这个数据的内容就是 .proto 文件里所定义的对应的 message 。可以写出以下的 Server.py。from concurrent import futures import grpc import HospitalAppointment_pb2 import HospitalAppointment_pb2_grpc appointment_storage = {} class AppointmentServiceServicer(HospitalAppointment_pb2_grpc.AppointmentServiceServicer): # 预约挂号 def bookAppointment(self, request, context): print("预约挂号的信息为：") print(request) print("------------------------------") BoolResult = getattr(HospitalAppointment_pb2, 'BoolResult') appointment_storage[request.appointmentId] = request return BoolResult(res=True) # 通过ID查询 def queryById(self, request, context): print("查询的ID为：") print(request) print("------------------------------") Appointment = getattr(HospitalAppointment_pb2, 'Appointment') appointment = appointment_storage.get(request.appointmentId) if appointment is None: print("没有找到该挂号ID") return Appointment() else: return appointment # 通过病人姓名查询 def queryByPatient(self, request, context): print("查询的病人姓名为：") print(request) print("------------------------------") patient_name = request.patientName AppointmentList = getattr(HospitalAppointment_pb2, 'AppointmentList') appointment_list = AppointmentList() for appointment in appointment_storage.values(): if appointment.patientName == patient_name: appointment_list.appointments.add( appointmentId = appointment.appointmentId, patientName = appointment.patientName, doctorName = appointment.doctorName, department = appointment.department, date = appointment.date, timeSlot = appointment.timeSlot ) if len(appointment_list.appointments) == 0: print("没有找到该病人挂的号") return appointment_list else: return appointment_list # 取消预约 def cancelAppointment(self, request, context): print("取消的挂号ID为：") print(request) print("------------------------------") BoolResult = getattr(HospitalAppointment_pb2, 'BoolResult') appointmentId = request.appointmentId if appointmentId in appointment_storage: del appointment_storage[appointmentId] return BoolResult(res=True) else: print("没有找到该挂号ID，取消失败") # 获取当前预约信息数量 def getStorageSize(self, request, context): StorageSize = getattr(HospitalAppointment_pb2, 'StorageSize') size = len(appointment_storage) return StorageSize(size=size) def serve(): server = grpc.server(futures.ThreadPoolExecutor(max_workers=10)) HospitalAppointment_pb2_grpc.add_AppointmentServiceServicer_to_server(AppointmentServiceServicer(), server) print("正在启动服务端") server.add_insecure_port('[::]:50051') server.start() print("服务端已启动，在端口 50051 监听") server.wait_for_termination() if __name__ == '__main__': serve()编写客户端代码创建 Client.py。外层一个 while True，然后给用户五个选择，除了实验内容要求的四个功能外，添加了一个退出功能用于退出 while 循环。from http.client import responses import grpc from pkg_resources import empty_provider import HospitalAppointment_pb2 import HospitalAppointment_pb2_grpc def run(): # 连接到 gRPC 服务器 channel = grpc.insecure_channel('localhost:50051') stub = HospitalAppointment_pb2_grpc.AppointmentServiceStub(channel) Appointment = getattr(HospitalAppointment_pb2, 'Appointment') AppointmentId = getattr(HospitalAppointment_pb2, 'AppointmentId') PatientName = getattr(HospitalAppointment_pb2, 'PatientName') Empty = getattr(HospitalAppointment_pb2, 'Empty') while True: print("请选择功能（输入对应的数字）：") print("1.预约挂号") print("2.通过ID查询预约信息") print("3.通过病人查询预约信息") print("4.取消预约") print("5.退出") opt = int(input()) # 预约挂号 if opt == 1: empty = Empty() size = stub.getStorageSize(empty) appointmentid = size.size + 1 patientname = input("请输入病人姓名：") doctorname = input("请输入医生姓名：") department = input("请输入科室：") date = input("请输入预约日期：") timeSlot = input("请输入预约时间段：") appointment = Appointment( appointmentId = appointmentid, patientName = patientname, doctorName=doctorname, department=department, date=date, timeSlot=timeSlot ) response = stub.bookAppointment(appointment) if response.res: print(f"预约成功，预约ID为： {appointmentid}" ) else: print("预约失败") print("------------------------------") # 通过ID查询 elif opt == 2: appointmentid = int(input("请输入查询的预约ID：")) query_by_id = AppointmentId(appointmentId=appointmentid) response = stub.queryById(query_by_id) if not response.ListFields(): print("没有查询到该ID对应的信息") else: print("查询结果为：") print(response) print("------------------------------") # 通过病人姓名查询 elif opt == 3: patientname = input("请输入要查询的病人姓名：") query_by_name = PatientName(patientName=patientname) response = stub.queryByPatient(query_by_name) if len(response.appointments) == 0: print("没有该病人的挂号记录") else: print("查询结果为：") print(response) print("------------------------------") elif opt == 4: appointmentid = int(input("请输入需要取消预约的ID：")) cancel_appointment = AppointmentId(appointmentId=appointmentid) response = stub.cancelAppointment(cancel_appointment) if response.res: print("取消成功") else: print("取消失败") else: break if __name__ == '__main__': run()运行示例
准备工作安装 jdk参考了 最详细jdk安装以及配置环境（保姆级教程）_jdk环境配置-CSDN博客安装 maven 并为 maven 设置阿里云镜像仓库参考了 全站最全Maven下载安装配置教学（2024更新...全版本）建议收藏...赠送IDEA配置Maven教程-CSDN博客用 maven 编译 Helloworld 版本的 java 程序用 cmd 在 DClab 目录下键入mvn archetype:generate -DgroupId=hoshiuz -DartifactId=helloworld -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false即可在 DClab 下创建 helloworld 项目。接着用 cmd 进入项目目录，使用 mvn compile 来编译项目。编译完成后，在 cmd 中键入 java -cp target/classes com.example.App来运行程序，可以在 cmd 中看到结果 Hello World!。题目基于 Socket 编写支持多客户端并发访问的 Key-Value 存储服务器，客户端可以存储（PUT key value）和检索（GET key）数据。数据存储在内存中（或使用简单的文件持久化）。服务器支持的指令格式：PUT key value ——存储键值对GET key——获取 key 对应的 valueDELETE key——删除 key 及其对应的 valueEXIT——关闭客户端连接用 JMeter 测试你的服务器程序的并发性能（平均响应时间和 TPS 指标）。JMeter 下载安装参考了 【超详细】Jmeter安装配置详细教程_jmeter安装教程以及jdk环境配置-CSDN博客设计思路可以考虑直接在服务端开一个静态的 HashMap，由于要实现多客户端并发，而 HashMap 是线程不安全的，所以应当使用 ConcurrentHashMap。然后剩余操作就和老师讲的示例差不多，服务端两个 socket，一个用来监听，另一个用来通信；客户端只需要用 socket 传指令即可。考虑服务端线程的编写，对于从客户端收到的一行字符串，如何将其识别为指令呢？由于给出的指令类别很少，只有四种，所以可以考虑将字符串 split(" ") 后，根据拆分出的第一个单词来分类讨论处理。但是这样处理会有一个问题：如果指令和操作数之间隔了很多空格的话，识别出的键与值可能会变成空格，所以处理方法为：把 split(" ") 改成 split("\\s+")，用正则表达式来根据连续的空格拆分这样剩下的就是一些简单的分类讨论和用 socket 通信了。代码KeyValueServer 类：// KeyValueServer 类 package lab1; import java.io.*; import java.net.*; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @SuppressWarnings("InfiniteLoopStatement") public class KeyValueServer { static Map<String, String> map = new ConcurrentHashMap<>(); public static void main(String[] args) throws Exception { System.out.println("在端口1234监听。"); try (ServerSocket listenSocket = new ServerSocket(1234)) { while (true) { try { Socket socket = listenSocket.accept(); ServerThread serverThread = new ServerThread(socket); serverThread.start(); } catch (IOException e) { e.printStackTrace(); } } } catch (Exception e) { e.printStackTrace(); } } }ServerThread 类// ServerThread 类 package lab1; import java.io.*; import java.net.Socket; public class ServerThread extends Thread{ Socket socket = null; public ServerThread(Socket socket) { this.socket = socket; } public void run() { InputStream is = null; InputStreamReader isr = null; BufferedReader br = null; OutputStream os = null; PrintWriter pw = null; try{ is = socket.getInputStream(); isr = new InputStreamReader(is); br = new BufferedReader(isr); os = socket.getOutputStream(); pw = new PrintWriter(os, true); String line = null; while((line=br.readLine())!=null) { System.out.println("来自客户端" + this.getName() + "的指令 : " + line); String[] parts = line.split("\\s+"); String command = parts[0]; if(command.equals("PUT")) { if(parts.length != 3) pw.println("指令格式不正确。"); else { String key = parts[1]; String value = parts[2]; KeyValueServer.map.put(key, value); pw.println("键值对(" + key + "," + value + ")存储成功！"); } } else if(command.equals("GET")) { if(parts.length != 2) pw.println("指令格式不正确。"); else { String key = parts[1]; String value = KeyValueServer.map.get(key); if (value != null) { pw.println("键" + key + "所对应的值为" + value + "。"); } else { pw.println("键" + key + "目前没有所对应的值。"); } } } else if(command.equals("DELETE")) { if(parts.length != 2) pw.println("指令格式不正确。"); else { String key = parts[1]; if (KeyValueServer.map.containsKey(key)) { String value = KeyValueServer.map.get(key); KeyValueServer.map.remove(key); pw.println("键值对(" + key + "," + value + ")删除成功！"); } else { pw.println("当前不存在以" + key + "为键的键值对。"); } } } else if(command.equals("EXIT")) { if(parts.length != 1) pw.println("指令格式不正确。"); else break; } else { pw.println("不存在该指令。"); } } } catch(IOException e){ e.printStackTrace(); } finally { try { if(pw!=null) pw.close(); if(os!=null) os.close(); if(br!=null) br.close(); if(is!=null) is.close(); if(isr!=null) isr.close(); if(socket!=null) socket.close(); } catch (IOException e) { e.printStackTrace(); } } } }KeyValueClient 类// KeyValueClient 类 package lab1; import java.io.*; import java.net.*; import java.util.Scanner; public class KeyValueClient { public static void main(String[] args) throws Exception { BufferedReader stdIn = new BufferedReader(new InputStreamReader(System.in)); try(Socket socket = new Socket("127.0.0.1", 1234)) { InputStream inStream = socket.getInputStream(); OutputStream outStream = socket.getOutputStream(); BufferedReader in = new BufferedReader(new InputStreamReader(inStream)); PrintWriter out = new PrintWriter(outStream, true); Scanner sc = new Scanner(System.in); System.out.println("已连接到服务端，请输入指令："); while(true) { String command = sc.nextLine(); out.println(command); if(command.equals("EXIT")) { System.out.println("已关闭连接。"); break; } String response = in.readLine(); if(response!=null) { System.out.println(response); } } } catch (IOException e) { e.printStackTrace(); } } }运行截图指标测试结果如图，平均响应时间为 5 毫秒，TPS为7050.7。