springBoot--OPCUA
·
最近公司有个需求需要进行对接opcua，之前从来没有对接过这方面的，但是在网上找资源时，发现这方面网上资源很少，只能硬着头皮搞，下面给大家分享一下踩的坑
连接的时候有很多种方式：
1.用户名密码2.证书，用户名
3.证书，密钥，用户名密码（前两种都是比较简单的，我是第三种情况，网上基本没资源）
首先是定义的一个config，主要是连接配置类：
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.api.config.OpcUaClientConfig;
import org.eclipse.milo.opcua.sdk.client.api.identity.UsernameProvider;
import org.eclipse.milo.opcua.stack.client.DiscoveryClient;
import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
import org.eclipse.milo.opcua.stack.core.types.structured.EndpointDescription;
import org.springblade.modules.wems.utils.CertificateLoader;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.security.PrivateKey;
import java.security.PublicKey;
import java.security.KeyPair;
import java.security.cert.X509Certificate;
import java.util.List;
import java.util.Optional;
@Slf4j
@Configuration
public class OpcUaConfig {
@Value("${opcua.client.username}")
private String username;
@Value("${opcua.client.password}")
private String password;
@Value("${opcua.client.endpoint-url}")
private String endpointUrl;
@Value("${opcua.client.application-uri}")
private String applicationUri;
@Value("${opcua.client.cert-path}")
private String certPath;
@Value("${opcua.client.key-path}")
private String keyPath;
@Bean
public OpcUaClient opcUaClient() throws Exception {
// 使用 ClassLoader 获取输入流
InputStream certInputStream = getClass().getClassLoader().getResourceAsStream(certPath);
InputStream keyInputStream = getClass().getClassLoader().getResourceAsStream(keyPath);
if (keyInputStream == null || certInputStream == null) {
throw new IllegalArgumentException("Resource not found: client_key.pem or client_cert.der");
}
// 临时将输入流写入文件路径（CertificateLoader 需要文件路径作为输入）
Path tempKeyPath = Files.createTempFile("client_key", ".pem");
Files.copy(keyInputStream, tempKeyPath, StandardCopyOption.REPLACE_EXISTING);
Path tempCertPath = Files.createTempFile("client_cert", ".der");
Files.copy(certInputStream, tempCertPath, StandardCopyOption.REPLACE_EXISTING);
// 调用 CertificateLoader 加载证书和私钥
PrivateKey privateKey = CertificateLoader.loadPrivateKey(tempKeyPath.toString());
X509Certificate clientCertificate = CertificateLoader.loadCertificate(tempCertPath.toString());
// 清理临时文件
Files.delete(tempKeyPath);
Files.delete(tempCertPath);
// PrivateKey privateKey = CertificateLoader.loadPrivateKey("src/main/resources/opcua/client_key.pem");
// X509Certificate clientCertificate = CertificateLoader.loadCertificate("src/main/resources/opcua/client_cert.der");
// 获取公钥
PublicKey serverPublicKey = clientCertificate.getPublicKey();
KeyPair clientKeyPair = new KeyPair(serverPublicKey, privateKey);
// Set up X509 identity
UsernameProvider usernameProvider = new UsernameProvider(username, password);
// Discover endpoints and create client
List<EndpointDescription> endpoints = DiscoveryClient.getEndpoints(endpointUrl).get();
Optional<EndpointDescription> endpoint = endpoints.stream()
.filter(e -> e.getSecurityPolicyUri().equals(SecurityPolicy.None.getUri())) // 尝试使用不加密的连接
.findFirst();
if (!endpoint.isPresent()) {
throw new Exception("No suitable endpoint found");
}
OpcUaClientConfig config = OpcUaClientConfig.builder()
.setEndpoint(endpoint.get())
.setCertificate(clientCertificate)
.setKeyPair(clientKeyPair)
.setIdentityProvider(usernameProvider)
.setApplicationUri(applicationUri)
.build();
OpcUaClient client = OpcUaClient.create(config);
client.connect().get(); // 确保客户端已经成功连接到服务器，后续代码才能正常进行查询、订阅和写入等操作。如果没有这行代码，程序不会自动进行连接，导致无法与 OPC UA 服务器交互。
return client;
}
}
第一个坑，这个有一个配置：applicationUri，因为我们的硬件团队没有提供这个，但是连接的时候如果不配置就会报错，这个时候可以使用证书来获取，下面这个方法可以直接拿到对应的数据
/**
* 根据der类型的证书获取证书中的ApplicationUri属性
* @param filePath
* @return 证书中的ApplicationUri属性
*/
public String getApplicationUri(String filePath){
String uri = null;
try {
// 加载证书文件
FileInputStream fis = new FileInputStream(filePath);
CertificateFactory cf = CertificateFactory.getInstance("X.509");
X509Certificate cert = (X509Certificate) cf.generateCertificate(fis);
// 获取 SAN 扩展字段
Collection<List<?>> sanEntries = cert.getSubjectAlternativeNames();
if (sanEntries != null) {
for (List<?> sanEntry : sanEntries) {
Integer type = (Integer) sanEntry.get(0);
if (type == 6) { // type 6 corresponds to URI in SAN
uri = (String) sanEntry.get(1);
log.info("ApplicationUri: " + uri);
break;
}
}
} else {
log.info("No SAN entries found in the certificate.");
}
return uri;
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
二：主要是一些工具类，包含了创建Node节点，读取简单节点，读取方法类型的节点，读取Node的上级节点，设置节点，监听变化节点等方式：
import lombok.extern.slf4j.Slf4j;
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.stack.core.AttributeId;
import org.eclipse.milo.opcua.stack.core.Identifiers;
import org.eclipse.milo.opcua.stack.core.types.builtin.*;
import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger;
import org.eclipse.milo.opcua.stack.core.types.enumerated.*;
import org.eclipse.milo.opcua.stack.core.types.structured.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.FileInputStream;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
@Slf4j
@Component
public class OpcUaUtils {
@Autowired
private OpcUaClient client;
public NodeId createNodeId(Integer nameSpaceIndex, Object identifier) {
NodeId nodeId;
if (identifier instanceof Integer) {
nodeId = new NodeId(nameSpaceIndex, (Integer) identifier);
} else if (identifier instanceof String) {
nodeId = new NodeId(nameSpaceIndex, (String) identifier);
} else if (identifier instanceof UUID) {
nodeId = new NodeId(nameSpaceIndex, identifier.toString());
} else if (identifier instanceof ByteString) {
nodeId = new NodeId(nameSpaceIndex, (ByteString) identifier);
} else {
throw new IllegalArgumentException("Unsupported identifier type: " + identifier.getClass().getName());
}
return nodeId;
}
public DataValue readNode(NodeId nodeId) {
try {
CompletableFuture<DataValue> future = client.readValue(0.0, TimestampsToReturn.Neither, nodeId);
return future.get();
} catch (InterruptedException | ExecutionException e) {
throw new RuntimeException("Failed to read node", e);
}
}
public Object readMethod(NodeId nodeId,List<String> args) throws Exception {
try {
// 动态设置输入参数
Variant[] inputArguments = args.stream()
.map(Variant::new) // 将每个参数封装为 Variant
.toArray(Variant[]::new); // 转换为 Variant 数组
NodeId objectId = readObjectNodeId(nodeId);
// 构造方法调用请求
CallMethodRequest callMethodRequest = new CallMethodRequest(
objectId, // 方法所属对象的 NodeId
nodeId, // 方法的 NodeId
inputArguments // 输入参数
);
// 调用方法
List<CallMethodRequest> requests = Arrays.asList(callMethodRequest);
CallMethodResult[] results = client.call(requests).get().getResults();
// 解析返回值
CallMethodResult result = results[0]; // 假设只有一个请求
if (result.getStatusCode().isGood()) {
log.info("Method called successfully!");
// 输出返回值
Variant[] outputArguments = result.getOutputArguments();
Object objectResult = null;
for (Variant output : outputArguments) {
log.info("Output: " + output.getValue());
objectResult = output.getValue();
}
return objectResult;
} else {
log.info("Method call failed: " + result.getStatusCode());
return null;
}
} catch (InterruptedException | ExecutionException e) {
throw new RuntimeException("Failed to read node", e);
}
}
public NodeId readObjectNodeId(NodeId methodNodeId) throws Exception {
// 浏览该节点的反向引用
BrowseDescription browseDescription = new BrowseDescription(
methodNodeId, // 要浏览的起始节点
BrowseDirection.Inverse, // 浏览方向：反向
Identifiers.HasComponent, // 关系类型：HasComponent
true, // 是否包含子类型
uint(NodeClass.Object.getValue()), // 父节点类型必须是 Object
uint(BrowseResultMask.All.getValue()) // 包含所有可用信息
);
ReferenceDescription[] references1 = client.browse(browseDescription).get().getReferences();
List<ReferenceDescription> references = Arrays.asList(references1);
// 查找第一个符合条件的父节点
if (!references.isEmpty()) {
ReferenceDescription parentRef = references.get(0);
// 将引用的 NodeId 转换为完整的 NodeId
NodeId parentNodeId = parentRef.getNodeId().toNodeId(client.getNamespaceTable()).orElse(null);
if (parentNodeId != null) {
log.info("Parent NodeId: " + parentNodeId);
return parentNodeId;
} else {
log.error("Failed to resolve parent NodeId.");
}
} else {
log.error("No parent node found for the method.");
}
return null;
}
public void writeNode(NodeId nodeId, Object value) {
try {
DataValue dataValue = new DataValue(new Variant(value));
client.writeValue(nodeId, dataValue).get();
log.info("Value written successfully to node: " + nodeId);
} catch (InterruptedException | ExecutionException e) {
throw new RuntimeException("Failed to write node", e);
}
}
/**
* 订阅某个节点
* @param nodeId
* @throws Exception
*/
public void subscribe(NodeId nodeId) throws Exception {
//创建发布间隔1000ms的订阅对象
client.getSubscriptionManager()
.createSubscription(1000.0)
.thenAccept(t -> {
//节点ns=2;s=test.device2.test2
ReadValueId readValueId = new ReadValueId(nodeId, AttributeId.Value.uid(), null, null);
//创建监控的参数
MonitoringParameters parameters = new MonitoringParameters(UInteger.valueOf(1), 1000.0, null, UInteger.valueOf(10), true);
//创建监控项请求
//该请求最后用于创建订阅。
MonitoredItemCreateRequest request = new MonitoredItemCreateRequest(readValueId, MonitoringMode.Reporting, parameters);
List<MonitoredItemCreateRequest> requests = new ArrayList<>();
requests.add(request);
//创建监控项，并且注册变量值改变时候的回调函数。
t.createMonitoredItems(
TimestampsToReturn.Both,
requests,
(item, id) -> item.setValueConsumer((it, val) -> {
// TODO 改变值的时候的回调函数 可以根据 it.getReadValueId().getNodeId()里面的getNamespaceIndex和getIdentifier两个属性进行判断从而进行逻辑处理
log.info("=====订阅nodeid====== :" + it.getReadValueId().getNodeId());
log.info("=====订阅value===== :" + val.getValue().getValue());
})
);
}).get();
//持续订阅
Thread.sleep(Long.MAX_VALUE);
}
/**
* 订阅多个节点
* @param nodeIds
* @throws Exception
*/
public void subscribe(List<NodeId> nodeIds) throws Exception {
// 创建发布间隔为 1000ms 的订阅对象
client.getSubscriptionManager()
.createSubscription(1000.0)
.thenAccept(subscription -> {
List<MonitoredItemCreateRequest> requests = new ArrayList<>();
// 遍历 NodeId 列表并创建监控项请求
for (NodeId nodeId : nodeIds) {
ReadValueId readValueId = new ReadValueId(nodeId, AttributeId.Value.uid(), null, null);
MonitoringParameters parameters = new MonitoringParameters(
UInteger.valueOf(1), 1000.0, null, UInteger.valueOf(10), true
);
MonitoredItemCreateRequest request = new MonitoredItemCreateRequest(
readValueId, MonitoringMode.Reporting, parameters
);
requests.add(request);
}
// 创建监控项，并注册变量值改变时的回调函数
subscription.createMonitoredItems(
TimestampsToReturn.Both,
requests,
(item, id) -> item.setValueConsumer((it, val) -> {
// TODO 改变值的时候的回调函数 可以根据 it.getReadValueId().getNodeId()里面的getNamespaceIndex和getIdentifier两个属性进行判断从而进行逻辑处理
log.info("===== 订阅的 NodeId =====: " + it.getReadValueId().getNodeId());
log.info("===== 订阅的值 =====: " + val.getValue().getValue());
})
);
}).get();
// 持续订阅
Thread.sleep(Long.MAX_VALUE);
}
/**
* 根据der类型的证书获取证书中的ApplicationUri属性
* @param filePath
* @return 证书中的ApplicationUri属性
*/
public String getApplicationUri(String filePath){
String uri = null;
try {
// 加载证书文件
// FileInputStream fis = new FileInputStream("src/main/resources/opcua/client_cert.der");
FileInputStream fis = new FileInputStream(filePath);
CertificateFactory cf = CertificateFactory.getInstance("X.509");
X509Certificate cert = (X509Certificate) cf.generateCertificate(fis);
// 获取 SAN 扩展字段
Collection<List<?>> sanEntries = cert.getSubjectAlternativeNames();
if (sanEntries != null) {
for (List<?> sanEntry : sanEntries) {
Integer type = (Integer) sanEntry.get(0);
if (type == 6) { // type 6 corresponds to URI in SAN
uri = (String) sanEntry.get(1);
log.info("ApplicationUri: " + uri);
break;
}
}
} else {
log.info("No SAN entries found in the certificate.");
}
return uri;
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
}
这个方法里面的基本够用了：
之前碰到的坑是，node节点有很多种类型，最简单的就类似于java中的String类型的格式，但是有一个比较复杂的是类似于一个函数的类型，但是函数类型读取的时候，还必须要有一个上级的ObjectNodeId，所以就多写了一个方法是查询反向的节点
import java.io.FileInputStream;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.security.KeyFactory;
import java.security.PrivateKey;
import java.security.cert.Certificate;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
import java.security.spec.PKCS8EncodedKeySpec;
public class CertificateLoader {
public static X509Certificate loadCertificate(String filePath) throws Exception {
CertificateFactory certificateFactory = CertificateFactory.getInstance("X.509");
try (FileInputStream fis = new FileInputStream(filePath)) {
Certificate cert = certificateFactory.generateCertificate(fis);
return (X509Certificate) cert;
}
}
public static PrivateKey loadPrivateKey(String filePath) throws Exception {
// 读取 PEM 文件的字节数据
byte[] keyBytes = Files.readAllBytes(Paths.get(filePath));
// 使用 PKCS8EncodedKeySpec 解析私钥
PKCS8EncodedKeySpec keySpec = new PKCS8EncodedKeySpec(keyBytes);
KeyFactory keyFactory = KeyFactory.getInstance("RSA");
return keyFactory.generatePrivate(keySpec);
}
}
至于OPCUA是干嘛用的，这个就需要各位小伙伴自己去查了，这次分享的主要是JAVA怎么去处理OPCUA的数据，如果本地测试可以下载一个Prosys OPCUA这个软件，在本地测试
魔乐社区（Modelers.cn) 是一个中立、公益的人工智能社区，提供人工智能工具、模型、数据的托管、展示与应用协同服务，为人工智能开发及爱好者搭建开放的学习交流平台。社区通过理事会方式运作，由全产业链共同建设、共同运营、共同享有，推动国产AI生态繁荣发展。
更多推荐
3
0- 0
已为社区贡献2条内容
所有评论(0)