本文主要讲述Server端如何异步（async）执行method，其基本原理如下：

1. Client端发送方法调用的请求
2. Server收到请求后会把该请求加入到内部队列里去
3. Server在另外一个线程中从队列里取出请求然后执行方法，最后把方法运行结果返回给Client

本文使用open62541 v1.1.1版本，运行环境是win10，ubuntu操作也是一样的。

一 配置open62541.h

本人使用的cmake-gui进行配置，

把UA_ENABLE_AMALGAMATION勾上，把UA_MULTITHREADING设置为100，其它使用默认配置。然后进行编译，具体过程请参考这篇文章。

二 操作过程

代码来自example目录下的tutorial_server_method_async.c，进行了简化。

首先需要在Server端定义一个方法，该方法接收一个String类型的输入参数，返回一个String类型的值。

下面是这个方法的回调，

static UA_StatusCode helloWorldMethodCallback(UA_Server *server,
	                const UA_NodeId *sessionId, void *sessionHandle,
					const UA_NodeId *methodId, void *methodContext,
					const UA_NodeId *objectId, void *objectContext,
					size_t inputSize, const UA_Variant *input,
					size_t outputSize, UA_Variant *output) 
{
    
    
	// 连接字符串，变成"Hello " + 输入参数
	UA_String *inputStr = (UA_String*)input->data;
	UA_String tmp = UA_STRING_ALLOC("Hello ");
	if (inputStr->length > 0) 
	{
    
    
		tmp.data = (UA_Byte *)UA_realloc(tmp.data, tmp.length + inputStr->length);
		memcpy(&tmp.data[tmp.length], inputStr->data, inputStr->length);
		tmp.length += inputStr->length;
	}

    // 把连接好的字符串拷贝到输出参数里
	UA_Variant_setScalarCopy(output, &tmp, &UA_TYPES[UA_TYPES_STRING]);

	// 添加字符串结束符'\0'，其值为0，所以calloc自动就会添加
	char* test = (char*)calloc(1, tmp.length + 1);
	memcpy(test, tmp.data, tmp.length);
	UA_String_clear(&tmp);

	UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "'Hello World (async)' was called");
	UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "    Data: %s", test);
	free(test);
	UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "'Hello World (async)' has ended");
	
	return UA_STATUSCODE_GOOD;
}

然后在Server端添加该方法，该方法有一个输入参数和一个输出参数。最关键的是最后一行语句，会把该方法设置为异步

static void addHellWorldMethod(UA_Server *server) 
{
    
    
	// 定义输入参数
	UA_Argument inputArgument;
	UA_Argument_init(&inputArgument);
	inputArgument.description = UA_LOCALIZEDTEXT("en-US", "A String");
	inputArgument.name = UA_STRING("MyInput");
	inputArgument.dataType = UA_TYPES[UA_TYPES_STRING].typeId;
	inputArgument.valueRank = UA_VALUERANK_SCALAR;

    // 定义输出参数
	UA_Argument outputArgument;
	UA_Argument_init(&outputArgument);
	outputArgument.description = UA_LOCALIZEDTEXT("en-US", "A String");
	outputArgument.name = UA_STRING("MyOutput");
	outputArgument.dataType = UA_TYPES[UA_TYPES_STRING].typeId;
	outputArgument.valueRank = UA_VALUERANK_SCALAR;

    // 添加方法
	UA_MethodAttributes helloAttr = UA_MethodAttributes_default;
	helloAttr.description = UA_LOCALIZEDTEXT("en-US", "Say `Hello World` async");
	helloAttr.displayName = UA_LOCALIZEDTEXT("en-US", "Hello World async");
	helloAttr.executable = true;
	helloAttr.userExecutable = true;
	UA_NodeId id = UA_NODEID_NUMERIC(1, 62541);
	UA_Server_addMethodNode(server, id,
		UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER),
		UA_NODEID_NUMERIC(0, UA_NS0ID_HASCOMPONENT),
		UA_QUALIFIEDNAME(1, "hello world"),
		helloAttr, &helloWorldMethodCallback,
		1, &inputArgument, 1, &outputArgument, NULL, NULL);
	
	
	// 把方法设置成async
	UA_Server_setMethodNodeAsync(server, id, UA_TRUE);
}

还需要一个线程来从队列中取出方法的调用请求，然后执行，主要有四个API，

• 取请求：UA_Server_getAsyncOperationNonBlocking()
• 执行请求：UA_Server_call()
• 返回执行结果：UA_Server_setAsyncOperationResult()
• 收尾：UA_CallMethodResult_clear()

注意globalServer是个全局变量，

THREAD_CALLBACK(ThreadWorker) 
{
    
    
	while (running) 
	{
    
    
		const UA_AsyncOperationRequest* request = NULL;
		void *context = NULL;
		UA_AsyncOperationType type;
		if (UA_Server_getAsyncOperationNonBlocking(globalServer, &type, &request, &context, NULL) == true) 
		{
    
    
			UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "Dequeue an async operation OK");

			UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "AsyncMethod_Testing: Got entry: OKAY");
			UA_CallMethodResult response = UA_Server_call(globalServer, &request->callMethodRequest);
			UA_Server_setAsyncOperationResult(globalServer, (UA_AsyncOperationResponse*)&response, context);
			UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "AsyncMethod_Testing: Call done: OKAY");
			UA_CallMethodResult_clear(&response);
		}
	}

	return 0;
}

main函数如下，

/* This callback will be called when a new entry is added to the Callrequest queue */
static void TestCallback(UA_Server *server) 
{
    
    
	UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "Dispatched an async method");
}

int main(void) 
{
    
    
	signal(SIGINT, stopHandler);
	signal(SIGTERM, stopHandler);

	globalServer = UA_Server_new();
	UA_ServerConfig *config = UA_Server_getConfig(globalServer);
	UA_ServerConfig_setDefault(config);

	/* 设置异步操作的通知回调 */
	config->asyncOperationNotifyCallback = TestCallback;

	/* 启动 Worker-Thread */
	THREAD_HANDLE hThread;
	THREAD_CREATE(hThread, ThreadWorker);

	/* 添加方法 */
	addHellWorldMethod(globalServer);

	UA_StatusCode retval = UA_Server_run(globalServer, &running);

	/* Shutdown the thread */
	THREAD_JOIN(hThread);

	UA_Server_delete(globalServer);
	return retval == UA_STATUSCODE_GOOD ? EXIT_SUCCESS : EXIT_FAILURE;
}

三 整体代码及运行

/* This work is licensed under a Creative Commons CCZero 1.0 Universal License.
* See http://creativecommons.org/publicdomain/zero/1.0/ for more information. */

/**
* Adding Async Methods to Objects
* -------------------------
*
* An object in an OPC UA information model may contain methods similar to
* objects in a programming language. Methods are represented by a MethodNode.
* Note that several objects may reference the same MethodNode. When an object
* type is instantiated, a reference to the method is added instead of copying
* the MethodNode. Therefore, the identifier of the context object is always
* explicitly stated when a method is called.
*
* The method callback takes as input a custom data pointer attached to the
* method node, the identifier of the object from which the method is called,
* and two arrays for the input and output arguments. The input and output
* arguments are all of type :ref:`variant`. Each variant may in turn contain a
* (multi-dimensional) array or scalar of any data type.
*
* Constraints for the method arguments are defined in terms of data type, value
* rank and array dimension (similar to variable definitions). The argument
* definitions are stored in child VariableNodes of the MethodNode with the
* respective BrowseNames ``(0, "InputArguments")`` and ``(0,
* "OutputArguments")``.
*
* Example: Hello World Method
* ^^^^^^^^^^^^^^^^^^^^^^^^^^^
* The method takes a string scalar and returns a string scalar with "Hello "
* prepended. The type and length of the input arguments is checked internally
* by the SDK, so that we don't have to verify the arguments in the callback. */

#include "open62541.h"

#include <signal.h>
#include <stdlib.h>

#ifndef WIN32
#include <pthread.h>
#define THREAD_HANDLE pthread_t
#define THREAD_CREATE(handle, callback) pthread_create(&handle, NULL, callback, NULL)
#define THREAD_JOIN(handle) pthread_join(handle, NULL)
#define THREAD_CALLBACK(name) static void * name(void *_)
#else
#include <windows.h>
#define THREAD_HANDLE HANDLE
#define THREAD_CREATE(handle, callback) { handle = CreateThread( NULL, 0, callback, NULL, 0, NULL); }
#define THREAD_JOIN(handle) WaitForSingleObject(handle, INFINITE)
#define THREAD_CALLBACK(name) static DWORD WINAPI name( LPVOID lpParam )
#endif

static UA_Server* globalServer;
static volatile UA_Boolean running = true;

static void stopHandler(int sign) 
{
    
    
	UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "received ctrl-c");
	running = false;
}

static UA_StatusCode helloWorldMethodCallback(UA_Server *server,
	                const UA_NodeId *sessionId, void *sessionHandle,
					const UA_NodeId *methodId, void *methodContext,
					const UA_NodeId *objectId, void *objectContext,
					size_t inputSize, const UA_Variant *input,
					size_t outputSize, UA_Variant *output) 
{
    
    
	// 连接字符串，变成"Hello " + 输入参数
	UA_String *inputStr = (UA_String*)input->data;
	UA_String tmp = UA_STRING_ALLOC("Hello ");
	if (inputStr->length > 0) 
	{
    
    
		tmp.data = (UA_Byte *)UA_realloc(tmp.data, tmp.length + inputStr->length);
		memcpy(&tmp.data[tmp.length], inputStr->data, inputStr->length);
		tmp.length += inputStr->length;
	}

    // 把连接好的字符串拷贝到输出参数里
	UA_Variant_setScalarCopy(output, &tmp, &UA_TYPES[UA_TYPES_STRING]);

	// 添加字符串结束符'\0'，其值为0，所以calloc自动就会添加
	char* test = (char*)calloc(1, tmp.length + 1);
	memcpy(test, tmp.data, tmp.length);
	UA_String_clear(&tmp);

	UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "'Hello World (async)' was called");
	UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "    Data: %s", test);
	free(test);
	UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "'Hello World (async)' has ended");
	
	return UA_STATUSCODE_GOOD;
}

static void addHellWorldMethod(UA_Server *server) 
{
    
    
	// 定义输入参数
	UA_Argument inputArgument;
	UA_Argument_init(&inputArgument);
	inputArgument.description = UA_LOCALIZEDTEXT("en-US", "A String");
	inputArgument.name = UA_STRING("MyInput");
	inputArgument.dataType = UA_TYPES[UA_TYPES_STRING].typeId;
	inputArgument.valueRank = UA_VALUERANK_SCALAR;

    // 定义输出参数
	UA_Argument outputArgument;
	UA_Argument_init(&outputArgument);
	outputArgument.description = UA_LOCALIZEDTEXT("en-US", "A String");
	outputArgument.name = UA_STRING("MyOutput");
	outputArgument.dataType = UA_TYPES[UA_TYPES_STRING].typeId;
	outputArgument.valueRank = UA_VALUERANK_SCALAR;

    // 添加方法
	UA_MethodAttributes helloAttr = UA_MethodAttributes_default;
	helloAttr.description = UA_LOCALIZEDTEXT("en-US", "Say `Hello World` async");
	helloAttr.displayName = UA_LOCALIZEDTEXT("en-US", "Hello World async");
	helloAttr.executable = true;
	helloAttr.userExecutable = true;
	UA_NodeId id = UA_NODEID_NUMERIC(1, 62541);
	UA_Server_addMethodNode(server, id,
		UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER),
		UA_NODEID_NUMERIC(0, UA_NS0ID_HASCOMPONENT),
		UA_QUALIFIEDNAME(1, "hello world"),
		helloAttr, &helloWorldMethodCallback,
		1, &inputArgument, 1, &outputArgument, NULL, NULL);
	
	
	// 把方法设置成async
	UA_Server_setMethodNodeAsync(server, id, UA_TRUE);
}

THREAD_CALLBACK(ThreadWorker) 
{
    
    
	while (running) 
	{
    
    
		const UA_AsyncOperationRequest* request = NULL;
		void *context = NULL;
		UA_AsyncOperationType type;
		if (UA_Server_getAsyncOperationNonBlocking(globalServer, &type, &request, &context, NULL) == true) 
		{
    
    
			UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "Dequeue an async operation OK");

			UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "AsyncMethod_Testing: Got entry: OKAY");
			UA_CallMethodResult response = UA_Server_call(globalServer, &request->callMethodRequest);
			UA_Server_setAsyncOperationResult(globalServer, (UA_AsyncOperationResponse*)&response, context);
			UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "AsyncMethod_Testing: Call done: OKAY");
			UA_CallMethodResult_clear(&response);
		}
	}

	return 0;
}

/* This callback will be called when a new entry is added to the Callrequest queue */
static void TestCallback(UA_Server *server) 
{
    
    
	UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "Dispatched an async method");
}

int main(void) 
{
    
    
	signal(SIGINT, stopHandler);
	signal(SIGTERM, stopHandler);

	globalServer = UA_Server_new();
	UA_ServerConfig *config = UA_Server_getConfig(globalServer);
	UA_ServerConfig_setDefault(config);

	/* 设置异步操作的通知回调 */
	config->asyncOperationNotifyCallback = TestCallback;

	/* 启动 Worker-Thread */
	THREAD_HANDLE hThread;
	THREAD_CREATE(hThread, ThreadWorker);

	/* 添加方法 */
	addHellWorldMethod(globalServer);

	UA_StatusCode retval = UA_Server_run(globalServer, &running);

	/* Shutdown the thread */
	THREAD_JOIN(hThread);

	UA_Server_delete(globalServer);
	return retval == UA_STATUSCODE_GOOD ? EXIT_SUCCESS : EXIT_FAILURE;
}

本人使用Visual Studio 2015进行编译连接（代码是跨平台的），ok后运行server，然后使用UaExpert进行连接，

右击该方法然后点击Call

在弹出的运行框里的输入参数里随便输入一段字符串，然后点击右下角的Call，

最后可以看到运行的结果，

同样，在server端可以看到对应的打印，

四 总结

本文主要讲述如何实现Server端方法调用的异步执行。这样Server可以在上一个方法执行没结束前可以接受其它的方法调用请求。

如果有写的不对的地方，希望能留言指正，谢谢阅读。