简电云 | Java + WebSocket 实现 OCPP 2.0.1 充电桩模拟程序

一、为什么需要充电桩模拟器

充电平台通常需要同时处理设备接入、订单、计费、告警和远程控制。如果所有测试都依赖真实充电桩，会遇到几个现实问题：

• 设备数量有限，难以验证大量设备同时在线；

• 插枪、刷卡、充电和拔枪需要人工操作，回归测试效率低；

• 断网、重复报文、事件乱序和异常停止不容易稳定复现；

• 不同厂商固件行为存在差异，平台问题与设备问题容易混在一起；

• 开发人员无法随时获得完整的充电现场环境。

模拟器并不是为了替代真实设备，而是把协议行为变成可重复、可编排的测试输入。开发阶段先用模拟器验证平台状态机，再使用真实充电桩完成硬件、计量和电气安全测试，能够明显降低联调成本。

本文实现以下最小闭环：

Java 充电桩模拟器 Java CSMS | | |--- WebSocket / ocpp2.0.1 ----->| |--- BootNotification ---------->| |<-- Accepted + interval ---------| |--- StatusNotification -------->| |--- Authorize ----------------->| |<-- Accepted --------------------| |--- TransactionEvent(Started) -->| |--- MeterValues --------------->| |--- TransactionEvent(Updated) -->| |--- TransactionEvent(Ended) ---->| |--- Heartbeat ----------------->|

二、OCPP 2.0.1 的 WebSocket 报文

OCPP 2.0.1 JSON 通过 WebSocket 传输，握手时使用子协议ocpp2.0.1。协议常用三种消息类型：

例如充电桩上线时发送：

[2,"msg-001","BootNotification",{ "reason":"PowerUp", "chargingStation":{ "model":"SIM-AC-22K", "vendorName":"DemoVendor" } }]

CSMS 返回：

[3,"msg-001",{ "currentTime":"2026-09-30T02:00:00Z", "interval":30, "status":"Accepted" }]

messageId用于关联请求和响应。一次连接中的 CALL 应使用唯一标识，客户端还需要维护“消息编号—等待结果”的映射。

本文为了展示 WebSocket 和 OCPP 报文的底层处理过程，不引入商业 OCPP SDK，而是使用 Jackson 构造消息。这样代码容易理解，但生产项目仍应补充完整的 OCPP 2.0.1 JSON Schema 校验和功能块实现。

三、创建 Maven 工程

示例使用 Java 17。工程结构如下：

ocpp201-java-demo/ ├── pom.xml └── src/main/java/com/example/ocpp/ ├── CsmsServer.java ├── CsmsEndpoint.java └── ChargePointSimulator.java

pom.xml使用 Tyrus 作为 Jakarta WebSocket 实现，Jackson 负责 JSON 解析：

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> ​ <groupId>com.example</groupId> <artifactId>ocpp201-java-demo</artifactId> <version>1.0.0</version> ​ <properties> <maven.compiler.release>17</maven.compiler.release> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> </properties> ​ <dependencies> <dependency> <groupId>jakarta.websocket</groupId> <artifactId>jakarta.websocket-api</artifactId> <version>2.1.1</version> </dependency> <dependency> <groupId>org.glassfish.tyrus</groupId> <artifactId>tyrus-server</artifactId> <version>2.1.5</version> </dependency> <dependency> <groupId>org.glassfish.tyrus</groupId> <artifactId>tyrus-container-grizzly-server</artifactId> <version>2.1.5</version> </dependency> <dependency> <groupId>org.glassfish.tyrus.bundles</groupId> <artifactId>tyrus-standalone-client</artifactId> <version>2.1.5</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.17.2</version> </dependency> </dependencies> ​ <build> <plugins> <plugin> <groupId>org.codehaus.mojo</groupId> <artifactId>exec-maven-plugin</artifactId> <version>3.3.0</version> </plugin> </plugins> </build> </project>

固定依赖版本有利于复现。Tyrus 不仅提供 WebSocket API，还提供可直接启动的服务端容器和客户端实现。

四、实现 CSMS WebSocket 服务端

1. 创建 WebSocket 端点

创建CsmsEndpoint.java：

package com.example.ocpp; ​ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ObjectNode; import jakarta.websocket.CloseReason; import jakarta.websocket.OnClose; import jakarta.websocket.OnError; import jakarta.websocket.OnMessage; import jakarta.websocket.OnOpen; import jakarta.websocket.Session; import jakarta.websocket.server.PathParam; import jakarta.websocket.server.ServerEndpoint; ​ import java.io.IOException; import java.time.Instant; ​ @ServerEndpoint( value = "/ocpp/{chargePointId}", subprotocols = {"ocpp2.0.1"} ) public class CsmsEndpoint { private static final ObjectMapper JSON = new ObjectMapper(); ​ @OnOpen public void onOpen( Session session, @PathParam("chargePointId") String chargePointId) throws IOException { if (!"ocpp2.0.1".equals(session.getNegotiatedSubprotocol())) { session.close(new CloseReason( CloseReason.CloseCodes.PROTOCOL_ERROR, "OCPP 2.0.1 subprotocol required")); return; } session.getUserProperties().put("chargePointId", chargePointId); System.out.printf("[Connected] id=%s, session=%s%n", chargePointId, session.getId()); } ​ @OnMessage public void onMessage(String text, Session session) throws IOException { JsonNode frame = JSON.readTree(text); if (!frame.isArray() || frame.size() < 3) { session.getBasicRemote().sendText(callError( "unknown", "FormationViolation", "Invalid frame")); return; } ​ int messageType = frame.get(0).asInt(); if (messageType != 2 || frame.size() != 4) { // 本示例服务端只处理充电桩发起的 CALL。 return; } ​ String messageId = frame.get(1).asText(); String action = frame.get(2).asText(); JsonNode payload = frame.get(3); ​ try { ObjectNode response = handleCall(session, action, payload); session.getBasicRemote().sendText(callResult(messageId, response)); } catch (IllegalArgumentException e) { session.getBasicRemote().sendText(callError( messageId, "NotSupported", e.getMessage())); } catch (Exception e) { session.getBasicRemote().sendText(callError( messageId, "InternalError", "Failed to process request")); } } ​ private ObjectNode handleCall( Session session, String action, JsonNode payload) { String cpId = (String) session.getUserProperties().get("chargePointId"); ObjectNode response = JSON.createObjectNode(); ​ switch (action) { case "BootNotification" -> { System.out.printf("[Boot] id=%s, payload=%s%n", cpId, payload); response.put("currentTime", Instant.now().toString()); response.put("interval", 30); response.put("status", "Accepted"); } case "Heartbeat" -> { System.out.printf("[Heartbeat] id=%s%n", cpId); response.put("currentTime", Instant.now().toString()); } case "StatusNotification" -> System.out.printf( "[Status] id=%s, evse=%s, connector=%s, status=%s%n", cpId, payload.path("evseId").asText(), payload.path("connectorId").asText(), payload.path("connectorStatus").asText()); case "Authorize" -> { String token = payload.path("idToken").path("idToken").asText(); String status = "TEST-001".equals(token) ? "Accepted" : "Invalid"; response.putObject("idTokenInfo").put("status", status); System.out.printf("[Authorize] token=%s, status=%s%n", token, status); } case "TransactionEvent" -> System.out.printf( "[Transaction] id=%s, event=%s, seq=%s, trigger=%s%n", payload.path("transactionInfo") .path("transactionId").asText(), payload.path("eventType").asText(), payload.path("seqNo").asText(), payload.path("triggerReason").asText()); case "MeterValues" -> System.out.printf( "[Meter] id=%s, evse=%s, values=%s%n", cpId, payload.path("evseId").asText(), payload.path("meterValue")); default -> throw new IllegalArgumentException( "Unsupported action: " + action); } return response; } ​ private String callResult(String messageId, ObjectNode payload) { ArrayNode frame = JSON.createArrayNode(); frame.add(3).add(messageId).add(payload); return frame.toString(); } ​ private String callError( String messageId, String code, String description) { ArrayNode frame = JSON.createArrayNode(); frame.add(4) .add(messageId) .add(code) .add(description) .add(JSON.createObjectNode()); return frame.toString(); } ​ @OnClose public void onClose(Session session, CloseReason reason) { System.out.printf("[Disconnected] session=%s, reason=%s%n", session.getId(), reason); } ​ @OnError public void onError(Session session, Throwable error) { System.err.printf("[WebSocket error] session=%s, error=%s%n", session == null ? "unknown" : session.getId(), error.getMessage()); } }

该端点完成四件事：校验协商后的 WebSocket 子协议、从路径取得充电桩编号、按 CALL 格式解析请求、使用相同messageId返回 CALLRESULT。

2. 启动服务端容器

创建CsmsServer.java：

package com.example.ocpp; ​ import org.glassfish.tyrus.server.Server; ​ import java.io.BufferedReader; import java.io.InputStreamReader; ​ public class CsmsServer { public static void main(String[] args) throws Exception { Server server = new Server( "0.0.0.0", 9000, "/", null, CsmsEndpoint.class); ​ try { server.start(); System.out.println( "CSMS started: ws://127.0.0.1:9000/ocpp/{chargePointId}"); System.out.println("Press Enter to stop."); new BufferedReader(new InputStreamReader(System.in)).readLine(); } finally { server.stop(); } } }

为了突出协议链路，示例只打印收到的数据。正式 CSMS 还需要设备认证、Schema 校验、事件持久化、幂等处理、订单状态机和审计日志。

五、实现 Java 充电桩模拟器

创建ChargePointSimulator.java：

package com.example.ocpp; ​ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ObjectNode; import jakarta.websocket.ClientEndpoint; import jakarta.websocket.CloseReason; import jakarta.websocket.ContainerProvider; import jakarta.websocket.OnClose; import jakarta.websocket.OnError; import jakarta.websocket.OnMessage; import jakarta.websocket.OnOpen; import jakarta.websocket.Session; import jakarta.websocket.WebSocketContainer; ​ import java.net.URI; import java.time.Instant; import java.util.Map; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; ​ @ClientEndpoint(subprotocols = {"ocpp2.0.1"}) public class ChargePointSimulator { private static final ObjectMapper JSON = new ObjectMapper(); private static final long RESPONSE_TIMEOUT_SECONDS = 10; ​ private final Map<String, CompletableFuture<JsonNode>> pending = new ConcurrentHashMap<>(); private Session session; private int heartbeatInterval = 30; ​ @OnOpen public void onOpen(Session session) { this.session = session; System.out.printf("Connected, subprotocol=%s%n", session.getNegotiatedSubprotocol()); } ​ @OnMessage public void onMessage(String text) { try { JsonNode frame = JSON.readTree(text); int messageType = frame.get(0).asInt(); String messageId = frame.get(1).asText(); ​ if (messageType == 3) { complete(messageId, frame.get(2)); } else if (messageType == 4) { String error = frame.get(2).asText() + ": " + frame.get(3).asText(); fail(messageId, new IllegalStateException(error)); } } catch (Exception e) { System.err.println("Invalid response: " + e.getMessage()); } } ​ @OnClose public void onClose(Session session, CloseReason reason) { IllegalStateException error = new IllegalStateException( "WebSocket closed: " + reason); pending.values().forEach(future -> future.completeExceptionally(error)); pending.clear(); } ​ @OnError public void onError(Session session, Throwable error) { System.err.println("WebSocket error: " + error.getMessage()); } ​ public JsonNode call(String action, ObjectNode payload) throws Exception { String messageId = UUID.randomUUID().toString(); CompletableFuture<JsonNode> future = new CompletableFuture<>(); pending.put(messageId, future); ​ ArrayNode frame = JSON.createArrayNode(); frame.add(2).add(messageId).add(action).add(payload); ​ try { session.getAsyncRemote().sendText(frame.toString(), result -> { if (!result.isOK()) { fail(messageId, result.getException()); } }); return future.get(RESPONSE_TIMEOUT_SECONDS, TimeUnit.SECONDS); } finally { pending.remove(messageId); } } ​ private void complete(String messageId, JsonNode payload) { CompletableFuture<JsonNode> future = pending.get(messageId); if (future != null) { future.complete(payload); } } ​ private void fail(String messageId, Throwable error) { CompletableFuture<JsonNode> future = pending.get(messageId); if (future != null) { future.completeExceptionally(error); } } ​ private void boot() throws Exception { ObjectNode station = JSON.createObjectNode(); station.put("model", "SIM-AC-22K"); station.put("vendorName", "DemoVendor"); station.put("serialNumber", "SIM-CP-001"); station.put("firmwareVersion", "1.0.0"); ​ ObjectNode payload = JSON.createObjectNode(); payload.put("reason", "PowerUp"); payload.set("chargingStation", station); ​ JsonNode response = call("BootNotification", payload); if (!"Accepted".equals(response.path("status").asText())) { throw new IllegalStateException( "Registration rejected: " + response); } heartbeatInterval = response.path("interval").asInt(30); System.out.printf("Boot accepted, heartbeat=%ds%n", heartbeatInterval); } ​ private void heartbeat() throws Exception { JsonNode response = call("Heartbeat", JSON.createObjectNode()); System.out.println("Heartbeat response: " + response); } ​ private void reportStatus(String status) throws Exception { ObjectNode payload = JSON.createObjectNode(); payload.put("timestamp", Instant.now().toString()); payload.put("connectorStatus", status); payload.put("evseId", 1); payload.put("connectorId", 1); call("StatusNotification", payload); } ​ private boolean authorize(String token) throws Exception { ObjectNode idToken = JSON.createObjectNode(); idToken.put("idToken", token); idToken.put("type", "ISO14443"); ​ ObjectNode payload = JSON.createObjectNode(); payload.set("idToken", idToken); JsonNode response = call("Authorize", payload); return "Accepted".equals( response.path("idTokenInfo").path("status").asText()); } ​ private void transactionEvent( String eventType, String triggerReason, int seqNo, String transactionId, String token) throws Exception { ObjectNode payload = JSON.createObjectNode(); payload.put("eventType", eventType); payload.put("timestamp", Instant.now().toString()); payload.put("triggerReason", triggerReason); payload.put("seqNo", seqNo); ​ ObjectNode transactionInfo = payload.putObject("transactionInfo"); transactionInfo.put("transactionId", transactionId); transactionInfo.put("chargingState", "Ended".equals(eventType) ? "Idle" : "Charging"); ​ ObjectNode evse = payload.putObject("evse"); evse.put("id", 1); evse.put("connectorId", 1); ​ if (token != null) { ObjectNode idToken = payload.putObject("idToken"); idToken.put("idToken", token); idToken.put("type", "ISO14443"); } call("TransactionEvent", payload); } ​ private void meterValues(double energyKwh) throws Exception { ObjectNode payload = JSON.createObjectNode(); payload.put("evseId", 1); ​ ObjectNode meterValue = JSON.createObjectNode(); meterValue.put("timestamp", Instant.now().toString()); ​ ObjectNode sampledValue = JSON.createObjectNode(); sampledValue.put("value", energyKwh); sampledValue.put("context", "Sample.Periodic"); sampledValue.put("measurand", "Energy.Active.Import.Register"); sampledValue.putObject("unitOfMeasure").put("unit", "kWh"); ​ meterValue.putArray("sampledValue").add(sampledValue); payload.putArray("meterValue").add(meterValue); call("MeterValues", payload); } ​ private void runChargeSession() throws Exception { String token = "TEST-001"; reportStatus("Available"); Thread.sleep(1_000); ​ if (!authorize(token)) { System.out.println("Token rejected: " + token); return; } ​ String transactionId = UUID.randomUUID().toString(); reportStatus("Occupied"); transactionEvent( "Started", "Authorized", 0, transactionId, token); ​ double[] energyValues = {0.2, 0.5, 0.9}; for (int i = 0; i < energyValues.length; i++) { Thread.sleep(2_000); meterValues(energyValues[i]); transactionEvent( "Updated", "MeterValuePeriodic", i + 1, transactionId, null); } ​ transactionEvent( "Ended", "StopAuthorized", 4, transactionId, token); reportStatus("Available"); System.out.println("Transaction finished: " + transactionId); } ​ public static void main(String[] args) throws Exception { String url = "ws://127.0.0.1:9000/ocpp/SIM-CP-001"; ChargePointSimulator simulator = new ChargePointSimulator(); WebSocketContainer container = ContainerProvider .getWebSocketContainer(); ​ Session session = container.connectToServer( simulator, URI.create(url)); try { simulator.boot(); simulator.runChargeSession(); ​ // 演示一次心跳。常驻模拟器应按 BootNotification 返回的 // interval 使用 ScheduledExecutorService 周期发送。 Thread.sleep(Math.min(simulator.heartbeatInterval, 3) * 1_000L); simulator.heartbeat(); } finally { session.close(); } } }

模拟器中的pending是关键结构。每发送一个 CALL，就以messageId保存一个CompletableFuture；收到 CALLRESULT 或 CALLERROR 后，再完成对应的 Future。这样即使多个请求同时在途，也不会把响应关联到错误的业务请求。

六、编译与运行

先编译工程：

mvn clean compile

打开第一个终端启动 CSMS：

mvn exec:java -Dexec.mainClass=com.example.ocpp.CsmsServer

打开第二个终端启动充电桩模拟器：

mvn exec:java -Dexec.mainClass=com.example.ocpp.ChargePointSimulator

CSMS 会依次输出类似日志：

[Connected] id=SIM-CP-001, session=... [Boot] id=SIM-CP-001, payload=... [Status] id=SIM-CP-001, evse=1, connector=1, status=Available [Authorize] token=TEST-001, status=Accepted [Status] id=SIM-CP-001, evse=1, connector=1, status=Occupied [Transaction] id=..., event=Started, seq=0, trigger=Authorized [Meter] id=SIM-CP-001, evse=1, values=... [Transaction] id=..., event=Updated, seq=1, trigger=MeterValuePeriodic [Transaction] id=..., event=Ended, seq=4, trigger=StopAuthorized [Status] id=SIM-CP-001, evse=1, connector=1, status=Available [Heartbeat] id=SIM-CP-001

OCPP 2.0.1 不再使用 OCPP 1.6 的StartTransaction和StopTransaction，而是统一使用TransactionEvent。充电站生成字符串类型的transactionId，eventType表示开始、更新或结束，seqNo表示事件在本次交易中的顺序。

七、几个容易踩坑的实现细节

1. WebSocket 路径不等于可信身份

示例从/ocpp/SIM-CP-001中取得充电桩编号，但客户端声明的编号不能直接作为可信身份。生产环境至少应使用wss://，并通过 HTTP Basic Authentication 或客户端证书校验设备身份，同时检查凭据与充电站编号的绑定关系。

2. 必须协商 OCPP 子协议

客户端和服务端都声明ocpp2.0.1，握手后还要检查Session.getNegotiatedSubprotocol()。如果没有协商出双方支持的版本，应关闭连接，不能继续按普通 WebSocket 文本处理。

3. BootNotification 决定设备后续行为

CSMS 可以返回：

• Accepted：注册成功，可以发送业务消息；

• Pending：平台暂未完成注册，应按interval再次请求；

• Rejected：注册被拒绝，不应继续发送业务消息。

演示代码只继续处理Accepted。完整模拟器还应实现 Pending 重试和 Rejected 退避策略。

4. WebSocket Ping 不能代替 Heartbeat

Ping/Pong 用于判断网络连接是否存活，OCPPHeartbeat用于协议层时钟同步和在线状态维护。两者用途不同。常驻模拟器应根据 BootNotification 返回的interval，使用ScheduledExecutorService周期发送 Heartbeat。

5. 交易事件必须保证顺序与幂等

seqNo应在同一交易内单调递增。设备断网恢复后可能补发事件，平台必须结合充电桩编号、transactionId、seqNo和事件内容去重，不能因为 Updated 重发而重复计费。

本文分别发送MeterValues与TransactionEvent(Updated)。实际项目也可以在TransactionEvent的meterValue中携带交易计量值。平台应保存原始表底，费用计算不应依赖浮点差值反复累加。

6. 状态与交易不能互相替代

StatusNotification(Occupied)只表示 EVSE 或连接器被占用，不一定已经产生有效交易；TransactionEvent(Started)表示交易建立，也不代表此刻一定有功率输出。平台需要分别维护设备、连接器和交易状态。

八、扩展为自动化测试工具

1. 支持远程启动和停止

当前 CSMS 只响应充电桩上行请求。下一步可以让 CSMS 主动发送RequestStartTransaction和RequestStopTransaction，模拟器则处理服务端 CALL 并返回 CALLRESULT。

远程启动时要保存remoteStartId，并在后续TransactionEvent中回传，使平台可以关联远程命令和实际交易。由于双方都可能发起 CALL，CSMS 也需要实现与模拟器相同的messageId和 Future 映射。

2. 使用场景文件驱动

可以把设备编号、EVSE 数量、Token、功率曲线和停止原因放入 YAML：

chargePointId: SIM-CP-001 evseId: 1 connectorId: 1 token: TEST-001 meterValuesKwh: [0.2, 0.5, 0.9] stopReason: Local

测试人员只需修改场景文件，不必改动 Java 代码。

3. 增加异常场景

4. 批量并发设备

批量测试时，每台模拟设备应拥有独立的 WebSocket Session、设备编号、交易编号和状态。可以使用线程池或虚拟线程并发创建客户端，但不要一开始就追求极高连接数。建议从 10、100、500 台逐级增加，同时观察 CSMS 的 CPU、内存、数据库连接、消息积压和响应时间。

5. 增加 Schema 校验和测试断言

当前代码只检查 WebSocket 数组结构，没有加载 OCPP 2.0.1 JSON Schema。平台测试工具至少应增加：

原始 WebSocket 帧 -> OCPP JSON Schema 校验 -> 消息解析与状态更新 -> 场景断言和测试报告

原始帧用于还原事实，Schema 用于发现字段问题，场景断言用于判断平台行为。日志中的 Token、证书信息和用户标识需要脱敏，并设置合理的保留期限。

九、从示例走向生产联调

本文代码适合本地开发和协议学习，不应直接作为生产充电桩固件或完整 CSMS。正式联调前至少需要补齐：

• 使用 WSS、设备认证和证书生命周期管理；

• 加载 OCPP 2.0.1 Schema，严格校验请求和响应；

• 建立设备、EVSE、连接器和交易的持久化状态模型；

• 正确处理 CALLERROR、超时、重试和 WebSocket 重连；

• 支持离线消息缓存、事件补传和重复消息幂等；

• 处理时间同步、计量单位、精度和异常表底；

• 增加远程启停、复位、解锁、固件升级和智能充电；

• 使用 OCA 一致性测试工具和真实设备完成最终验证。

代码能够收发 OCPP 报文，不代表产品已经通过 OCPP 认证。正式项目仍应以 Open Charge Alliance 发布的 OCPP 2.0.1 规范、勘误、实施指南和认证要求为准。

十、总结

本文使用 Java 17、Jakarta WebSocket、Tyrus 和 Jackson 实现了一个 OCPP 2.0.1 最小通信闭环。CSMS 接收 WebSocket 连接并处理 CALL，充电桩模拟器完成上线、状态上报、鉴权、交易开始、计量上报、交易结束和心跳。

一个真正有价值的模拟器，核心不是生成几条 JSON，而是复现连接、设备状态、授权、交易和计量之间的时序关系。在此基础上加入远程命令、异常注入、批量并发、Schema 校验和自动断言，就可以形成供研发、测试和交付团队共同使用的 OCPP 2.0.1 测试工具。

说明：本文代码用于技术学习和联调演示，不替代 OCPP 2.0.1 正式规范，也不构成生产部署的完整安全方案。